fix: self-heal missing taxonomy embeddings

.env.example

@@ -127,11 +127,17 @@ WEBHOOK_MAX_COUNT=500
 # EMBEDDING_MODEL=<model name>        (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.

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: ""

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)
 	}

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)
 

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) {

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))
+	})
+}

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