diff --git a/.env.example b/.env.example index eb12ecdd..7af1dac1 100644 --- a/.env.example +++ b/.env.example @@ -127,11 +127,17 @@ WEBHOOK_MAX_COUNT=500 # EMBEDDING_MODEL= (required to enable embeddings; no default) # EMBEDDING_NORMALIZE=false (optional; L2-normalize vectors client-side; cosine similarity is scale-invariant, so usually unneeded) # EMBEDDING_MAX_CONCURRENT=5 (worker concurrency; default 5) -# EMBEDDING_MAX_ATTEMPTS=3 (River job retries before failing; default 3) +# EMBEDDING_MAX_ATTEMPTS=5 (River job attempts before a transient failure is reconciled; default 5) +# EMBEDDING_JOB_TIMEOUT_SECONDS=60 (deadline for one embedding job; independent of other enrichments) # EMBEDDING_BATCH_SIZE=1 (document micro-batch size; 1 disables batching; provider must support batches) # EMBEDDING_BATCH_MAX_WAIT_MS=25 (maximum time to collect a partial batch; default 25ms) # EMBEDDING_BATCH_MAX_IN_FLIGHT=1 (maximum concurrent provider batch requests; default 1) # EMBEDDING_HTTP_DISABLE_KEEP_ALIVES=false (open a new OpenAI-compatible provider connection per request; useful for Kubernetes service distribution) +# EMBEDDING_RECONCILE_ENABLED=false (periodically repair missing taxonomy embeddings; opt in per deployment) +# EMBEDDING_RECONCILE_INTERVAL_SECONDS=300 (leader-elected sweep interval) +# EMBEDDING_RECONCILE_RETRY_AFTER_SECONDS=900 (cooldown before retrying the same transient failure) +# EMBEDDING_RECONCILE_TARGET_DEPTH=100 (maximum runnable jobs kept in the low-priority repair lane) +# EMBEDDING_RECONCILE_MAX_CONCURRENT=1 (worker concurrency reserved for repaired records) # Translation (language enrichment) is optional. To enable, set both TRANSLATION_PROVIDER and TRANSLATION_MODEL; if either is unset, translation is disabled and no translation jobs run. # Open-text feedback (value_text) is translated into each tenant's configured target_language (Hub tenant settings), falling back to TRANSLATION_DEFAULT_LANGUAGE when a tenant has none. Same providers/auth model as embeddings. diff --git a/charts/hub/values.yaml b/charts/hub/values.yaml index 8817d97e..3ce4c2b0 100644 --- a/charts/hub/values.yaml +++ b/charts/hub/values.yaml @@ -322,6 +322,17 @@ config: # Optional: use with EMBEDDING_PROVIDER=openai to target a self-hosted OpenAI-compatible embeddings endpoint. # Example: http://text-embeddings-inference.default.svc.cluster.local/v1 EMBEDDING_BASE_URL: "" + # Embedding jobs have their own provider deadline and bounded retry budget. These settings are + # provider-neutral and apply equally to managed and self-hosted embedding APIs. + EMBEDDING_MAX_ATTEMPTS: "5" + EMBEDDING_JOB_TIMEOUT_SECONDS: "60" + # Opt in after TAXONOMY_SERVICE_URL/TOKEN and the taxonomy embedding model are configured. + # Repair work uses a separate low-priority, single-worker lane so it cannot delay live records. + EMBEDDING_RECONCILE_ENABLED: "false" + EMBEDDING_RECONCILE_INTERVAL_SECONDS: "300" + EMBEDDING_RECONCILE_RETRY_AFTER_SECONDS: "900" + EMBEDDING_RECONCILE_TARGET_DEPTH: "100" + EMBEDDING_RECONCILE_MAX_CONCURRENT: "1" # Optional: internal URL Hub uses to start taxonomy generation jobs. # Example: http://taxonomy.default.svc.cluster.local:8000 TAXONOMY_SERVICE_URL: "" diff --git a/cmd/api/app.go b/cmd/api/app.go index 1f30cd35..32c5b55f 100644 --- a/cmd/api/app.go +++ b/cmd/api/app.go @@ -827,7 +827,7 @@ func runEnrichmentBacklogPoller( // gauge for as long as the scan stays slow. A separate timeout off the same parent keeps // the bound and drops the coupling. failedCtx, cancelFailed := context.WithTimeout(ctx, enrichmentBacklogQueryTimeout) - refreshFailedRecords(failedCtx, statusRepo, failures) + refreshFailedRecords(failedCtx, statusRepo, failures, cfg.taxonomyEmbeddingModel) cancelFailed() } @@ -1110,12 +1110,13 @@ func refreshFailedRecords( ctx context.Context, statusRepo *repository.EnrichmentStatusRepository, failures observability.EnrichmentFailureMetrics, + taxonomyEmbeddingModel string, ) { if failures == nil { return } - counts, err := statusRepo.CountFailedRecordsAggregate(ctx) + counts, err := statusRepo.CountFailedRecordsAggregate(ctx, taxonomyEmbeddingModel) if err != nil { failures.ClearFailedRecords() @@ -1131,11 +1132,16 @@ func refreshFailedRecords( // Zero the known buckets before applying, so an enrichment whose last failure was resolved // reports 0 rather than keeping its final non-zero reading forever. The query returns no row // for an empty bucket, which would otherwise be indistinguishable from "not refreshed". - for _, enrichment := range []string{ + enrichments := []string{ observability.EnrichmentTypeSentiment, observability.EnrichmentTypeEmotions, observability.EnrichmentTypeTranslation, - } { + } + if taxonomyEmbeddingModel != "" { + enrichments = append(enrichments, observability.EnrichmentTypeTaxonomyEmbedding) + } + + for _, enrichment := range enrichments { failures.SetFailedRecords(enrichment, true, 0) failures.SetFailedRecords(enrichment, false, 0) } diff --git a/cmd/backfill-embeddings/main.go b/cmd/backfill-embeddings/main.go index 2394683d..2be4d046 100644 --- a/cmd/backfill-embeddings/main.go +++ b/cmd/backfill-embeddings/main.go @@ -219,7 +219,15 @@ func run() int { } docPrefix := service.EmbeddingPrefixForProvider(providerCanonical) - embeddingWorker := workers.NewFeedbackEmbeddingWorker(feedbackRecordsService, embeddingClient, docPrefix, nil) + embeddingWorker := workers.NewFeedbackEmbeddingWorkerWithOptions( + feedbackRecordsService, + embeddingClient, + docPrefix, + nil, + cfg.Embedding.JobTimeout.Duration(), + nil, + nil, + ) riverWorkers := river.NewWorkers() river.AddWorker(riverWorkers, embeddingWorker) diff --git a/cmd/worker/app.go b/cmd/worker/app.go index cce70a36..b42c07a2 100644 --- a/cmd/worker/app.go +++ b/cmd/worker/app.go @@ -10,6 +10,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" "github.com/riverqueue/river" "github.com/riverqueue/river/riverdriver/riverpgxv5" + "github.com/riverqueue/river/rivertype" sdkmetric "go.opentelemetry.io/otel/sdk/metric" sdktrace "go.opentelemetry.io/otel/sdk/trace" @@ -134,6 +135,7 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) { var ( translationRecordsService *service.FeedbackRecordsService embeddingBatch *service.BatchingEmbeddingClient + embeddingReconcileService *service.EmbeddingReconcileService ) if providerName != "" { @@ -206,6 +208,24 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) { deps.EmbeddingClient = embeddingClient deps.EmbeddingDocPrefix = docPrefix deps.EmbeddingMetrics = embeddingMetrics + + if embeddingReconcileConfigured(cfg) { + embeddingReconcileService = service.NewEmbeddingReconcileService( + embeddingsRepo, + taxonomyEmbeddingModel, + cfg.Embedding.ReconcileTargetDepth, + cfg.Embedding.MaxAttempts, + cfg.Embedding.ReconcileRetryAfter.Duration(), + ) + deps.EmbeddingReconcileSweeper = embeddingReconcileService + + slog.Info("taxonomy embedding reconciliation configured", + "interval", cfg.Embedding.ReconcileInterval.Duration(), + "retry_after", cfg.Embedding.ReconcileRetryAfter.Duration(), + "target_depth", cfg.Embedding.ReconcileTargetDepth, + "max_concurrent", cfg.Embedding.ReconcileMaxConcurrent, + ) + } } if cfg.Translation.Provider != "" && cfg.Translation.Model != "" { @@ -323,6 +343,30 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) { Queues: queues, Workers: riverWorkers, } + + if embeddingReconcileService != nil { + riverCfg.PeriodicJobs = append(riverCfg.PeriodicJobs, river.NewPeriodicJob( + river.PeriodicInterval(cfg.Embedding.ReconcileInterval.Duration()), + func() (river.JobArgs, *river.InsertOpts) { + return service.EmbeddingReconcileArgs{}, &river.InsertOpts{ + Queue: service.EmbeddingReconcileQueueName, + MaxAttempts: 1, + UniqueOpts: river.UniqueOpts{ + ByArgs: true, + ByState: []rivertype.JobState{ + rivertype.JobStateAvailable, + rivertype.JobStatePending, + rivertype.JobStateRetryable, + rivertype.JobStateRunning, + rivertype.JobStateScheduled, + }, + }, + } + }, + &river.PeriodicJobOpts{RunOnStart: true}, + )) + } + if cfg.River.JobTimeoutSec.Duration() > 0 { riverCfg.JobTimeout = cfg.River.JobTimeoutSec.Duration() } @@ -352,6 +396,10 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) { translationRecordsService.SetEmbeddingInserter(riverClient) } + if embeddingReconcileService != nil { + embeddingReconcileService.SetInserter(riverClient) + } + return &WorkerApp{ cfg: cfg, db: db, @@ -362,6 +410,16 @@ func NewWorkerApp(cfg *config.Config, db *pgxpool.Pool) (*WorkerApp, error) { }, nil } +// embeddingReconcileConfigured keeps automatic repair aligned with the taxonomy embedding +// backlog contract: embeddings and taxonomy must be configured before the worker spends provider +// calls creating taxonomy-only vectors. Translation is intentionally not a gate because taxonomy +// input falls back to the source text when a deployment does not translate records. +func embeddingReconcileConfigured(cfg *config.Config) bool { + return cfg.Embedding.ReconcileEnabled && + cfg.Embedding.Provider != "" && cfg.Embedding.Model != "" && + cfg.Taxonomy.ServiceURL != "" +} + // embeddingProviderAndModel returns (canonical provider, model) when embeddings are enabled // (provider and model set and supported). Otherwise ("", ""). func embeddingProviderAndModel(cfg *config.Config) (provider, model string) { diff --git a/cmd/worker/app_test.go b/cmd/worker/app_test.go new file mode 100644 index 00000000..548d6766 --- /dev/null +++ b/cmd/worker/app_test.go @@ -0,0 +1,51 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/formbricks/hub/internal/config" +) + +func TestEmbeddingReconcileConfigured(t *testing.T) { + base := func() *config.Config { + return &config.Config{ + Embedding: config.EmbeddingConfig{ + ReconcileEnabled: true, + Provider: "openai", + Model: "embedding-model", + }, + Taxonomy: config.TaxonomyConfig{ServiceURL: "https://taxonomy.example.com"}, + } + } + + t.Run("translation is not required because taxonomy input falls back to source text", func(t *testing.T) { + assert.True(t, embeddingReconcileConfigured(base())) + }) + + t.Run("explicit opt in is required", func(t *testing.T) { + cfg := base() + cfg.Embedding.ReconcileEnabled = false + assert.False(t, embeddingReconcileConfigured(cfg)) + }) + + t.Run("embedding model is required", func(t *testing.T) { + cfg := base() + cfg.Embedding.Model = "" + assert.False(t, embeddingReconcileConfigured(cfg)) + }) + + t.Run("taxonomy integration is required", func(t *testing.T) { + cfg := base() + cfg.Taxonomy.ServiceURL = "" + assert.False(t, embeddingReconcileConfigured(cfg)) + }) + + t.Run("taxonomy token without a service URL is not configured", func(t *testing.T) { + cfg := base() + cfg.Taxonomy.ServiceURL = "" + cfg.Taxonomy.ServiceToken = "configured-token" + assert.False(t, embeddingReconcileConfigured(cfg)) + }) +} diff --git a/internal/config/config.go b/internal/config/config.go index e2d215a1..5d6d59f7 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -49,6 +49,12 @@ var ( //nolint:gosec // test default URL, not a production secret const DefaultDatabaseURL = "postgres://postgres:postgres@localhost:5432/test_db?sslmode=disable" +const ( + defaultEmbeddingJobTimeout = 60 * time.Second + defaultEmbeddingReconcileInterval = 5 * time.Minute + defaultEmbeddingReconcileRetryAfter = 15 * time.Minute +) + // Config holds all application configuration in nested groups. type Config struct { Server ServerConfig @@ -134,19 +140,25 @@ type MessagePublisherConfig struct { // EmbeddingConfig holds embedding provider and queue settings. type EmbeddingConfig struct { - ProviderAPIKey string `env:"EMBEDDING_PROVIDER_API_KEY"` - Provider string `env:"EMBEDDING_PROVIDER"` - Model string `env:"EMBEDDING_MODEL"` - BaseURL string `env:"EMBEDDING_BASE_URL"` - MaxConcurrent int `env:"EMBEDDING_MAX_CONCURRENT" env-default:"5"` - MaxAttempts int `env:"EMBEDDING_MAX_ATTEMPTS" env-default:"3"` - BatchSize int `env:"EMBEDDING_BATCH_SIZE" env-default:"1"` - BatchMaxWaitMs int `env:"EMBEDDING_BATCH_MAX_WAIT_MS" env-default:"25"` - BatchMaxInFlight int `env:"EMBEDDING_BATCH_MAX_IN_FLIGHT" env-default:"1"` - HTTPDisableKeepAlives bool `env:"EMBEDDING_HTTP_DISABLE_KEEP_ALIVES" env-default:"false"` - Normalize bool `env:"EMBEDDING_NORMALIZE" env-default:"false"` - GoogleCloudProject string `env:"EMBEDDING_GOOGLE_CLOUD_PROJECT"` - GoogleCloudLocation string `env:"EMBEDDING_GOOGLE_CLOUD_LOCATION"` + ProviderAPIKey string `env:"EMBEDDING_PROVIDER_API_KEY"` + Provider string `env:"EMBEDDING_PROVIDER"` + Model string `env:"EMBEDDING_MODEL"` + BaseURL string `env:"EMBEDDING_BASE_URL"` + MaxConcurrent int `env:"EMBEDDING_MAX_CONCURRENT" env-default:"5"` + MaxAttempts int `env:"EMBEDDING_MAX_ATTEMPTS" env-default:"5"` + JobTimeout DurationSec `env:"EMBEDDING_JOB_TIMEOUT_SECONDS" env-default:"60"` + BatchSize int `env:"EMBEDDING_BATCH_SIZE" env-default:"1"` + BatchMaxWaitMs int `env:"EMBEDDING_BATCH_MAX_WAIT_MS" env-default:"25"` + BatchMaxInFlight int `env:"EMBEDDING_BATCH_MAX_IN_FLIGHT" env-default:"1"` + ReconcileEnabled bool `env:"EMBEDDING_RECONCILE_ENABLED"` + ReconcileInterval DurationSec `env:"EMBEDDING_RECONCILE_INTERVAL_SECONDS" env-default:"300"` + ReconcileRetryAfter DurationSec `env:"EMBEDDING_RECONCILE_RETRY_AFTER_SECONDS" env-default:"900"` + ReconcileTargetDepth int `env:"EMBEDDING_RECONCILE_TARGET_DEPTH" env-default:"100"` + ReconcileMaxConcurrent int `env:"EMBEDDING_RECONCILE_MAX_CONCURRENT" env-default:"1"` + HTTPDisableKeepAlives bool `env:"EMBEDDING_HTTP_DISABLE_KEEP_ALIVES" env-default:"false"` + Normalize bool `env:"EMBEDDING_NORMALIZE" env-default:"false"` + GoogleCloudProject string `env:"EMBEDDING_GOOGLE_CLOUD_PROJECT"` + GoogleCloudLocation string `env:"EMBEDDING_GOOGLE_CLOUD_LOCATION"` } // TranslationConfig holds the feedback open-text translation enrichment settings @@ -449,7 +461,6 @@ func applyDefaults(cfg *Config) { // or, worse, flow into InsertOpts where River substitutes its default of 25 attempts — 25 // LLM calls per failing job instead of the intended 3. for _, tunables := range []struct{ maxConcurrent, maxAttempts *int }{ - {&cfg.Embedding.MaxConcurrent, &cfg.Embedding.MaxAttempts}, {&cfg.Translation.MaxConcurrent, &cfg.Translation.MaxAttempts}, {&cfg.Sentiment.MaxConcurrent, &cfg.Sentiment.MaxAttempts}, {&cfg.Emotions.MaxConcurrent, &cfg.Emotions.MaxAttempts}, @@ -463,6 +474,18 @@ func applyDefaults(cfg *Config) { } } + if cfg.Embedding.MaxConcurrent <= 0 { + cfg.Embedding.MaxConcurrent = 5 + } + + if cfg.Embedding.MaxAttempts <= 0 { + cfg.Embedding.MaxAttempts = 5 + } + + if cfg.Embedding.JobTimeout.Duration() <= 0 { + cfg.Embedding.JobTimeout = DurationSec(defaultEmbeddingJobTimeout) + } + if cfg.Embedding.BatchSize <= 0 { cfg.Embedding.BatchSize = 1 } @@ -475,6 +498,22 @@ func applyDefaults(cfg *Config) { cfg.Embedding.BatchMaxInFlight = 1 } + if cfg.Embedding.ReconcileInterval.Duration() <= 0 { + cfg.Embedding.ReconcileInterval = DurationSec(defaultEmbeddingReconcileInterval) + } + + if cfg.Embedding.ReconcileRetryAfter.Duration() <= 0 { + cfg.Embedding.ReconcileRetryAfter = DurationSec(defaultEmbeddingReconcileRetryAfter) + } + + if cfg.Embedding.ReconcileTargetDepth <= 0 { + cfg.Embedding.ReconcileTargetDepth = 100 + } + + if cfg.Embedding.ReconcileMaxConcurrent <= 0 { + cfg.Embedding.ReconcileMaxConcurrent = 1 + } + // Default the cache size only when the operator did not set it. An explicit 0 (or // negative) disables the cache: NewCachedTenantSettings treats size <= 0 as "no // caching". cleanenv does not reliably apply env-default to nested-struct fields, so diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 2be7f11e..049dd4bd 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -391,6 +391,36 @@ func TestLoad_EmbeddingBatchSettings(t *testing.T) { } } +func TestLoad_EmbeddingRecoverySettings(t *testing.T) { + t.Setenv("API_KEY", "test-api-key") + t.Setenv("EMBEDDING_MAX_ATTEMPTS", "7") + t.Setenv("EMBEDDING_JOB_TIMEOUT_SECONDS", "75") + t.Setenv("EMBEDDING_RECONCILE_ENABLED", "true") + t.Setenv("EMBEDDING_RECONCILE_INTERVAL_SECONDS", "240") + t.Setenv("EMBEDDING_RECONCILE_RETRY_AFTER_SECONDS", "600") + t.Setenv("EMBEDDING_RECONCILE_TARGET_DEPTH", "64") + t.Setenv("EMBEDDING_RECONCILE_MAX_CONCURRENT", "2") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load() error = %v", err) + } + + if cfg.Embedding.MaxAttempts != 7 || cfg.Embedding.JobTimeout.Duration() != 75*time.Second { + t.Errorf("Embedding attempt/timeout = (%d, %v), want (7, 75s)", + cfg.Embedding.MaxAttempts, cfg.Embedding.JobTimeout.Duration()) + } + + if !cfg.Embedding.ReconcileEnabled || cfg.Embedding.ReconcileInterval.Duration() != 4*time.Minute || + cfg.Embedding.ReconcileRetryAfter.Duration() != 10*time.Minute || + cfg.Embedding.ReconcileTargetDepth != 64 || cfg.Embedding.ReconcileMaxConcurrent != 2 { + t.Errorf("Embedding reconcile settings = (%v, %v, %v, %d, %d), want (true, 4m, 10m, 64, 2)", + cfg.Embedding.ReconcileEnabled, cfg.Embedding.ReconcileInterval.Duration(), + cfg.Embedding.ReconcileRetryAfter.Duration(), + cfg.Embedding.ReconcileTargetDepth, cfg.Embedding.ReconcileMaxConcurrent) + } +} + func TestLoad_EmbeddingBaseURLValidation(t *testing.T) { tests := []struct { name string @@ -686,8 +716,21 @@ func TestApplyDefaults(t *testing.T) { t.Errorf("Embedding.MaxConcurrent = %d, want 5", cfg.Embedding.MaxConcurrent) } - if cfg.Embedding.MaxAttempts != 3 { - t.Errorf("Embedding.MaxAttempts = %d, want 3", cfg.Embedding.MaxAttempts) + if cfg.Embedding.MaxAttempts != 5 { + t.Errorf("Embedding.MaxAttempts = %d, want 5", cfg.Embedding.MaxAttempts) + } + + if cfg.Embedding.JobTimeout.Duration() != 60*time.Second { + t.Errorf("Embedding.JobTimeout = %v, want 60s", cfg.Embedding.JobTimeout.Duration()) + } + + if cfg.Embedding.ReconcileInterval.Duration() != 5*time.Minute || + cfg.Embedding.ReconcileRetryAfter.Duration() != 15*time.Minute || + cfg.Embedding.ReconcileTargetDepth != 100 || cfg.Embedding.ReconcileMaxConcurrent != 1 { + t.Errorf("Embedding reconcile defaults = (%v, %v, %d, %d), want (5m, 15m, 100, 1)", + cfg.Embedding.ReconcileInterval.Duration(), cfg.Embedding.ReconcileRetryAfter.Duration(), + cfg.Embedding.ReconcileTargetDepth, + cfg.Embedding.ReconcileMaxConcurrent) } if cfg.Embedding.BatchSize != 1 || cfg.Embedding.BatchMaxWaitMs != 25 || cfg.Embedding.BatchMaxInFlight != 1 { diff --git a/internal/googleai/client.go b/internal/googleai/client.go index 22e5d153..1bfa7e5c 100644 --- a/internal/googleai/client.go +++ b/internal/googleai/client.go @@ -391,9 +391,38 @@ func wrapGenaiError(op string, err error) error { return huberrors.NewRateLimitError(genaiRetryAfter(apiErr), wrapped) } + if isGenaiInputLengthError(apiErr) { + return huberrors.NewTerminalProviderError(huberrors.TerminalReasonLength, wrapped) + } + return wrapped } +// isGenaiInputLengthError deliberately requires an explicit token/input limit phrase. INVALID_ARGUMENT +// also covers deployment mistakes, which must stay retryable after an operator corrects them rather +// than becoming permanent record exclusions. +func isGenaiInputLengthError(apiErr genai.APIError) bool { + if apiErr.Code != http.StatusBadRequest && apiErr.Code != http.StatusRequestEntityTooLarge { + return false + } + + message := strings.ToLower(apiErr.Message) + for _, marker := range []string{ + "input token count exceeds", + "maximum number of tokens", + "maximum context length", + "context length exceeded", + "input is too long", + "too many tokens", + } { + if strings.Contains(message, marker) { + return true + } + } + + return false +} + // genaiRetryAfter extracts the RetryInfo retryDelay from a RESOURCE_EXHAUSTED error's // details, or 0 when absent or unparseable. func genaiRetryAfter(apiErr genai.APIError) time.Duration { diff --git a/internal/googleai/client_test.go b/internal/googleai/client_test.go index be24cb56..79dcfa1a 100644 --- a/internal/googleai/client_test.go +++ b/internal/googleai/client_test.go @@ -322,6 +322,52 @@ func TestCreateEmbedding_RateLimitReturnsRateLimitError(t *testing.T) { assert.Equal(t, 17*time.Second, rateLimited.RetryAfter) } +func TestWrapGenaiErrorClassifiesOnlyExplicitInputLengthFailures(t *testing.T) { + tests := []struct { + name string + err genai.APIError + terminal bool + }{ + { + name: "input token limit", + err: genai.APIError{ + Code: http.StatusBadRequest, + Status: "INVALID_ARGUMENT", + Message: "The input token count exceeds the maximum number of tokens allowed", + }, + terminal: true, + }, + { + name: "generic invalid argument stays recoverable", + err: genai.APIError{ + Code: http.StatusBadRequest, + Status: "INVALID_ARGUMENT", + Message: "the configured output dimensionality is unsupported", + }, + }, + { + name: "permission failure stays recoverable", + err: genai.APIError{ + Code: http.StatusForbidden, + Status: "PERMISSION_DENIED", + Message: "permission denied", + }, + }, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + err := wrapGenaiError("gemini embedding", testCase.err) + reason, terminal := huberrors.TerminalReasonOf(err) + assert.Equal(t, testCase.terminal, terminal) + + if terminal { + assert.Equal(t, huberrors.TerminalReasonLength, reason) + } + }) + } +} + // TestTerminalEmptyReason pins which empty Gemini responses are permanent for the input and which // stay retryable. Same asymmetry as the OpenAI client: when a reason is ambiguous, retry, because // a false terminal abandons a record for good. diff --git a/internal/models/enrichment_failure.go b/internal/models/enrichment_failure.go index 380a3fef..3f7ee72d 100644 --- a/internal/models/enrichment_failure.go +++ b/internal/models/enrichment_failure.go @@ -1,6 +1,10 @@ package models -import "github.com/google/uuid" +import ( + "time" + + "github.com/google/uuid" +) // The enrichment names. These are the values in migration 022's CHECK, the `enrichment` metric // label, and the discriminator on every failure marker, so they are declared once rather than @@ -9,6 +13,9 @@ const ( EnrichmentNameTranslation = "translation" EnrichmentNameSentiment = "sentiment" EnrichmentNameEmotions = "emotions" + // EnrichmentNameTaxonomyEmbedding is the translated-text embedding consumed by taxonomy. + // Raw search embeddings are intentionally not reconciled by this pipeline. + EnrichmentNameTaxonomyEmbedding = "taxonomy_embedding" ) // The two non-terminal reasons. Both mean "did not succeed this time"; they differ in which half @@ -49,4 +56,9 @@ type EnrichmentFailure struct { Reason string // Attempts spent before giving up. Diagnostic only. Attempts int + // ContextKey and SourceUpdatedAt bind a taxonomy-embedding marker to the exact model and + // record revision that failed. They are empty for the other enrichment types. Without both, + // a terminal failure from an old model or content revision could suppress repair forever. + ContextKey string + SourceUpdatedAt *time.Time } diff --git a/internal/models/taxonomy.go b/internal/models/taxonomy.go index fc5e693c..aebf21ae 100644 --- a/internal/models/taxonomy.go +++ b/internal/models/taxonomy.go @@ -77,14 +77,16 @@ type TaxonomyScope struct { // TaxonomyFieldOption describes a feedback field that can be used for taxonomy generation. type TaxonomyFieldOption struct { - TenantID string `json:"tenant_id"` - SourceType string `json:"source_type"` - SourceID string `json:"source_id"` - SourceName string `json:"source_name,omitempty"` - FieldID string `json:"field_id"` - FieldLabel string `json:"field_label,omitempty"` - RecordCount int `json:"record_count"` - EmbeddingCount int `json:"embedding_count"` + TenantID string `json:"tenant_id"` + SourceType string `json:"source_type"` + SourceID string `json:"source_id"` + SourceName string `json:"source_name,omitempty"` + FieldID string `json:"field_id"` + FieldLabel string `json:"field_label,omitempty"` + RecordCount int `json:"record_count"` + EmbeddingCount int `json:"embedding_count"` + EmbeddingFailedCount int `json:"embedding_failed_count"` + EmbeddingFailedTerminalCount int `json:"embedding_failed_terminal_count"` } // TaxonomyFieldsResponse contains taxonomy-capable field options. diff --git a/internal/observability/names.go b/internal/observability/names.go index a285abc9..bff49e06 100644 --- a/internal/observability/names.go +++ b/internal/observability/names.go @@ -138,12 +138,14 @@ var allowedEmbeddingOutcomeStatuses = map[string]bool{ // allowedEmbeddingWorkerReasons for hub_embedding_worker_errors_total. var allowedEmbeddingWorkerReasons = map[string]bool{ - "embedding_api_failed": true, - "get_record_failed": true, - "update_failed": true, - "tenant_write_conflict": true, - "rate_limited": true, - "superseded": true, + "embedding_api_failed": true, + "get_record_failed": true, + "update_failed": true, + "tenant_write_conflict": true, + "rate_limited": true, + "reconcile_failed": true, + "failure_marker_write_failed": true, + "superseded": true, } // AllowedEmbeddingProviderReason returns true if reason is allowed for embedding provider errors. diff --git a/internal/openai/client.go b/internal/openai/client.go index 0623a475..f6ca1c8c 100644 --- a/internal/openai/client.go +++ b/internal/openai/client.go @@ -437,9 +437,43 @@ func wrapOpenAIError(op string, err error) error { return huberrors.NewRateLimitError(openaiRetryAfter(apiErr), wrapped) } + if isOpenAIInputLengthError(apiErr) { + return huberrors.NewTerminalProviderError(huberrors.TerminalReasonLength, wrapped) + } + return wrapped } +// isOpenAIInputLengthError recognizes only explicit per-request input/context limit failures. +// Other 4xx responses (auth, model configuration, dimensions, malformed requests) remain +// retryable at the record layer because classifying them as terminal would permanently suppress +// every record after an operator fixes the deployment. +func isOpenAIInputLengthError(apiErr *openaisdk.Error) bool { + if apiErr == nil || (apiErr.StatusCode != http.StatusBadRequest && apiErr.StatusCode != http.StatusRequestEntityTooLarge) { + return false + } + + code := strings.ToLower(strings.TrimSpace(apiErr.Code)) + if code == "context_length_exceeded" || code == "input_too_long" || code == "max_tokens_exceeded" { + return true + } + + message := strings.ToLower(apiErr.Message) + for _, marker := range []string{ + "maximum context length", + "context length exceeded", + "input token count exceeds", + "input is too long", + "too many tokens", + } { + if strings.Contains(message, marker) { + return true + } + } + + return false +} + // openaiRetryAfter reads the retry hint from a 429 response: retry-after-ms // (milliseconds — OpenAI sends it for sub-second waits), then Retry-After as // delta-seconds (integer or fractional per the SDK's own parser), then Retry-After diff --git a/internal/openai/client_test.go b/internal/openai/client_test.go index 44cbeac6..d30e70b3 100644 --- a/internal/openai/client_test.go +++ b/internal/openai/client_test.go @@ -639,6 +639,58 @@ func TestCreateEmbedding_RateLimitReturnsRateLimitError(t *testing.T) { assert.Equal(t, 9*time.Second, rateLimited.RetryAfter) } +func TestWrapOpenAIErrorClassifiesOnlyExplicitInputLengthFailures(t *testing.T) { + tests := []struct { + name string + err *openaisdk.Error + terminal bool + }{ + { + name: "standard context length code", + err: &openaisdk.Error{ + StatusCode: http.StatusBadRequest, + Code: "context_length_exceeded", + }, + terminal: true, + }, + { + name: "vllm context length message", + err: &openaisdk.Error{ + StatusCode: http.StatusBadRequest, + Message: "This model's maximum context length is 8192 tokens", + }, + terminal: true, + }, + { + name: "invalid model stays recoverable", + err: &openaisdk.Error{ + StatusCode: http.StatusBadRequest, + Code: "model_not_found", + Message: "the configured model does not exist", + }, + }, + { + name: "permission failure stays recoverable", + err: &openaisdk.Error{ + StatusCode: http.StatusForbidden, + Message: "permission denied", + }, + }, + } + + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { + err := wrapOpenAIError("openai embedding", testCase.err) + reason, terminal := huberrors.TerminalReasonOf(err) + assert.Equal(t, testCase.terminal, terminal) + + if terminal { + assert.Equal(t, huberrors.TerminalReasonLength, reason) + } + }) + } +} + // TestCompletionTextTerminalClassification pins which empty-completion outcomes are permanent for // the input and which stay retryable. The asymmetry is deliberate: a false terminal abandons a // record for good, a false transient costs a few wasted calls. diff --git a/internal/repository/embeddings_repository.go b/internal/repository/embeddings_repository.go index d41a91b8..015a6a43 100644 --- a/internal/repository/embeddings_repository.go +++ b/internal/repository/embeddings_repository.go @@ -23,6 +23,20 @@ import ( var errEmbeddingBackfillTenantRequired = errors.New("tenant id is required for tenant embedding backfill") const ( + // trimSpaceCharactersSQL is the complete Unicode White_Space set used by Go strings.TrimSpace. + // Unicode escape literals keep the characters visible and make the result stable across supported + // PostgreSQL versions (PostgreSQL 16 interprets E'\v' as the literal letter "v"). + trimSpaceCharactersSQL = `U&'\0009\000A\000B\000C\000D\0020' || + U&'\0085\00A0\1680\2000\2001\2002\2003\2004\2005\2006\2007\2008\2009\200A' || + U&'\2028\2029\202F\205F\3000'` + // taxonomyEmbeddingInputTextSQL mirrors BuildEmbeddingInputFromValues: prefer non-blank + // translated text, then fall back to non-blank source text. + taxonomyEmbeddingInputTextSQL = `COALESCE( + NULLIF(btrim(fr.value_text_translated, ` + trimSpaceCharactersSQL + `), ''), + NULLIF(btrim(fr.value_text, ` + trimSpaceCharactersSQL + `), '') + )` + taxonomyEmbeddingEligibleTextSQL = taxonomyEmbeddingInputTextSQL + ` IS NOT NULL` + // hnswEfSearch increases HNSW graph traversal candidates (default 40); higher improves recall. hnswEfSearch = 200 // hnswIterativeScanMode makes the HNSW scan resume past ef_search candidates until the query's @@ -287,6 +301,106 @@ func (r *EmbeddingsRepository) ListTenantFeedbackRecordIDsForBackfillByInputKind return r.listFeedbackRecordIDsForBackfillByInputKind(ctx, model, inputKind, tenantID, true, afterID, limit) } +// ListPendingTaxonomyEmbeddingIDs returns eligible text records that still lack the exact +// taxonomy embedding model and do not already have an in-flight embedding job on any queue. +// Terminal failures are excluded: retrying content the provider has conclusively rejected would +// spend a call on every sweep forever. Non-terminal failures become eligible after retryBefore, +// preventing the same poison records from monopolizing every bounded sweep. Candidates are +// ordered oldest first so successful/cooldown-filtered sweeps advance through the backlog rather +// than repeatedly selecting whichever tenant is ingesting the newest records. +// +// The ordered eligibility scan is deliberately deployment-wide rather than tenant-scoped: the +// repair lane must make progress without requiring tenant discovery or allowing a busy tenant to +// monopolize it. Its operational impact is bounded by the five-minute sweep cadence and two-minute +// worker deadline, while the target-depth LIMIT bounds returned rows and queued work (not scan +// cost). Changing the access path requires production-scale query-plan evidence because the +// missing-embedding predicate lives in a separate table. +func (r *EmbeddingsRepository) ListPendingTaxonomyEmbeddingIDs( + ctx context.Context, model string, retryBefore time.Time, limit int, +) ([]uuid.UUID, error) { + if limit <= 0 { + return []uuid.UUID{}, nil + } + + rows, err := r.db.Query(ctx, ` + SELECT fr.id + FROM feedback_records fr + LEFT JOIN feedback_record_enrichment_failures failure + ON failure.feedback_record_id = fr.id + AND failure.enrichment = $3 + AND failure.context_key = $1 + AND failure.source_updated_at = fr.updated_at + WHERE fr.field_type = 'text' + AND `+taxonomyEmbeddingEligibleTextSQL+` + AND NOT EXISTS ( + SELECT 1 FROM embeddings e + WHERE e.feedback_record_id = fr.id AND e.model = $1 + ) + AND (failure.feedback_record_id IS NULL OR ( + NOT failure.terminal AND failure.failed_at <= $4 + )) + AND NOT EXISTS ( + SELECT 1 + FROM river_job job + WHERE job.kind = 'feedback_embedding' + AND job.state IN ('available', 'pending', 'retryable', 'running', 'scheduled') + -- River's GIN index on args supports containment; separate ->> equality + -- predicates would filter every job remaining after the kind lookup. + AND job.args @> jsonb_build_object( + 'feedback_record_id', fr.id::text, + 'model', $1::text, + 'input_kind', $2::text + ) + ) + ORDER BY fr.collected_at, fr.id + LIMIT $5`, + model, + models.EmbeddingInputKindTaxonomyTranslated, + models.EnrichmentNameTaxonomyEmbedding, + retryBefore, + limit, + ) + if err != nil { + return nil, fmt.Errorf("list pending taxonomy embedding ids: %w", err) + } + defer rows.Close() + + ids := make([]uuid.UUID, 0, limit) + + for rows.Next() { + var id uuid.UUID + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("scan pending taxonomy embedding id: %w", err) + } + + ids = append(ids, id) + } + + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate pending taxonomy embedding ids: %w", err) + } + + return ids, nil +} + +// CountRunnableEmbeddingJobs returns the current runnable depth of one embedding queue. Retryable +// jobs are included: although they are waiting out backoff rather than occupying a worker now, +// excluding them would let every sweep add another copy and defeat the target-depth bound. +func (r *EmbeddingsRepository) CountRunnableEmbeddingJobs(ctx context.Context, queue string) (int, error) { + var count int + if err := r.db.QueryRow(ctx, ` + SELECT COUNT(*)::int + FROM river_job + WHERE queue = $1 + AND kind = 'feedback_embedding' + AND state IN ('available', 'pending', 'retryable', 'running', 'scheduled')`, queue, + ).Scan(&count); err != nil { + return 0, fmt.Errorf("count runnable embedding jobs: %w", err) + } + + return count, nil +} + //nolint:funcorder // scoped helper stays with the public backfill methods func (r *EmbeddingsRepository) listFeedbackRecordIDsForBackfillByInputKind( ctx context.Context, diff --git a/internal/repository/enrichment_failures_repository.go b/internal/repository/enrichment_failures_repository.go index 866536fe..916bf995 100644 --- a/internal/repository/enrichment_failures_repository.go +++ b/internal/repository/enrichment_failures_repository.go @@ -31,8 +31,8 @@ func NewEnrichmentFailuresRepository(db *pgxpool.Pool) *EnrichmentFailuresReposi } // enrichmentFailureLockKeyParam is the placeholder carrying the tenant write-lock key, after the -// six inserted values. -const enrichmentFailureLockKeyParam = 7 +// eight inserted values. +const enrichmentFailureLockKeyParam = 9 // recordEnrichmentFailureSQL upserts the marker for one (record, enrichment). // @@ -49,14 +49,17 @@ const enrichmentFailureLockKeyParam = 7 // init, and every fragment is a compile-time literal, never caller input. var recordEnrichmentFailureSQL = ` INSERT INTO feedback_record_enrichment_failures - (feedback_record_id, enrichment, tenant_id, failed_at, attempts, terminal, reason) - SELECT $1, $2, $3, NOW(), $4, $5, $6 + (feedback_record_id, enrichment, tenant_id, failed_at, attempts, terminal, reason, + context_key, source_updated_at) + SELECT $1, $2, $3, NOW(), $4, $5, $6, NULLIF($7, ''), $8 WHERE ` + tenantWriteLockGate(enrichmentFailureLockKeyParam) + ` ON CONFLICT (feedback_record_id, enrichment) DO UPDATE SET failed_at = NOW(), attempts = EXCLUDED.attempts, terminal = EXCLUDED.terminal, - reason = EXCLUDED.reason` + reason = EXCLUDED.reason, + context_key = EXCLUDED.context_key, + source_updated_at = EXCLUDED.source_updated_at` // RecordFailure persists one enrichment failure. // @@ -81,6 +84,8 @@ func (r *EnrichmentFailuresRepository) RecordFailure(ctx context.Context, failur failure.Attempts, failure.Terminal, failure.Reason, + failure.ContextKey, + failure.SourceUpdatedAt, TenantWriteLockKey(failure.TenantID), ) if err != nil { diff --git a/internal/repository/enrichment_status_repository.go b/internal/repository/enrichment_status_repository.go index 28cca00b..b84ae49f 100644 --- a/internal/repository/enrichment_status_repository.go +++ b/internal/repository/enrichment_status_repository.go @@ -49,7 +49,8 @@ type EnrichmentStatusCounts struct { // enrichmentEligibleText is the data-level eligibility predicate: an open-text field with content. // -// It trims the full ASCII whitespace set (space, tab, VT, FF, CR, LF). This deliberately does NOT +// It trims the same Unicode White_Space set as Go strings.TrimSpace, using version-stable Unicode +// escapes so PostgreSQL 16 cannot interpret `\v` as the literal letter "v". 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 @@ -57,14 +58,10 @@ type EnrichmentStatusCounts struct { // 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') <> ''` +const enrichmentEligibleText = `fr.field_type = 'text' AND fr.value_text IS NOT NULL AND btrim(fr.value_text, ` + + trimSpaceCharactersSQL + `) <> ''` // enrichmentEffectiveTarget resolves a tenant's effective translation target: its own // target_language, falling back to the deployment default ($1). An empty result means translation @@ -248,7 +245,7 @@ const countTaxonomyEmbeddingBacklogAggregateSQL = ` SELECT COUNT(*) FROM feedback_records fr WHERE fr.field_type = 'text' - AND COALESCE(NULLIF(btrim(fr.value_text_translated), ''), NULLIF(btrim(fr.value_text), '')) IS NOT NULL + AND ` + taxonomyEmbeddingEligibleTextSQL + ` AND NOT EXISTS ( SELECT 1 FROM embeddings e @@ -543,7 +540,15 @@ var countFailedRecordsAggregateSQL = ` AND ((f.enrichment = 'sentiment' AND fr.sentiment IS NULL AND ` + enrichmentSentimentOn + `) OR (f.enrichment = 'emotions' AND fr.emotions_classified_at IS NULL AND fr.emotions IS NULL AND ` + enrichmentEmotionsOn + `) - OR (f.enrichment = 'translation' AND fr.translation_lang_key IS NULL)) + OR (f.enrichment = 'translation' AND fr.translation_lang_key IS NULL) + OR (f.enrichment = 'taxonomy_embedding' + AND $1 <> '' + AND f.context_key = $1 + AND f.source_updated_at = fr.updated_at + AND NOT EXISTS ( + SELECT 1 FROM embeddings e + WHERE e.feedback_record_id = fr.id AND e.model = $1 + ))) GROUP BY f.enrichment, f.terminal` // CountFailedRecordsAggregate returns the cross-tenant failed-record counts per enrichment. @@ -556,9 +561,9 @@ var countFailedRecordsAggregateSQL = ` // for a deployment-wide gauge, and written down so the two are not mistaken for a bug when they // disagree. func (r *EnrichmentStatusRepository) CountFailedRecordsAggregate( - ctx context.Context, + ctx context.Context, taxonomyEmbeddingModel string, ) ([]FailedRecordCount, error) { - rows, err := r.db.Query(ctx, countFailedRecordsAggregateSQL) + rows, err := r.db.Query(ctx, countFailedRecordsAggregateSQL, taxonomyEmbeddingModel) if err != nil { return nil, fmt.Errorf("count failed records: %w", err) } diff --git a/internal/repository/feedback_records_repository.go b/internal/repository/feedback_records_repository.go index 30f0b372..80155f2f 100644 --- a/internal/repository/feedback_records_repository.go +++ b/internal/repository/feedback_records_repository.go @@ -302,7 +302,15 @@ func (r *FeedbackRecordsRepository) SetTranslation( // The record now has no translation and, with empty text, is owed none — so a marker // describing a failed attempt on the old text has nothing left to describe. - return clearEnrichmentFailure(ctx, dbTx, feedbackRecordID, models.EnrichmentNameTranslation) + if err := clearEnrichmentFailure( + ctx, dbTx, feedbackRecordID, models.EnrichmentNameTranslation, + ); err != nil { + return err + } + + return clearEnrichmentFailure( + ctx, dbTx, feedbackRecordID, models.EnrichmentNameTaxonomyEmbedding, + ) } // Setting a translation persists only while langKey still equals the tenant's current @@ -337,7 +345,18 @@ func (r *FeedbackRecordsRepository) SetTranslation( return huberrors.ErrTranslationSuperseded } - return clearEnrichmentFailure(ctx, dbTx, feedbackRecordID, models.EnrichmentNameTranslation) + if err := clearEnrichmentFailure( + ctx, dbTx, feedbackRecordID, models.EnrichmentNameTranslation, + ); err != nil { + return err + } + + // A new translation changes the taxonomy embedding input. Clear any terminal marker from + // the previous content so the new input is eligible for reconciliation if its direct + // enqueue is lost. + return clearEnrichmentFailure( + ctx, dbTx, feedbackRecordID, models.EnrichmentNameTaxonomyEmbedding, + ) }) } diff --git a/internal/repository/taxonomy_repository.go b/internal/repository/taxonomy_repository.go index 69de2cc7..b0a29839 100644 --- a/internal/repository/taxonomy_repository.go +++ b/internal/repository/taxonomy_repository.go @@ -68,11 +68,22 @@ func (r *TaxonomyRepository) ListFieldOptions( fr.field_id, COALESCE(MAX(fr.field_label) FILTER (WHERE fr.field_label IS NOT NULL AND btrim(fr.field_label) <> ''), ''), COUNT(*)::int, - COUNT(e.feedback_record_id)::int + COUNT(e.feedback_record_id)::int, + COUNT(failure.feedback_record_id) FILTER ( + WHERE e.feedback_record_id IS NULL AND NOT failure.terminal + )::int, + COUNT(failure.feedback_record_id) FILTER ( + WHERE e.feedback_record_id IS NULL AND failure.terminal + )::int FROM feedback_records fr LEFT JOIN embeddings e ON e.feedback_record_id = fr.id AND e.model = $2 + LEFT JOIN feedback_record_enrichment_failures failure + ON failure.feedback_record_id = fr.id + AND failure.enrichment = 'taxonomy_embedding' + AND failure.context_key = $2 + AND failure.source_updated_at = fr.updated_at WHERE fr.tenant_id = $1 - AND COALESCE(NULLIF(btrim(fr.value_text_translated), ''), NULLIF(btrim(fr.value_text), '')) IS NOT NULL + AND `+taxonomyEmbeddingEligibleTextSQL+` GROUP BY fr.tenant_id, fr.source_type, COALESCE(NULLIF(btrim(fr.source_id), ''), ''), fr.field_id ORDER BY fr.source_type, COALESCE(NULLIF(btrim(fr.source_id), ''), ''), fr.field_id`, tenantID, embeddingModel, @@ -95,6 +106,8 @@ func (r *TaxonomyRepository) ListFieldOptions( &option.FieldLabel, &option.RecordCount, &option.EmbeddingCount, + &option.EmbeddingFailedCount, + &option.EmbeddingFailedTerminalCount, ); err != nil { return nil, fmt.Errorf("scan taxonomy field option: %w", err) } @@ -129,7 +142,7 @@ func (r *TaxonomyRepository) CountScopeInput( FROM feedback_records fr LEFT JOIN embeddings e ON e.feedback_record_id = fr.id AND e.model = $2 WHERE fr.tenant_id = $1 - AND COALESCE(NULLIF(btrim(fr.value_text_translated), ''), NULLIF(btrim(fr.value_text), '')) IS NOT NULL`, + AND `+taxonomyEmbeddingEligibleTextSQL, scope.TenantID, embeddingModel, ).Scan(&recordCount, &embeddingCount) if err != nil { @@ -150,7 +163,7 @@ func (r *TaxonomyRepository) CountScopeInput( AND fr.source_type = $2 AND NULLIF(btrim(fr.source_id), '') IS NOT DISTINCT FROM NULLIF(btrim($3), '') AND fr.field_id = $4 - AND COALESCE(NULLIF(btrim(fr.value_text_translated), ''), NULLIF(btrim(fr.value_text), '')) IS NOT NULL`, + AND `+taxonomyEmbeddingEligibleTextSQL, scope.TenantID, scope.SourceType, scope.SourceID, scope.FieldID, embeddingModel, ).Scan(&recordCount, &embeddingCount, &fieldLabel) if err != nil { @@ -1445,10 +1458,7 @@ func materializeRunInputSnapshot( FROM feedback_records fr INNER JOIN embeddings e ON e.feedback_record_id = fr.id AND e.model = $4 WHERE fr.tenant_id = $2 - AND COALESCE( - NULLIF(btrim(fr.value_text_translated), ''), - NULLIF(btrim(fr.value_text), '') - ) IS NOT NULL + AND `+taxonomyEmbeddingEligibleTextSQL+` ORDER BY fr.collected_at DESC, fr.id ASC LIMIT $3 ) selected`, @@ -1475,10 +1485,7 @@ func materializeRunInputSnapshot( AND fr.source_type = $3 AND NULLIF(btrim(fr.source_id), '') IS NOT DISTINCT FROM NULLIF(btrim($4), '') AND fr.field_id = $5 - AND COALESCE( - NULLIF(btrim(fr.value_text_translated), ''), - NULLIF(btrim(fr.value_text), '') - ) IS NOT NULL + AND `+taxonomyEmbeddingEligibleTextSQL+` ORDER BY fr.collected_at DESC, fr.id ASC LIMIT $6 ) selected`, @@ -1562,7 +1569,7 @@ func queryMaterializedRunInputRows( COALESCE(NULLIF(btrim(fr.source_id), ''), ''), fr.field_id, COALESCE(fr.field_label, ''), - COALESCE(NULLIF(btrim(fr.value_text_translated), ''), NULLIF(btrim(fr.value_text), '')), + `+taxonomyEmbeddingInputTextSQL+`, e.embedding FROM taxonomy_run_input_records input INNER JOIN feedback_records fr diff --git a/internal/service/embedding_reconcile.go b/internal/service/embedding_reconcile.go new file mode 100644 index 00000000..c8729a40 --- /dev/null +++ b/internal/service/embedding_reconcile.go @@ -0,0 +1,150 @@ +package service + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/google/uuid" + "github.com/riverqueue/river" + "github.com/riverqueue/river/rivertype" + + "github.com/formbricks/hub/internal/models" +) + +// ErrEmbeddingReconcileInserterUnset reports a sweep invoked before the River client was attached. +var ErrEmbeddingReconcileInserterUnset = errors.New("embedding reconcile: inserter not set") + +const embeddingRepairPriority = 4 + +// EmbeddingReconcileRepository is the data boundary needed by the taxonomy embedding sweep. +type EmbeddingReconcileRepository interface { + ListPendingTaxonomyEmbeddingIDs( + ctx context.Context, model string, retryBefore time.Time, limit int, + ) ([]uuid.UUID, error) + CountRunnableEmbeddingJobs(ctx context.Context, queue string) (int, error) +} + +// EmbeddingReconcileService keeps the low-priority repair lane topped up without materializing the +// complete deployment backlog or crowding live embedding work. +type EmbeddingReconcileService struct { + repo EmbeddingReconcileRepository + inserter RiverBatchInserter + model string + targetDepth int + maxAttempts int + retryAfter time.Duration +} + +// NewEmbeddingReconcileService creates a level-triggered taxonomy embedding reconciler. +func NewEmbeddingReconcileService( + repo EmbeddingReconcileRepository, + model string, + targetDepth int, + maxAttempts int, + retryAfter time.Duration, +) *EmbeddingReconcileService { + return &EmbeddingReconcileService{ + repo: repo, + model: model, + targetDepth: targetDepth, + maxAttempts: maxAttempts, + retryAfter: retryAfter, + } +} + +// SetInserter attaches River after its worker registry has been built. +func (s *EmbeddingReconcileService) SetInserter(inserter RiverBatchInserter) { + s.inserter = inserter +} + +// EmbeddingReconcileResult reports one sweep's bounded queue action. +type EmbeddingReconcileResult struct { + Found int + Enqueued int + Depth int + AtTarget bool +} + +// Sweep tops the repair queue up to targetDepth with records that are still missing their current +// taxonomy embedding. The query excludes active jobs across all queues and terminal failures, +// and cools transient failures before making them eligible again. +func (s *EmbeddingReconcileService) Sweep(ctx context.Context) (EmbeddingReconcileResult, error) { + result := EmbeddingReconcileResult{} + if s.inserter == nil { + return result, ErrEmbeddingReconcileInserterUnset + } + + depth, err := s.repo.CountRunnableEmbeddingJobs(ctx, EmbeddingsReconcileQueueName) + if err != nil { + return result, fmt.Errorf("read repair queue depth: %w", err) + } + + result.Depth = depth + + room := s.targetDepth - depth + if room <= 0 { + result.AtTarget = true + + return result, nil + } + + ids, err := s.repo.ListPendingTaxonomyEmbeddingIDs(ctx, s.model, time.Now().Add(-s.retryAfter), room) + if err != nil { + return result, fmt.Errorf("list pending taxonomy embeddings: %w", err) + } + + result.Found = len(ids) + + if len(ids) == 0 { + return result, nil + } + + params := make([]river.InsertManyParams, 0, len(ids)) + for _, id := range ids { + params = append(params, river.InsertManyParams{ + Args: FeedbackEmbeddingArgs{ + FeedbackRecordID: id, + Model: s.model, + InputKind: models.EmbeddingInputKindTaxonomyTranslated, + ValueTextHash: "reconcile", + }, + InsertOpts: &river.InsertOpts{ + Queue: EmbeddingsReconcileQueueName, + Priority: embeddingRepairPriority, + MaxAttempts: s.maxAttempts, + UniqueOpts: river.UniqueOpts{ + ByArgs: true, + ByState: []rivertype.JobState{ + rivertype.JobStateAvailable, + rivertype.JobStatePending, + rivertype.JobStateRetryable, + rivertype.JobStateRunning, + rivertype.JobStateScheduled, + }, + }, + }, + }) + } + + results, err := s.inserter.InsertMany(ctx, params) + if err != nil { + return result, fmt.Errorf("enqueue taxonomy embedding repairs: %w", err) + } + + for _, inserted := range results { + if inserted != nil && !inserted.UniqueSkippedAsDuplicate { + result.Enqueued++ + } + } + + slog.InfoContext(ctx, "embedding reconcile: repair jobs enqueued", + "found", result.Found, + "enqueued", result.Enqueued, + "queue_depth_before", result.Depth, + ) + + return result, nil +} diff --git a/internal/service/embedding_reconcile_job_args.go b/internal/service/embedding_reconcile_job_args.go new file mode 100644 index 00000000..b7e180bb --- /dev/null +++ b/internal/service/embedding_reconcile_job_args.go @@ -0,0 +1,21 @@ +package service + +import "github.com/riverqueue/river" + +const ( + // EmbeddingReconcileQueueName carries the level-triggered sweep itself. The queue is serialized; + // River's elected leader schedules one job per interval across all worker replicas. + EmbeddingReconcileQueueName = "embedding_reconcile" + // EmbeddingsReconcileQueueName carries repaired taxonomy embeddings. It is deliberately + // separate from live embeddings so old backlog cannot delay a record that just arrived. + EmbeddingsReconcileQueueName = "embeddings_reconcile" +) + +// EmbeddingReconcileArgs is an argument-free, level-triggered taxonomy embedding sweep. +// Database state and deployment configuration are read when the job runs. +type EmbeddingReconcileArgs struct{} + +// Kind returns the River job kind. +func (EmbeddingReconcileArgs) Kind() string { return "embedding_reconcile" } + +var _ river.JobArgs = EmbeddingReconcileArgs{} diff --git a/internal/service/embedding_reconcile_test.go b/internal/service/embedding_reconcile_test.go new file mode 100644 index 00000000..1e08cfd8 --- /dev/null +++ b/internal/service/embedding_reconcile_test.go @@ -0,0 +1,187 @@ +package service + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/google/uuid" + "github.com/riverqueue/river" + "github.com/riverqueue/river/rivertype" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/models" +) + +type stubEmbeddingReconcileRepository struct { + depth int + depthErr error + ids []uuid.UUID + listErr error + listModel string + listLimit int + retryBefore time.Time + depthQueue string +} + +func (r *stubEmbeddingReconcileRepository) ListPendingTaxonomyEmbeddingIDs( + _ context.Context, model string, retryBefore time.Time, limit int, +) ([]uuid.UUID, error) { + r.listModel = model + r.listLimit = limit + r.retryBefore = retryBefore + + if r.listErr != nil { + return nil, r.listErr + } + + if len(r.ids) > limit { + return r.ids[:limit], nil + } + + return r.ids, nil +} + +func (r *stubEmbeddingReconcileRepository) CountRunnableEmbeddingJobs( + _ context.Context, queue string, +) (int, error) { + r.depthQueue = queue + + return r.depth, r.depthErr +} + +type recordingBatchInserter struct { + params []river.InsertManyParams + results []*rivertype.JobInsertResult + err error +} + +func (i *recordingBatchInserter) InsertMany( + _ context.Context, params []river.InsertManyParams, +) ([]*rivertype.JobInsertResult, error) { + i.params = params + + if i.err != nil { + return nil, i.err + } + + if i.results != nil { + return i.results, nil + } + + results := make([]*rivertype.JobInsertResult, len(params)) + for index := range results { + results[index] = &rivertype.JobInsertResult{} + } + + return results, nil +} + +func TestEmbeddingReconcileSweepTopsUpBoundedRepairQueue(t *testing.T) { + ids := []uuid.UUID{uuid.Must(uuid.NewV7()), uuid.Must(uuid.NewV7()), uuid.Must(uuid.NewV7())} + repo := &stubEmbeddingReconcileRepository{depth: 98, ids: ids} + inserter := &recordingBatchInserter{} + retryAfter := 15 * time.Minute + reconciler := NewEmbeddingReconcileService(repo, "taxonomy:model", 100, 5, retryAfter) + reconciler.SetInserter(inserter) + + started := time.Now() + result, err := reconciler.Sweep(context.Background()) + require.NoError(t, err) + require.Equal(t, EmbeddingsReconcileQueueName, repo.depthQueue) + require.Equal(t, "taxonomy:model", repo.listModel) + require.Equal(t, 2, repo.listLimit, "only the available target-depth room may be queried") + assert.WithinDuration(t, started.Add(-retryAfter), repo.retryBefore, time.Second) + require.Len(t, inserter.params, 2) + assert.Equal(t, EmbeddingReconcileResult{Found: 2, Enqueued: 2, Depth: 98}, result) + + for index, param := range inserter.params { + args, ok := param.Args.(FeedbackEmbeddingArgs) + require.True(t, ok) + assert.Equal(t, ids[index], args.FeedbackRecordID) + assert.Equal(t, "taxonomy:model", args.Model) + assert.Equal(t, models.EmbeddingInputKindTaxonomyTranslated, args.InputKind) + assert.Equal(t, "reconcile", args.ValueTextHash) + require.NotNil(t, param.InsertOpts) + assert.Equal(t, EmbeddingsReconcileQueueName, param.InsertOpts.Queue) + assert.Equal(t, 4, param.InsertOpts.Priority) + assert.Equal(t, 5, param.InsertOpts.MaxAttempts) + assert.True(t, param.InsertOpts.UniqueOpts.ByArgs) + assert.ElementsMatch(t, []rivertype.JobState{ + rivertype.JobStateAvailable, + rivertype.JobStatePending, + rivertype.JobStateRetryable, + rivertype.JobStateRunning, + rivertype.JobStateScheduled, + }, param.InsertOpts.UniqueOpts.ByState) + } +} + +func TestEmbeddingReconcileSweepStopsAtTarget(t *testing.T) { + repo := &stubEmbeddingReconcileRepository{depth: 100} + inserter := &recordingBatchInserter{} + reconciler := NewEmbeddingReconcileService(repo, "taxonomy:model", 100, 5, 15*time.Minute) + reconciler.SetInserter(inserter) + + result, err := reconciler.Sweep(context.Background()) + require.NoError(t, err) + assert.True(t, result.AtTarget) + assert.Equal(t, 100, result.Depth) + assert.Zero(t, repo.listLimit, "a full lane must not scan the record backlog") + assert.Empty(t, inserter.params) +} + +func TestEmbeddingReconcileSweepCountsUniqueSkipsTruthfully(t *testing.T) { + repo := &stubEmbeddingReconcileRepository{ + ids: []uuid.UUID{uuid.Must(uuid.NewV7()), uuid.Must(uuid.NewV7())}, + } + inserter := &recordingBatchInserter{results: []*rivertype.JobInsertResult{ + {}, + {UniqueSkippedAsDuplicate: true}, + }} + reconciler := NewEmbeddingReconcileService(repo, "taxonomy:model", 2, 5, 15*time.Minute) + reconciler.SetInserter(inserter) + + result, err := reconciler.Sweep(context.Background()) + require.NoError(t, err) + assert.Equal(t, 2, result.Found) + assert.Equal(t, 1, result.Enqueued) +} + +func TestEmbeddingReconcileSweepPropagatesBoundedFailures(t *testing.T) { + t.Run("inserter is required", func(t *testing.T) { + reconciler := NewEmbeddingReconcileService( + &stubEmbeddingReconcileRepository{}, "taxonomy:model", 100, 5, 15*time.Minute) + _, err := reconciler.Sweep(context.Background()) + require.ErrorIs(t, err, ErrEmbeddingReconcileInserterUnset) + }) + + t.Run("depth read", func(t *testing.T) { + reconciler := NewEmbeddingReconcileService( + &stubEmbeddingReconcileRepository{depthErr: errors.New("db unavailable")}, + "taxonomy:model", 100, 5, 15*time.Minute) + reconciler.SetInserter(&recordingBatchInserter{}) + _, err := reconciler.Sweep(context.Background()) + require.ErrorContains(t, err, "read repair queue depth") + }) + + t.Run("record scan", func(t *testing.T) { + reconciler := NewEmbeddingReconcileService( + &stubEmbeddingReconcileRepository{listErr: errors.New("db unavailable")}, + "taxonomy:model", 100, 5, 15*time.Minute) + reconciler.SetInserter(&recordingBatchInserter{}) + _, err := reconciler.Sweep(context.Background()) + require.ErrorContains(t, err, "list pending taxonomy embeddings") + }) + + t.Run("batch insert", func(t *testing.T) { + reconciler := NewEmbeddingReconcileService( + &stubEmbeddingReconcileRepository{ids: []uuid.UUID{uuid.Must(uuid.NewV7())}}, + "taxonomy:model", 100, 5, 15*time.Minute) + reconciler.SetInserter(&recordingBatchInserter{err: errors.New("river unavailable")}) + _, err := reconciler.Sweep(context.Background()) + require.ErrorContains(t, err, "enqueue taxonomy embedding repairs") + }) +} diff --git a/internal/service/job_inserter.go b/internal/service/job_inserter.go index 8c56423d..cae2f2ed 100644 --- a/internal/service/job_inserter.go +++ b/internal/service/job_inserter.go @@ -13,3 +13,10 @@ import ( type RiverJobInserter interface { Insert(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error) } + +// RiverBatchInserter inserts a bounded group of River jobs. The concrete River client satisfies +// both inserter interfaces; keeping this seam small makes reconciliation tests independent of a +// database-backed queue. +type RiverBatchInserter interface { + InsertMany(ctx context.Context, params []river.InsertManyParams) ([]*rivertype.JobInsertResult, error) +} diff --git a/internal/service/job_kinds.go b/internal/service/job_kinds.go index 8c193296..cb7d2669 100644 --- a/internal/service/job_kinds.go +++ b/internal/service/job_kinds.go @@ -8,6 +8,9 @@ type JobKindSpec struct { Args river.JobArgs // Queue is the River queue the kind is inserted on. Queue string + // ReconcileQueue is an optional lower-concurrency lane for repaired historical work handled by + // the same worker kind. Keeping it separate prevents a large repair backlog delaying live data. + ReconcileQueue string } // Kind returns the River job kind this spec describes. @@ -28,12 +31,13 @@ func (s JobKindSpec) Kind() string { return s.Args.Kind() } func JobKindSpecs() []JobKindSpec { return []JobKindSpec{ {Args: WebhookDispatchArgs{}, Queue: river.QueueDefault}, - {Args: FeedbackEmbeddingArgs{}, Queue: EmbeddingsQueueName}, + {Args: FeedbackEmbeddingArgs{}, Queue: EmbeddingsQueueName, ReconcileQueue: EmbeddingsReconcileQueueName}, {Args: FeedbackTranslationArgs{}, Queue: TranslationsQueueName}, {Args: TenantTranslationBackfillArgs{}, Queue: TranslationBackfillsQueueName}, {Args: FeedbackSentimentArgs{}, Queue: SentimentsQueueName}, {Args: FeedbackEmotionsArgs{}, Queue: EmotionsQueueName}, {Args: FeedbackRecordsPurgeArgs{}, Queue: FeedbackRecordsPurgeQueueName}, + {Args: EmbeddingReconcileArgs{}, Queue: EmbeddingReconcileQueueName}, } } @@ -50,13 +54,22 @@ func distinctQueues(specs []JobKindSpec) []string { seen := make(map[string]struct{}, len(specs)) queues := make([]string, 0, len(specs)) - for _, spec := range specs { - if _, ok := seen[spec.Queue]; ok { - continue + add := func(queue string) { + if queue == "" { + return + } + + if _, ok := seen[queue]; ok { + return } - seen[spec.Queue] = struct{}{} - queues = append(queues, spec.Queue) + seen[queue] = struct{}{} + queues = append(queues, queue) + } + + for _, spec := range specs { + add(spec.Queue) + add(spec.ReconcileQueue) } return queues diff --git a/internal/service/job_kinds_test.go b/internal/service/job_kinds_test.go index 1e268b38..02b19168 100644 --- a/internal/service/job_kinds_test.go +++ b/internal/service/job_kinds_test.go @@ -18,6 +18,7 @@ func TestJobKindSpecs(t *testing.T) { "feedback_sentiment": SentimentsQueueName, "feedback_emotions": EmotionsQueueName, "feedback_records_purge": FeedbackRecordsPurgeQueueName, + "embedding_reconcile": EmbeddingReconcileQueueName, } specs := JobKindSpecs() @@ -27,6 +28,12 @@ func TestJobKindSpecs(t *testing.T) { wantQueue, ok := want[spec.Kind()] require.True(t, ok, "unexpected job kind %q", spec.Kind()) require.Equal(t, wantQueue, spec.Queue, "kind %q is on the wrong queue", spec.Kind()) + + if spec.Kind() == "feedback_embedding" { + require.Equal(t, EmbeddingsReconcileQueueName, spec.ReconcileQueue) + } else { + require.Empty(t, spec.ReconcileQueue) + } } } @@ -34,11 +41,13 @@ func TestJobQueueNames(t *testing.T) { require.Equal(t, []string{ river.QueueDefault, EmbeddingsQueueName, + EmbeddingsReconcileQueueName, TranslationsQueueName, TranslationBackfillsQueueName, SentimentsQueueName, EmotionsQueueName, FeedbackRecordsPurgeQueueName, + EmbeddingReconcileQueueName, }, JobQueueNames()) } diff --git a/internal/service/webhook_provider.go b/internal/service/webhook_provider.go index 8e40167e..2de16510 100644 --- a/internal/service/webhook_provider.go +++ b/internal/service/webhook_provider.go @@ -8,17 +8,11 @@ import ( "github.com/google/uuid" "github.com/riverqueue/river" - "github.com/riverqueue/river/rivertype" "github.com/formbricks/hub/internal/models" "github.com/formbricks/hub/internal/observability" ) -// WebhookDispatchInserter inserts webhook_dispatch jobs in batch (e.g. River client). -type WebhookDispatchInserter interface { - InsertMany(ctx context.Context, params []river.InsertManyParams) ([]*rivertype.JobInsertResult, error) -} - // WebhookProviderRepository lists tenant-scoped webhooks eligible for event fan-out. type WebhookProviderRepository interface { ListEnabledForEventTypeAndTenant(ctx context.Context, eventType string, tenantID *string) ([]models.Webhook, error) @@ -27,7 +21,7 @@ type WebhookProviderRepository interface { // WebhookProvider implements eventPublisher by enqueueing one River job per (event, webhook). type WebhookProvider struct { repo WebhookProviderRepository - inserter WebhookDispatchInserter + inserter RiverBatchInserter maxAttempts int maxFanOut int enqueueMaxRetries int @@ -41,7 +35,7 @@ type WebhookProvider struct { // enqueueMaxRetries, enqueueInitialBackoff, enqueueMaxBackoff configure retries when InsertMany fails (transient River/DB errors). // metrics may be nil when metrics are disabled. func NewWebhookProvider( - inserter WebhookDispatchInserter, repo WebhookProviderRepository, + inserter RiverBatchInserter, repo WebhookProviderRepository, maxAttempts, maxFanOut int, enqueueMaxRetries int, enqueueInitialBackoff, enqueueMaxBackoff time.Duration, metrics observability.WebhookMetrics, diff --git a/internal/workers/embedding_reconcile.go b/internal/workers/embedding_reconcile.go new file mode 100644 index 00000000..bc9c8031 --- /dev/null +++ b/internal/workers/embedding_reconcile.go @@ -0,0 +1,78 @@ +package workers + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/riverqueue/river" + + "github.com/formbricks/hub/internal/observability" + "github.com/formbricks/hub/internal/service" +) + +const embeddingReconcileTimeout = 2 * time.Minute + +// EmbeddingReconcileSweeper is the service boundary used by the periodic worker. +type EmbeddingReconcileSweeper interface { + Sweep(ctx context.Context) (service.EmbeddingReconcileResult, error) +} + +// EmbeddingReconcileWorker runs one bounded, level-triggered taxonomy embedding sweep. +type EmbeddingReconcileWorker struct { + river.WorkerDefaults[service.EmbeddingReconcileArgs] + + sweeper EmbeddingReconcileSweeper + metrics observability.EmbeddingMetrics +} + +// NewEmbeddingReconcileWorker creates the periodic sweep worker. +func NewEmbeddingReconcileWorker( + sweeper EmbeddingReconcileSweeper, + metrics observability.EmbeddingMetrics, +) *EmbeddingReconcileWorker { + return &EmbeddingReconcileWorker{sweeper: sweeper, metrics: metrics} +} + +// Timeout gives the database scan and bounded insert batch its own deadline rather than inheriting +// River's unrelated global job timeout. +func (w *EmbeddingReconcileWorker) Timeout(*river.Job[service.EmbeddingReconcileArgs]) time.Duration { + return embeddingReconcileTimeout +} + +// Work executes one sweep. A failure is returned for observability, but scheduled jobs use one +// attempt because the next interval is the retry and sees the same level-triggered backlog. +func (w *EmbeddingReconcileWorker) Work( + ctx context.Context, + _ *river.Job[service.EmbeddingReconcileArgs], +) error { + ctx, cancel := context.WithTimeout(ctx, embeddingReconcileTimeout) + defer cancel() + + result, err := w.sweeper.Sweep(ctx) + if err != nil { + if w.metrics != nil { + w.metrics.RecordWorkerError(ctx, "reconcile_failed") + } + + slog.ErrorContext(ctx, "embedding reconcile: sweep failed", "error", err) + + return fmt.Errorf("embedding reconcile: %w", err) + } + + if w.metrics != nil && result.Enqueued > 0 { + w.metrics.RecordJobsEnqueued(ctx, int64(result.Enqueued)) + } + + slog.InfoContext(ctx, "embedding reconcile: sweep complete", + "found", result.Found, + "enqueued", result.Enqueued, + "queue_depth_before", result.Depth, + "at_target_depth", result.AtTarget, + ) + + return nil +} + +var _ river.Worker[service.EmbeddingReconcileArgs] = (*EmbeddingReconcileWorker)(nil) diff --git a/internal/workers/embedding_reconcile_test.go b/internal/workers/embedding_reconcile_test.go new file mode 100644 index 00000000..7073b5ef --- /dev/null +++ b/internal/workers/embedding_reconcile_test.go @@ -0,0 +1,54 @@ +package workers + +import ( + "context" + "errors" + "testing" + + "github.com/riverqueue/river" + "github.com/riverqueue/river/rivertype" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/service" +) + +type stubEmbeddingReconcileWorkerSweeper struct { + result service.EmbeddingReconcileResult + err error +} + +func (s stubEmbeddingReconcileWorkerSweeper) Sweep( + context.Context, +) (service.EmbeddingReconcileResult, error) { + return s.result, s.err +} + +func embeddingReconcileJob() *river.Job[service.EmbeddingReconcileArgs] { + return &river.Job[service.EmbeddingReconcileArgs]{ + JobRow: &rivertype.JobRow{Attempt: 1, MaxAttempts: 1}, + } +} + +func TestEmbeddingReconcileWorkerRecordsEnqueuedJobs(t *testing.T) { + metrics := newCountingEmbeddingMetrics() + worker := NewEmbeddingReconcileWorker(stubEmbeddingReconcileWorkerSweeper{ + result: service.EmbeddingReconcileResult{Found: 5, Enqueued: 3}, + }, metrics) + + require.NoError(t, worker.Work(context.Background(), embeddingReconcileJob())) + assert.Equal(t, int64(3), metrics.jobsEnqueued) + assert.Zero(t, metrics.workerErr["reconcile_failed"]) +} + +func TestEmbeddingReconcileWorkerRecordsSweepFailure(t *testing.T) { + metrics := newCountingEmbeddingMetrics() + worker := NewEmbeddingReconcileWorker(stubEmbeddingReconcileWorkerSweeper{ + err: errors.New("database unavailable"), + }, metrics) + + err := worker.Work(context.Background(), embeddingReconcileJob()) + require.ErrorContains(t, err, "embedding reconcile") + assert.Equal(t, 1, metrics.workerErr["reconcile_failed"]) + assert.Zero(t, metrics.jobsEnqueued) +} diff --git a/internal/workers/enrichment_worker.go b/internal/workers/enrichment_worker.go index a53673fe..1c18b50e 100644 --- a/internal/workers/enrichment_worker.go +++ b/internal/workers/enrichment_worker.go @@ -52,8 +52,8 @@ type FailureRecorder interface { // lose when nothing is healthy. const enrichmentFailureWriteTimeout = time.Second -// enrichmentJobTimeout limits one enrichment job run; shared by all four pipelines (LLM and -// embedding calls dominate, and every provider client keeps its own shorter HTTP timeout). +// enrichmentJobTimeout limits one structured enrichment job run. Embeddings use their own +// configurable deadline because CPU inference can legitimately take longer under bounded load. const enrichmentJobTimeout = 30 * time.Second // enrichmentWorkerConfig configures an enrichmentWorker: how to read the record and extract its id, diff --git a/internal/workers/feedback_embedding.go b/internal/workers/feedback_embedding.go index 12869136..0338f781 100644 --- a/internal/workers/feedback_embedding.go +++ b/internal/workers/feedback_embedding.go @@ -25,8 +25,13 @@ type FeedbackEmbeddingWorker struct { embeddingClient service.EmbeddingClient docPrefix string // model-specific prefix for document embedding metrics observability.EmbeddingMetrics + jobTimeout time.Duration + failures FailureRecorder + failureMetrics observability.EnrichmentFailureMetrics } +const defaultEmbeddingJobTimeout = 60 * time.Second + // feedbackEmbeddingService is the minimal interface needed by the worker. type feedbackEmbeddingService interface { GetFeedbackRecord(ctx context.Context, id uuid.UUID) (*models.FeedbackRecord, error) @@ -45,17 +50,42 @@ func NewFeedbackEmbeddingWorker( docPrefix string, metrics observability.EmbeddingMetrics, ) *FeedbackEmbeddingWorker { + return NewFeedbackEmbeddingWorkerWithOptions( + embeddingService, embeddingClient, docPrefix, metrics, + defaultEmbeddingJobTimeout, nil, nil, + ) +} + +// NewFeedbackEmbeddingWorkerWithOptions creates an embedding worker with an explicit job deadline +// and durable failure recording. The simple constructor remains for tools/tests that do not need +// deployment wiring. +func NewFeedbackEmbeddingWorkerWithOptions( + embeddingService feedbackEmbeddingService, + embeddingClient service.EmbeddingClient, + docPrefix string, + metrics observability.EmbeddingMetrics, + jobTimeout time.Duration, + failures FailureRecorder, + failureMetrics observability.EnrichmentFailureMetrics, +) *FeedbackEmbeddingWorker { + if jobTimeout <= 0 { + jobTimeout = defaultEmbeddingJobTimeout + } + return &FeedbackEmbeddingWorker{ embeddingService: embeddingService, embeddingClient: embeddingClient, docPrefix: docPrefix, metrics: metrics, + jobTimeout: jobTimeout, + failures: failures, + failureMetrics: failureMetrics, } } // Timeout limits how long a single embedding job can run. func (w *FeedbackEmbeddingWorker) Timeout(*river.Job[service.FeedbackEmbeddingArgs]) time.Duration { - return enrichmentJobTimeout + return w.jobTimeout } // Work loads the record, generates or clears the embedding, and persists it. @@ -122,14 +152,17 @@ func (w *FeedbackEmbeddingWorker) Work(ctx context.Context, job *river.Job[servi embedding, err := w.embeddingClient.CreateEmbedding(ctx, text) if err != nil { - return w.handleEmbedError(ctx, err, job, log, start) + return w.handleEmbedError(ctx, err, job, record, log, start) } err = w.embeddingService.SetEmbedding(ctx, args.FeedbackRecordID, args.Model, embedding, stillCurrent) if err != nil { isLastAttempt := job.Attempt >= job.MaxAttempts - return w.handleSetEmbeddingError(ctx, err, log, start, isLastAttempt, "set feedback record embedding") + return w.handleSetEmbeddingError( + ctx, err, log, start, record, job.Args.Model, inputKind, job.Attempt, isLastAttempt, + "set feedback record embedding", + ) } log.Info("embedding: stored") @@ -147,7 +180,12 @@ func (w *FeedbackEmbeddingWorker) Work(ctx context.Context, job *river.Job[servi // jobs than the provider's rate limit and would otherwise mass-discard them as failed_final // (mirrors the classify workers) — while anything else retries, failing on the last attempt. func (w *FeedbackEmbeddingWorker) handleEmbedError( - ctx context.Context, err error, job *river.Job[service.FeedbackEmbeddingArgs], log *slog.Logger, start time.Time, + ctx context.Context, + err error, + job *river.Job[service.FeedbackEmbeddingArgs], + record *models.FeedbackRecord, + log *slog.Logger, + start time.Time, ) error { if delay, ok := rateLimitSnoozeDelay(err, job.CreatedAt); ok { if w.metrics != nil { @@ -164,6 +202,33 @@ func (w *FeedbackEmbeddingWorker) handleEmbedError( return river.JobSnooze(delay) } + inputKind := models.NormalizeEmbeddingInputKind(job.Args.InputKind) + + if reason, terminal := huberrors.TerminalReasonOf(err); terminal { + if w.metrics != nil { + w.metrics.RecordWorkerError(ctx, "embedding_api_failed") + w.metrics.RecordEmbeddingOutcome(ctx, "failed_final") + w.metrics.RecordEmbeddingDuration(ctx, time.Since(start), "failed_final") + } + + if w.failureMetrics != nil && + inputKind == models.EmbeddingInputKindTaxonomyTranslated { + w.failureMetrics.RecordTerminalFailure( + ctx, models.EnrichmentNameTaxonomyEmbedding, string(reason)) + } + + w.markTaxonomyEmbeddingFailed( + ctx, log, record, inputKind, job.Args.Model, job.Attempt, true, string(reason)) + log.Error("embedding: provider failed permanently for this record, not retrying", + "reason", string(reason), + "attempt", job.Attempt, + "error", err, + ) + + //nolint:wrapcheck // River must see JobCancel directly to suppress remaining attempts. + return river.JobCancel(fmt.Errorf("embedding API (terminal, %s): %w", reason, err)) + } + isLastAttempt := job.Attempt >= job.MaxAttempts if w.metrics != nil { @@ -179,6 +244,10 @@ func (w *FeedbackEmbeddingWorker) handleEmbedError( } if isLastAttempt { + w.markTaxonomyEmbeddingFailed( + ctx, log, record, inputKind, job.Args.Model, job.Attempt, false, + models.EnrichmentFailureReasonProviderError, + ) log.Error("embedding: API failed (final attempt)", "error", err, ) @@ -200,6 +269,10 @@ func (w *FeedbackEmbeddingWorker) handleSetEmbeddingError( err error, log *slog.Logger, start time.Time, + record *models.FeedbackRecord, + model string, + inputKind models.EmbeddingInputKind, + attempt int, isLastAttempt bool, action string, ) error { @@ -255,6 +328,13 @@ func (w *FeedbackEmbeddingWorker) handleSetEmbeddingError( w.metrics.RecordEmbeddingDuration(ctx, time.Since(start), outcome) } + if isLastAttempt { + w.markTaxonomyEmbeddingFailed( + ctx, log, record, inputKind, model, attempt, false, + models.EnrichmentFailureReasonWriteFailed, + ) + } + log.Error("embedding: "+action+" failed", "final_attempt", isLastAttempt, "error", err, @@ -264,6 +344,51 @@ func (w *FeedbackEmbeddingWorker) handleSetEmbeddingError( } } +// markTaxonomyEmbeddingFailed writes failure bookkeeping only for the translated taxonomy model. +// Raw search embeddings are intentionally outside the taxonomy progress/reconcile contract. +func (w *FeedbackEmbeddingWorker) markTaxonomyEmbeddingFailed( + ctx context.Context, + log *slog.Logger, + record *models.FeedbackRecord, + inputKind models.EmbeddingInputKind, + model string, + attempts int, + terminal bool, + reason string, +) { + if w.failures == nil || record == nil || + models.NormalizeEmbeddingInputKind(inputKind) != models.EmbeddingInputKindTaxonomyTranslated { + return + } + + writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), enrichmentFailureWriteTimeout) + defer cancel() + + err := w.failures.RecordFailure(writeCtx, models.EnrichmentFailure{ + FeedbackRecordID: record.ID, + TenantID: record.TenantID, + Enrichment: models.EnrichmentNameTaxonomyEmbedding, + Terminal: terminal, + Reason: reason, + Attempts: attempts, + ContextKey: model, + SourceUpdatedAt: &record.UpdatedAt, + }) + if err == nil || errors.Is(err, huberrors.ErrNotFound) || errors.Is(err, huberrors.ErrTenantWriteConflict) { + return + } + + if w.metrics != nil { + w.metrics.RecordWorkerError(ctx, "failure_marker_write_failed") + } + + log.Error("embedding: could not record taxonomy embedding failure", + "terminal", terminal, + "reason", reason, + "error", err, + ) +} + // handleEmptyText clears the embedding for text fields when value_text is empty, or records skip for non-text. func (w *FeedbackEmbeddingWorker) handleEmptyText( ctx context.Context, @@ -280,7 +405,18 @@ func (w *FeedbackEmbeddingWorker) handleEmptyText( if err != nil { isLastAttempt := job.Attempt >= job.MaxAttempts - return w.handleSetEmbeddingError(ctx, err, log, start, isLastAttempt, "clear feedback record embedding") + return w.handleSetEmbeddingError( + ctx, + err, + log, + start, + record, + job.Args.Model, + models.NormalizeEmbeddingInputKind(job.Args.InputKind), + job.Attempt, + isLastAttempt, + "clear feedback record embedding", + ) } if w.metrics != nil { diff --git a/internal/workers/feedback_embedding_test.go b/internal/workers/feedback_embedding_test.go index 20258052..6c73d376 100644 --- a/internal/workers/feedback_embedding_test.go +++ b/internal/workers/feedback_embedding_test.go @@ -9,6 +9,8 @@ import ( "github.com/google/uuid" "github.com/riverqueue/river" "github.com/riverqueue/river/rivertype" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/formbricks/hub/internal/huberrors" "github.com/formbricks/hub/internal/models" @@ -18,15 +20,18 @@ import ( // countingEmbeddingMetrics records outcome/worker-error counts for assertions. type countingEmbeddingMetrics struct { - outcomes map[string]int - workerErr map[string]int + outcomes map[string]int + workerErr map[string]int + jobsEnqueued int64 } func newCountingEmbeddingMetrics() *countingEmbeddingMetrics { return &countingEmbeddingMetrics{outcomes: map[string]int{}, workerErr: map[string]int{}} } -func (m *countingEmbeddingMetrics) RecordJobsEnqueued(context.Context, int64) {} +func (m *countingEmbeddingMetrics) RecordJobsEnqueued(_ context.Context, count int64) { + m.jobsEnqueued += count +} func (m *countingEmbeddingMetrics) RecordProviderError(context.Context, string) {} func (m *countingEmbeddingMetrics) RecordEmbeddingOutcome(_ context.Context, status string) { @@ -44,6 +49,20 @@ func (m *countingEmbeddingMetrics) AddEmbeddingBatchInFlight(context.Context, in var _ observability.EmbeddingMetrics = (*countingEmbeddingMetrics)(nil) +type countingFailureMetrics struct { + terminal int +} + +func (m *countingFailureMetrics) RecordTerminalFailure(context.Context, string, string) { + m.terminal++ +} + +func (m *countingFailureMetrics) SetFailedRecords(string, bool, int64) {} + +func (m *countingFailureMetrics) ClearFailedRecords() {} + +var _ observability.EnrichmentFailureMetrics = (*countingFailureMetrics)(nil) + type mockEmbeddingService struct { record *models.FeedbackRecord getErr error @@ -110,6 +129,99 @@ func translatedTextRecord(valueText, valueTextTranslated string) *models.Feedbac return record } +func taxonomyEmbeddingJob(attempt int) *river.Job[service.FeedbackEmbeddingArgs] { + job := embeddingJob() + job.Attempt = attempt + job.MaxAttempts = 5 + job.Args.InputKind = models.EmbeddingInputKindTaxonomyTranslated + + return job +} + +func TestFeedbackEmbeddingWorkerUsesConfiguredTimeout(t *testing.T) { + worker := NewFeedbackEmbeddingWorkerWithOptions( + &mockEmbeddingService{}, &mockEmbeddingClient{}, "", nil, 75*time.Second, nil, nil) + + assert.Equal(t, 75*time.Second, worker.Timeout(embeddingJob())) +} + +func TestFeedbackEmbeddingWorkerRecordsTaxonomyFailures(t *testing.T) { + record := translatedTextRecord("Bonjour", "Hello") + record.ID = uuid.Must(uuid.NewV7()) + record.TenantID = "tenant-embedding-failure" + + t.Run("transient provider failure records only after final attempt", func(t *testing.T) { + failures := &recordingFailureRecorder{} + worker := NewFeedbackEmbeddingWorkerWithOptions( + &mockEmbeddingService{record: record}, + &mockEmbeddingClient{err: errors.New("provider unavailable")}, + "", nil, time.Minute, failures, nil) + + require.Error(t, worker.Work(context.Background(), taxonomyEmbeddingJob(1))) + assert.Empty(t, failures.calls) + + require.Error(t, worker.Work(context.Background(), taxonomyEmbeddingJob(5))) + require.Len(t, failures.calls, 1) + failure := failures.calls[0] + assert.Equal(t, record.ID, failure.FeedbackRecordID) + assert.Equal(t, record.TenantID, failure.TenantID) + assert.Equal(t, models.EnrichmentNameTaxonomyEmbedding, failure.Enrichment) + assert.False(t, failure.Terminal) + assert.Equal(t, models.EnrichmentFailureReasonProviderError, failure.Reason) + assert.Equal(t, 5, failure.Attempts) + assert.Equal(t, "test-model", failure.ContextKey) + require.NotNil(t, failure.SourceUpdatedAt) + assert.Equal(t, record.UpdatedAt, *failure.SourceUpdatedAt) + }) + + t.Run("terminal provider failure cancels immediately", func(t *testing.T) { + failures := &recordingFailureRecorder{} + worker := NewFeedbackEmbeddingWorkerWithOptions( + &mockEmbeddingService{record: record}, + &mockEmbeddingClient{err: huberrors.NewTerminalProviderError( + huberrors.TerminalReasonContentFilter, errors.New("blocked"))}, + "", nil, time.Minute, failures, nil) + + err := worker.Work(context.Background(), taxonomyEmbeddingJob(1)) + + var cancelErr *river.JobCancelError + require.ErrorAs(t, err, &cancelErr) + require.Len(t, failures.calls, 1) + assert.True(t, failures.calls[0].Terminal) + assert.Equal(t, string(huberrors.TerminalReasonContentFilter), failures.calls[0].Reason) + assert.Equal(t, 1, failures.calls[0].Attempts) + }) + + t.Run("final write failure is distinguished", func(t *testing.T) { + failures := &recordingFailureRecorder{} + worker := NewFeedbackEmbeddingWorkerWithOptions( + &mockEmbeddingService{record: record, setErr: errors.New("database unavailable")}, + &mockEmbeddingClient{embedding: []float32{0.1}}, + "", nil, time.Minute, failures, nil) + + require.Error(t, worker.Work(context.Background(), taxonomyEmbeddingJob(5))) + require.Len(t, failures.calls, 1) + assert.False(t, failures.calls[0].Terminal) + assert.Equal(t, models.EnrichmentFailureReasonWriteFailed, failures.calls[0].Reason) + }) + + t.Run("raw embedding failures never enter taxonomy state", func(t *testing.T) { + failures := &recordingFailureRecorder{} + failureMetrics := &countingFailureMetrics{} + worker := NewFeedbackEmbeddingWorkerWithOptions( + &mockEmbeddingService{record: record}, + &mockEmbeddingClient{err: huberrors.NewTerminalProviderError( + huberrors.TerminalReasonContentFilter, errors.New("blocked"))}, + "", nil, time.Minute, failures, failureMetrics) + job := embeddingJob() + + var cancelErr *river.JobCancelError + require.ErrorAs(t, worker.Work(context.Background(), job), &cancelErr) + assert.Empty(t, failures.calls) + assert.Zero(t, failureMetrics.terminal) + }) +} + func TestFeedbackEmbeddingWorker_GetNotFoundRecordsSkipped(t *testing.T) { metrics := newCountingEmbeddingMetrics() svc := &mockEmbeddingService{getErr: huberrors.NewNotFoundError("feedback record", "gone")} diff --git a/internal/workers/wiring.go b/internal/workers/wiring.go index f23d514c..b481eb3d 100644 --- a/internal/workers/wiring.go +++ b/internal/workers/wiring.go @@ -25,6 +25,8 @@ type RiverDeps struct { EmbeddingClient service.EmbeddingClient EmbeddingDocPrefix string EmbeddingMetrics observability.EmbeddingMetrics + // EmbeddingReconcileSweeper is non-nil only when automatic taxonomy embedding repair is enabled. + EmbeddingReconcileSweeper EmbeddingReconcileSweeper // Translation worker (optional; if TranslationClient is nil, translation worker is not registered) TranslationService translationWorkerService @@ -94,10 +96,30 @@ func NewRiverWorkersAndQueues( } if deps.EmbeddingClient != nil { - embeddingWorker := NewFeedbackEmbeddingWorker(deps.EmbeddingService, deps.EmbeddingClient, deps.EmbeddingDocPrefix, deps.EmbeddingMetrics) + embeddingWorker := NewFeedbackEmbeddingWorkerWithOptions( + deps.EmbeddingService, + deps.EmbeddingClient, + deps.EmbeddingDocPrefix, + deps.EmbeddingMetrics, + cfg.Embedding.JobTimeout.Duration(), + deps.Failures, + deps.FailureMetrics, + ) river.AddWorker(workers, embeddingWorker) queues[service.EmbeddingsQueueName] = river.QueueConfig{MaxWorkers: maxEmbedding} + // Keep draining repairs already queued by an earlier configuration even when the + // periodic reconciler is later disabled. Both queues use the same embedding worker. + queues[service.EmbeddingsReconcileQueueName] = river.QueueConfig{ + MaxWorkers: cfg.Embedding.ReconcileMaxConcurrent, + } + + if deps.EmbeddingReconcileSweeper != nil { + river.AddWorker(workers, NewEmbeddingReconcileWorker( + deps.EmbeddingReconcileSweeper, deps.EmbeddingMetrics)) + + queues[service.EmbeddingReconcileQueueName] = river.QueueConfig{MaxWorkers: 1} + } } if deps.TranslationClient != nil { diff --git a/internal/workers/wiring_test.go b/internal/workers/wiring_test.go index e8d00153..3f26bb60 100644 --- a/internal/workers/wiring_test.go +++ b/internal/workers/wiring_test.go @@ -36,6 +36,12 @@ func (stubFeedbackRecordsPurgeService) Purge( return &models.FeedbackRecordsPurgeCounts{}, nil } +type stubEmbeddingReconcileSweeper struct{} + +func (stubEmbeddingReconcileSweeper) Sweep(context.Context) (service.EmbeddingReconcileResult, error) { + return service.EmbeddingReconcileResult{}, nil +} + // kindProbe re-registers a job kind to observe whether it is already registered. // river.AddWorkerSafely errors only on a duplicate kind, and *river.Workers exposes no way to // enumerate its kinds (workersMap is unexported with no accessor), so this is the only way to assert @@ -68,9 +74,10 @@ func fullRiverDeps() RiverDeps { WebhookSender: &mockSender{}, WebhookMetrics: newCountingWebhookMetrics(), - EmbeddingService: &mockEmbeddingService{}, - EmbeddingClient: &mockEmbeddingClient{}, - EmbeddingMetrics: &countingEmbeddingMetrics{}, + EmbeddingService: &mockEmbeddingService{}, + EmbeddingClient: &mockEmbeddingClient{}, + EmbeddingMetrics: &countingEmbeddingMetrics{}, + EmbeddingReconcileSweeper: stubEmbeddingReconcileSweeper{}, TranslationService: &mockTranslationWorkerService{}, TranslationClient: &stubTranslationClient{}, @@ -113,8 +120,9 @@ func TestNewRiverWorkersAndQueuesCoversEveryJobKind(t *testing.T) { assertKindRegistered[service.FeedbackSentimentArgs](t, workerBundle, true) assertKindRegistered[service.FeedbackEmotionsArgs](t, workerBundle, true) assertKindRegistered[service.FeedbackRecordsPurgeArgs](t, workerBundle, true) + assertKindRegistered[service.EmbeddingReconcileArgs](t, workerBundle, true) - const probedKinds = 7 + const probedKinds = 8 if got := len(service.JobKindSpecs()); got != probedKinds { t.Fatalf("JobKindSpecs has %d kinds but %d are probed above — add a probe for the new kind "+ "and register a worker for it in NewRiverWorkersAndQueues", got, probedKinds) @@ -157,6 +165,30 @@ func TestNewRiverWorkersAndQueuesWithoutOptionalClients(t *testing.T) { assertKindRegistered[service.FeedbackEmotionsArgs](t, workerBundle, false) } +// TestNewRiverWorkersAndQueuesDrainsExistingRepairsWhenSweeperDisabled ensures disabling future +// sweeps cannot strand repair jobs that were queued by an earlier worker configuration. +func TestNewRiverWorkersAndQueuesDrainsExistingRepairsWhenSweeperDisabled(t *testing.T) { + cfg := &config.Config{} + cfg.Embedding.MaxConcurrent = 3 + cfg.Embedding.ReconcileMaxConcurrent = 1 + deps := fullRiverDeps() + deps.EmbeddingReconcileSweeper = nil + + workerBundle, queues := NewRiverWorkersAndQueues(cfg, deps) + + if got := queues[service.EmbeddingsReconcileQueueName].MaxWorkers; got != 1 { + t.Fatalf("queue %q MaxWorkers = %d, want 1", service.EmbeddingsReconcileQueueName, got) + } + + _, sweepQueueRegistered := queues[service.EmbeddingReconcileQueueName] + if sweepQueueRegistered { + t.Fatalf("queue %q registered with sweeper disabled, want absent", service.EmbeddingReconcileQueueName) + } + + assertKindRegistered[service.FeedbackEmbeddingArgs](t, workerBundle, true) + assertKindRegistered[service.EmbeddingReconcileArgs](t, workerBundle, false) +} + // TestNewRiverWorkersAndQueuesUsesConfiguredConcurrency pins each queue to its own configured // concurrency. hub-worker is the only caller, so a queue silently picking up another enrichment's // limit — or a zero, which would stall it entirely — would otherwise go unnoticed. diff --git a/migrations/024_add_taxonomy_embedding_failures.sql b/migrations/024_add_taxonomy_embedding_failures.sql new file mode 100644 index 00000000..8462a11b --- /dev/null +++ b/migrations/024_add_taxonomy_embedding_failures.sql @@ -0,0 +1,38 @@ +-- +goose Up +-- Taxonomy-only embeddings use the same durable failure state as the other enrichments. The +-- existing primary key, tenant index, reason set, retention, and purge cascade all apply +-- unchanged. The two nullable context columns prevent a marker from an old model or record +-- revision suppressing a new attempt. +ALTER TABLE feedback_record_enrichment_failures + ADD COLUMN context_key TEXT, + ADD COLUMN source_updated_at TIMESTAMPTZ; + +ALTER TABLE feedback_record_enrichment_failures + DROP CONSTRAINT IF EXISTS feedback_record_enrichment_failures_enrichment_valid; + +ALTER TABLE feedback_record_enrichment_failures + ADD CONSTRAINT feedback_record_enrichment_failures_enrichment_valid CHECK ( + (enrichment IN ('translation', 'sentiment', 'emotions') + AND context_key IS NULL AND source_updated_at IS NULL) + OR + (enrichment = 'taxonomy_embedding' + AND NULLIF(btrim(context_key), '') IS NOT NULL AND source_updated_at IS NOT NULL) + ); + +-- +goose Down +-- Remove rows the old constraint cannot represent before restoring it. Failure markers are +-- advisory and can be reconstructed by a future attempt, so rollback remains safe. +DELETE FROM feedback_record_enrichment_failures +WHERE enrichment = 'taxonomy_embedding'; + +ALTER TABLE feedback_record_enrichment_failures + DROP CONSTRAINT IF EXISTS feedback_record_enrichment_failures_enrichment_valid; + +ALTER TABLE feedback_record_enrichment_failures + ADD CONSTRAINT feedback_record_enrichment_failures_enrichment_valid CHECK ( + enrichment IN ('translation', 'sentiment', 'emotions') + ); + +ALTER TABLE feedback_record_enrichment_failures + DROP COLUMN IF EXISTS source_updated_at, + DROP COLUMN IF EXISTS context_key; diff --git a/openapi.yaml b/openapi.yaml index 393faa55..37b293a1 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -3909,6 +3909,14 @@ components: type: integer format: int64 description: Number of those records that have an embedding. + embedding_failed_count: + type: integer + format: int64 + description: Number of missing taxonomy embeddings whose latest bounded attempts failed transiently and can be retried. + embedding_failed_terminal_count: + type: integer + format: int64 + description: Number of missing taxonomy embeddings rejected permanently for the current model and record revision. required: - tenant_id - source_type @@ -3916,6 +3924,8 @@ components: - field_id - record_count - embedding_count + - embedding_failed_count + - embedding_failed_terminal_count TaxonomyFieldsOutputBody: type: object additionalProperties: false diff --git a/tests/embedding_reconcile_test.go b/tests/embedding_reconcile_test.go new file mode 100644 index 00000000..c967d7f9 --- /dev/null +++ b/tests/embedding_reconcile_test.go @@ -0,0 +1,156 @@ +package tests + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/google/uuid" + "github.com/riverqueue/river" + "github.com/riverqueue/river/riverdriver/riverpgxv5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/formbricks/hub/internal/config" + "github.com/formbricks/hub/internal/huberrors" + "github.com/formbricks/hub/internal/models" + "github.com/formbricks/hub/internal/repository" + "github.com/formbricks/hub/internal/service" + "github.com/formbricks/hub/pkg/database" +) + +// TestListPendingTaxonomyEmbeddingIDs exercises the real repair-selection query against River and +// enrichment failure state. Each exclusion is part of the safety contract: omitting one either +// duplicates provider work or permanently abandons recoverable records. +func TestListPendingTaxonomyEmbeddingIDs(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() + + embeddingsRepo := repository.NewEmbeddingsRepository(db) + failuresRepo := repository.NewEnrichmentFailuresRepository(db) + riverClient, err := river.NewClient(riverpgxv5.New(db), &river.Config{}) + require.NoError(t, err) + + tenantID := "embedding-reconcile-" + uuid.NewString() + model := "taxonomy:reconcile-" + uuid.NewString() + + t.Cleanup(func() { + _, _ = db.Exec(ctx, `DELETE FROM river_job WHERE args->>'feedback_record_id' IN ( + SELECT id::text FROM feedback_records WHERE tenant_id = $1 + )`, tenantID) + _, _ = db.Exec(ctx, `DELETE FROM feedback_records WHERE tenant_id = $1`, tenantID) + }) + + seedValues := func(label, valueText, valueTextTranslated string) *models.FeedbackRecord { + t.Helper() + + var record models.FeedbackRecord + + err := db.QueryRow(ctx, ` + INSERT INTO feedback_records ( + source_type, source_id, field_id, field_label, field_type, + value_text, value_text_translated, tenant_id, submission_id, collected_at + ) + VALUES ('formbricks', $1, 'feedback', 'Feedback', 'text'::field_type_enum, + $2, $3, $4, $5, '-infinity'::timestamptz) + RETURNING id, tenant_id, updated_at`, + "source-"+label, valueText, valueTextTranslated, tenantID, "submission-"+uuid.NewString(), + ).Scan(&record.ID, &record.TenantID, &record.UpdatedAt) + require.NoError(t, err) + + return &record + } + seed := func(label string) *models.FeedbackRecord { + t.Helper() + + return seedValues(label, "raw "+label, "translated "+label) + } + + eligible := seed("eligible") + embedded := seed("already-embedded") + activeJob := seed("active-job") + terminalCurrent := seed("terminal-current") + terminalOldModel := seed("terminal-old-model") + terminalOldRevision := seed("terminal-old-revision") + transientFailure := seed("transient-failure") + transientRetryable := seed("transient-retryable") + + whitespaceOnly := make([]*models.FeedbackRecord, 0, 7) + for i, value := range []string{"\t", "\n", "\r\n", "\v", "\f", "\u00a0", "\u3000"} { + whitespaceOnly = append(whitespaceOnly, seedValues( + fmt.Sprintf("whitespace-%d", i), value, value, + )) + } + + vector := make([]float32, models.EmbeddingVectorDimensions) + vector[0] = 0.5 + require.NoError(t, embeddingsRepo.Upsert(ctx, embedded.ID, model, vector, nil)) + + _, err = riverClient.Insert(ctx, service.FeedbackEmbeddingArgs{ + FeedbackRecordID: activeJob.ID, + Model: model, + InputKind: models.EmbeddingInputKindTaxonomyTranslated, + ValueTextHash: "live-event", + }, &river.InsertOpts{Queue: service.EmbeddingsQueueName, MaxAttempts: 5}) + require.NoError(t, err) + + recordFailure := func(record *models.FeedbackRecord, contextKey string, terminal bool, reason string) { + t.Helper() + require.NoError(t, failuresRepo.RecordFailure(ctx, models.EnrichmentFailure{ + FeedbackRecordID: record.ID, + TenantID: tenantID, + Enrichment: models.EnrichmentNameTaxonomyEmbedding, + Terminal: terminal, + Reason: reason, + Attempts: 5, + ContextKey: contextKey, + SourceUpdatedAt: &record.UpdatedAt, + })) + } + + recordFailure(terminalCurrent, model, true, string(huberrors.TerminalReasonContentFilter)) + recordFailure(terminalOldModel, "taxonomy:retired-model", true, string(huberrors.TerminalReasonContentFilter)) + recordFailure(terminalOldRevision, model, true, string(huberrors.TerminalReasonContentFilter)) + recordFailure(transientFailure, model, false, models.EnrichmentFailureReasonProviderError) + recordFailure(transientRetryable, model, false, models.EnrichmentFailureReasonProviderError) + _, err = db.Exec(ctx, `UPDATE feedback_record_enrichment_failures + SET failed_at = NOW() - interval '1 hour' + WHERE feedback_record_id = $1 AND enrichment = 'taxonomy_embedding'`, transientRetryable.ID) + require.NoError(t, err) + + // Simulate an edit after the terminal failure. A marker tied to the old revision must not + // suppress the new content, even though it was written for the same record and model. + _, err = db.Exec(ctx, `UPDATE feedback_records SET updated_at = updated_at + interval '1 second' WHERE id = $1`, + terminalOldRevision.ID) + require.NoError(t, err) + + // This repository is shared by local integration runs, and a unique model makes every existing + // fixture look missing. The fixture timestamps sort first so the assertion does not depend on a + // database-wide inspection limit; bounded production top-up is asserted in the service unit test. + ids, err := embeddingsRepo.ListPendingTaxonomyEmbeddingIDs(ctx, model, time.Now().Add(-15*time.Minute), 1_000) + require.NoError(t, err) + + got := make(map[uuid.UUID]struct{}, len(ids)) + for _, id := range ids { + got[id] = struct{}{} + } + + for _, record := range []*models.FeedbackRecord{eligible, terminalOldModel, terminalOldRevision, transientRetryable} { + assert.Contains(t, got, record.ID, "recoverable record %s must be selected", record.ID) + } + + for _, record := range []*models.FeedbackRecord{embedded, activeJob, terminalCurrent, transientFailure} { + assert.NotContains(t, got, record.ID, "already handled record %s must not be selected", record.ID) + } + + for _, record := range whitespaceOnly { + assert.NotContains(t, got, record.ID, "whitespace-only record %s must not be selected", record.ID) + } +} diff --git a/tests/enrichment_failures_test.go b/tests/enrichment_failures_test.go index 714115e8..f9bf44c8 100644 --- a/tests/enrichment_failures_test.go +++ b/tests/enrichment_failures_test.go @@ -411,7 +411,7 @@ func TestCountFailedRecordsAggregateIsGated(t *testing.T) { return 0 } - before, err := statusRepo.CountFailedRecordsAggregate(ctx) + before, err := statusRepo.CountFailedRecordsAggregate(ctx, "") require.NoError(t, err) // A tenant that switched sentiment off. Its failures are history, not work. @@ -435,7 +435,7 @@ func TestCountFailedRecordsAggregateIsGated(t *testing.T) { liveRecord := seedEnrichmentRecord(t, frepo, live, models.FieldTypeText, "genuinely failed sentiment") insertFailureMarker(t, db, liveRecord.ID, live, "sentiment", false, "provider_error") - after, err := statusRepo.CountFailedRecordsAggregate(ctx) + after, err := statusRepo.CountFailedRecordsAggregate(ctx, "") require.NoError(t, err) delta := countFor("sentiment", false, after) - countFor("sentiment", false, before) @@ -450,6 +450,65 @@ func TestCountFailedRecordsAggregateIsGated(t *testing.T) { } } +func TestCountFailedRecordsAggregateIncludesOnlyCurrentTaxonomyEmbeddingFailures(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() + + failureRepo := repository.NewEnrichmentFailuresRepository(db) + feedbackRepo := repository.NewFeedbackRecordsRepository(db) + statusRepo := repository.NewEnrichmentStatusRepository(db) + embeddingsRepo := repository.NewEmbeddingsRepository(db) + tenantID := "aggregate-taxonomy-failure-" + uuid.NewString() + model := "taxonomy:aggregate-" + uuid.NewString() + + t.Cleanup(func() { + _, _ = db.Exec(ctx, `DELETE FROM feedback_records WHERE tenant_id = $1`, tenantID) + }) + + countFor := func(counts []repository.FailedRecordCount) int64 { + for _, count := range counts { + if count.Enrichment == models.EnrichmentNameTaxonomyEmbedding && !count.Terminal { + return count.Count + } + } + + return 0 + } + + before, err := statusRepo.CountFailedRecordsAggregate(ctx, model) + require.NoError(t, err) + + record := seedEnrichmentRecord(t, feedbackRepo, tenantID, models.FieldTypeText, "embedding provider timed out") + require.NoError(t, failureRepo.RecordFailure(ctx, models.EnrichmentFailure{ + FeedbackRecordID: record.ID, + TenantID: tenantID, + Enrichment: models.EnrichmentNameTaxonomyEmbedding, + Reason: models.EnrichmentFailureReasonProviderError, + Attempts: 5, + ContextKey: model, + SourceUpdatedAt: &record.UpdatedAt, + })) + + afterFailure, err := statusRepo.CountFailedRecordsAggregate(ctx, model) + require.NoError(t, err) + assert.Equal(t, int64(1), countFor(afterFailure)-countFor(before)) + + embedding := make([]float32, models.EmbeddingVectorDimensions) + embedding[0] = 0.25 + require.NoError(t, embeddingsRepo.Upsert(ctx, record.ID, model, embedding, nil)) + + afterSuccess, err := statusRepo.CountFailedRecordsAggregate(ctx, model) + require.NoError(t, err) + assert.Equal(t, countFor(before), countFor(afterSuccess), + "a stale marker must stop counting once the exact embedding exists") +} + func countRowsIn(ctx context.Context, t *testing.T, db *pgxpool.Pool, query string, args ...any) int64 { t.Helper() diff --git a/tests/enrichment_status_test.go b/tests/enrichment_status_test.go index 6c619c5d..4793a645 100644 --- a/tests/enrichment_status_test.go +++ b/tests/enrichment_status_test.go @@ -81,16 +81,19 @@ func TestCountEnrichmentStatus(t *testing.T) { // 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, "\v") // vertical tab is whitespace on every supported PG + mkRecord(tenant, models.FieldTypeText, "\u00a0\u3000") // Unicode whitespace mirrors Go TrimSpace mkRecord(tenant, models.FieldTypeText, "") // empty content + mkRecord(tenant, models.FieldTypeText, "v") // PG16 must not treat E'\v' as this literal letter 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(4), counts.SentimentEligible, "4 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(4), 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(4), counts.TranslationEligible, "all 4 have the effective target de-DE") assert.Equal(t, int64(1), counts.TranslationDone, "only doneAll's lang key matches; fr-FR is stale") }) @@ -320,6 +323,8 @@ func TestCountEnrichmentBacklogAggregateIfLeader(t *testing.T) { require.True(t, isLeader, "the first replica to poll becomes the leader") seedEnrichmentRecord(t, fallbackRepo, testTenantID("taxonomy-backlog-text"), models.FieldTypeText, "taxonomy pending") + seedEnrichmentRecord(t, fallbackRepo, testTenantID("taxonomy-backlog-ascii-space"), models.FieldTypeText, "\t\r\n") + seedEnrichmentRecord(t, fallbackRepo, testTenantID("taxonomy-backlog-unicode-space"), models.FieldTypeText, "\u00a0\u3000") seedEnrichmentRecord( t, fallbackRepo, testTenantID("taxonomy-backlog-categorical"), models.FieldTypeCategorical, "not enrichable") @@ -327,7 +332,7 @@ func TestCountEnrichmentBacklogAggregateIfLeader(t *testing.T) { require.NoError(t, err) require.True(t, isLeader) assert.Equal(t, before.TaxonomyEmbeddingPending+1, counts.TaxonomyEmbeddingPending, - "only the text record enters the live translation-to-taxonomy backlog") + "only the non-blank text record enters the live translation-to-taxonomy backlog") // Prove the leader returns the real aggregate, not a zero value. want, err := repository.NewEnrichmentStatusRepository(dbLeader).CountEnrichmentBacklogAggregate(ctx, "") diff --git a/tests/taxonomy_api_test.go b/tests/taxonomy_api_test.go index fd1cad3b..c72ebea9 100644 --- a/tests/taxonomy_api_test.go +++ b/tests/taxonomy_api_test.go @@ -303,6 +303,69 @@ func TestTaxonomyAPI_PublicReadAndEdit(t *testing.T) { }) } +func TestTaxonomyAPI_FieldOptionsExposeCurrentEmbeddingFailures(t *testing.T) { + ctx := context.Background() + harness := setupTaxonomyAPIServer(t) + scope := uniqueTaxonomyScope("tax-api-embedding-failures") + failuresRepo := repository.NewEnrichmentFailuresRepository(harness.db) + + t.Cleanup(func() { + _, _ = harness.db.Exec(ctx, `DELETE FROM feedback_records WHERE tenant_id = $1`, scope.TenantID) + }) + + seed := func(label string) *models.FeedbackRecord { + t.Helper() + + record := &models.FeedbackRecord{} + err := harness.db.QueryRow(ctx, ` + INSERT INTO feedback_records ( + source_type, source_id, field_id, field_label, field_type, + value_text, tenant_id, submission_id + ) + VALUES ($1, $2, $3, 'Feedback', 'text'::field_type_enum, $4, $5, $6) + RETURNING id, tenant_id, updated_at`, + scope.SourceType, scope.SourceID, scope.FieldID, "feedback "+label, + scope.TenantID, "submission-"+uuid.NewString(), + ).Scan(&record.ID, &record.TenantID, &record.UpdatedAt) + require.NoError(t, err) + + return record + } + + transient := seed("transient") + terminal := seed("terminal") + oldModel := seed("old-model") + + recordFailure := func(record *models.FeedbackRecord, model string, terminal bool, reason string) { + t.Helper() + require.NoError(t, failuresRepo.RecordFailure(ctx, models.EnrichmentFailure{ + FeedbackRecordID: record.ID, + TenantID: record.TenantID, + Enrichment: models.EnrichmentNameTaxonomyEmbedding, + Terminal: terminal, + Reason: reason, + Attempts: 5, + ContextKey: model, + SourceUpdatedAt: &record.UpdatedAt, + })) + } + + recordFailure(transient, taxonomyEmbeddingModel, false, models.EnrichmentFailureReasonProviderError) + recordFailure(terminal, taxonomyEmbeddingModel, true, "content_filter") + recordFailure(oldModel, "taxonomy:retired-model", true, "content_filter") + + var responseBody models.TaxonomyFieldsResponse + requestTaxonomyJSON(ctx, t, http.MethodGet, + taxonomyURL(harness.server.URL, "/v1/taxonomy/fields", url.Values{"tenant_id": {scope.TenantID}}), + harness.apiKey, nil, http.StatusOK, &responseBody) + + require.Len(t, responseBody.Data, 1) + assert.Equal(t, 3, responseBody.Data[0].RecordCount) + assert.Zero(t, responseBody.Data[0].EmbeddingCount) + assert.Equal(t, 1, responseBody.Data[0].EmbeddingFailedCount) + assert.Equal(t, 1, responseBody.Data[0].EmbeddingFailedTerminalCount) +} + // TestTaxonomyAPI_TenantIsolation proves the public endpoints reject another tenant's // identifiers: reads and edits 404, and node record drilldown returns nothing. func TestTaxonomyAPI_TenantIsolation(t *testing.T) { @@ -741,6 +804,57 @@ func TestTaxonomyAPI_InternalServiceEndpoints(t *testing.T) { assert.Equal(t, translated, input.Records[0].ValueText) }) + t.Run("run input falls back from whitespace translation and excludes blank embedded rows", func(t *testing.T) { + scope := uniqueTaxonomyScope("tax-internal-translation-whitespace") + cleanupTaxonomyTenant(ctx, t, harness.db, scope.TenantID) + + original := "Source text remains usable" + + var validID, blankID uuid.UUID + + err := harness.db.QueryRow(ctx, ` + INSERT INTO feedback_records ( + source_type, source_id, field_id, field_label, field_type, + value_text, value_text_translated, tenant_id, submission_id + ) + VALUES ($1, $2, $3, 'Feedback', 'text'::field_type_enum, $4, U&'\3000', $5, $6) + RETURNING id`, + scope.SourceType, scope.SourceID, scope.FieldID, original, + scope.TenantID, "submission-"+uuid.NewString(), + ).Scan(&validID) + require.NoError(t, err) + + err = harness.db.QueryRow(ctx, ` + INSERT INTO feedback_records ( + source_type, source_id, field_id, field_label, field_type, + value_text, value_text_translated, tenant_id, submission_id + ) + VALUES ($1, $2, $3, 'Feedback', 'text'::field_type_enum, U&'\000B', U&'\00A0\3000', $4, $5) + RETURNING id`, + scope.SourceType, scope.SourceID, scope.FieldID, + scope.TenantID, "submission-"+uuid.NewString(), + ).Scan(&blankID) + require.NoError(t, err) + + embedding := make([]float32, models.EmbeddingVectorDimensions) + embedding[0] = 0.25 + require.NoError(t, harness.embeddingsRepo.Upsert(ctx, validID, taxonomyEmbeddingModel, embedding, nil)) + require.NoError(t, harness.embeddingsRepo.Upsert(ctx, blankID, taxonomyEmbeddingModel, embedding, nil)) + + recordCount, embeddingCount, _, err := harness.repo.CountScopeInput(ctx, scope, taxonomyEmbeddingModel) + require.NoError(t, err) + assert.Equal(t, 1, recordCount) + assert.Equal(t, 1, embeddingCount) + + runID := startRunForScope(ctx, t, harness, scope) + inputURL := harness.server.URL + "/internal/v1/taxonomy/runs/" + runID.String() + "/input" + + var input models.TaxonomyRunInputResponse + requestTaxonomyJSON(ctx, t, http.MethodGet, inputURL, harness.internalToken, nil, http.StatusOK, &input) + require.Len(t, input.Records, 1) + assert.Equal(t, original, input.Records[0].ValueText) + }) + t.Run("run input includes translated-only records", func(t *testing.T) { scope := uniqueTaxonomyScope("tax-internal-translated-only-input") cleanupTaxonomyTenant(ctx, t, harness.db, scope.TenantID) @@ -798,6 +912,24 @@ func TestTaxonomyAPI_InternalServiceEndpoints(t *testing.T) { seedEmbeddedFeedback(ctx, t, harness, firstFieldScope, taxonomyMinEmbeddedRecords) seedEmbeddedFeedback(ctx, t, harness, secondFieldScope, taxonomyMinEmbeddedRecords+1) + var blankID uuid.UUID + + err := harness.db.QueryRow(ctx, ` + INSERT INTO feedback_records ( + source_type, source_id, field_id, field_label, field_type, + value_text, value_text_translated, tenant_id, submission_id + ) + VALUES ($1, $2, $3, 'Feedback', 'text'::field_type_enum, U&'\000B', U&'\00A0\3000', $4, $5) + RETURNING id`, + firstFieldScope.SourceType, firstFieldScope.SourceID, firstFieldScope.FieldID, + directoryScope.TenantID, "submission-"+uuid.NewString(), + ).Scan(&blankID) + require.NoError(t, err) + + embedding := make([]float32, models.EmbeddingVectorDimensions) + embedding[0] = 0.25 + require.NoError(t, harness.embeddingsRepo.Upsert(ctx, blankID, taxonomyEmbeddingModel, embedding, nil)) + runID := startRunForScope(ctx, t, harness, directoryScope) inputURL := harness.server.URL + "/internal/v1/taxonomy/runs/" + runID.String() + "/input" diff --git a/tests/taxonomy_no_source_test.go b/tests/taxonomy_no_source_test.go index b9f5efdd..0dbb0228 100644 --- a/tests/taxonomy_no_source_test.go +++ b/tests/taxonomy_no_source_test.go @@ -70,6 +70,24 @@ func TestTaxonomyNoSourceScope(t *testing.T) { require.NoError(t, err) } + // The taxonomy UI and CreateRun coverage gate must use the same input eligibility as the + // embedding worker and reconciler. Blank translated text falls back to source text, while + // records containing only ASCII or Unicode whitespace are excluded entirely. + //nolint:dupword // The two blank rows intentionally repeat each value in raw and translated columns. + _, err := db.Exec(ctx, ` + INSERT INTO feedback_records ( + source_type, source_id, field_id, field_label, field_type, + value_text, value_text_translated, tenant_id, submission_id + ) + VALUES + ($1, NULL, $2, 'Feedback', 'text'::field_type_enum, 'Fallback text', U&'\3000', $3, $4), + ($1, NULL, $2, 'Feedback', 'text'::field_type_enum, U&'\000B', U&'\000B', $3, $5), + ($1, NULL, $2, 'Feedback', 'text'::field_type_enum, U&'\00A0\3000', U&'\00A0\3000', $3, $6)`, + sourceType, fieldID, tenantID, + "submission-"+uuid.NewString(), "submission-"+uuid.NewString(), "submission-"+uuid.NewString(), + ) + require.NoError(t, err) + // Discovery surfaces the records as a single "no source" bucket with empty SourceID. options, err := repo.ListFieldOptions(ctx, tenantID, embeddingModel) require.NoError(t, err) @@ -84,7 +102,8 @@ func TestTaxonomyNoSourceScope(t *testing.T) { require.NotNil(t, noSource, "expected a discovered field option for the no-source bucket") require.Empty(t, noSource.SourceID, "no-source bucket must expose an empty source_id") - require.Equal(t, 2, noSource.RecordCount, "NULL and blank source_id must collapse into one bucket") + require.Equal(t, 3, noSource.RecordCount, + "source text plus translated-whitespace fallback count; whitespace-only records do not") // Counting the empty-source scope matches both NULL and blank feedback rows. scope := models.TaxonomyScope{ @@ -96,7 +115,14 @@ func TestTaxonomyNoSourceScope(t *testing.T) { recordCount, _, _, err := repo.CountScopeInput(ctx, scope, embeddingModel) require.NoError(t, err) - require.Equal(t, 2, recordCount, "empty-source scope must null-safe match NULL/blank source rows") + require.Equal(t, 3, recordCount, "field scope must use worker-equivalent text eligibility") + + directoryCount, _, _, err := repo.CountScopeInput(ctx, models.TaxonomyScope{ + ScopeType: models.TaxonomyScopeTypeDirectory, + TenantID: tenantID, + }, embeddingModel) + require.NoError(t, err) + require.Equal(t, 3, directoryCount, "directory scope must use worker-equivalent text eligibility") // A taxonomy run can be created for the empty-source scope and is found by the // in-progress guard (empty string is a valid, comparable key in taxonomy tables).