diff --git a/cmd/silo/main.go b/cmd/silo/main.go index a5098b26..c1e36143 100644 --- a/cmd/silo/main.go +++ b/cmd/silo/main.go @@ -1538,6 +1538,7 @@ func main() { cfg.Recommendations.TasteProfilesCron, cfg.Recommendations.CowatchCron, cfg.Recommendations.RecommendationsCron, + cfg.Recommendations.EmbeddingsJobTimeout, ) if err != nil { slog.Error("failed to create recommendation worker", "error", err) diff --git a/internal/config/config.go b/internal/config/config.go index 1668ef51..f45994e1 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -246,6 +246,10 @@ type RecommendationsConfig struct { TasteDecayHalfLifeDays float64 `yaml:"-"` DiversityLambda float64 `yaml:"-"` CowatchCron string `yaml:"-"` + // EmbeddingsJobTimeout bounds a single embedding backfill run. A local + // CPU embedder over a large catalog needs hours, so this defaults to 24h + // (replacing a hardcoded 30m that truncated large first-run backfills). + EmbeddingsJobTimeout time.Duration `yaml:"-"` } // AIConfig holds the shared connection settings for Silo's AI features diff --git a/internal/config/db_loader.go b/internal/config/db_loader.go index 7e23fbe9..3bd3927b 100644 --- a/internal/config/db_loader.go +++ b/internal/config/db_loader.go @@ -413,6 +413,10 @@ func LoadFromDB(m map[string]string) (*Config, error) { }()) cfg.Recommendations.EmbeddingAuthToken = stringOr(m, "recommendations.embedding_auth_token", stringOr(m, "recommendations.openai_api_key", "")) cfg.Recommendations.EmbeddingsCron = stringOr(m, "recommendations.embeddings_cron", "0 3 * * *") + cfg.Recommendations.EmbeddingsJobTimeout, err = durationOr(m, "recommendations.embeddings_job_timeout", 24*time.Hour) + if err != nil { + return nil, err + } cfg.Recommendations.TasteProfilesCron = stringOr(m, "recommendations.taste_profiles_cron", "0 4 * * *") cfg.Recommendations.RecommendationsCron = stringOr(m, "recommendations.recommendations_cron", "0 5 * * *") tasteDecayHalfLife, err := floatOr(m, "recommendations.taste_decay_half_life_days", 180) diff --git a/internal/recommendations/worker.go b/internal/recommendations/worker.go index 2a64dc1b..5bdc3960 100644 --- a/internal/recommendations/worker.go +++ b/internal/recommendations/worker.go @@ -29,6 +29,7 @@ type Worker struct { profileRefreshCh chan profileRefreshRequest profileRefreshPending map[string]struct{} cancelFunc context.CancelFunc + embeddingsJobTimeout time.Duration } const tasteProfileRefreshSubjectsQuery = ` @@ -45,13 +46,17 @@ const tasteProfileRefreshSubjectsQuery = ` SELECT DISTINCT user_id, profile_id FROM user_watchlist` // NewWorker creates a new recommendation Worker. -func NewWorker(engine *Engine, embeddingsCron, tasteProfilesCron, cowatchCron, recommendationsCron string) (*Worker, error) { +func NewWorker(engine *Engine, embeddingsCron, tasteProfilesCron, cowatchCron, recommendationsCron string, embeddingsJobTimeout time.Duration) (*Worker, error) { + if embeddingsJobTimeout <= 0 { + embeddingsJobTimeout = 24 * time.Hour + } w := &Worker{ engine: engine, cron: cron.New(), running: make(map[JobName]bool), profileRefreshCh: make(chan profileRefreshRequest, 256), profileRefreshPending: make(map[string]struct{}), + embeddingsJobTimeout: embeddingsJobTimeout, } if _, err := w.cron.AddFunc(embeddingsCron, w.runEmbeddings); err != nil { @@ -123,9 +128,9 @@ func (w *Worker) TriggerEmbeddings() error { } go func() { defer w.setRunning(JobEmbeddings, false) - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), w.embeddingsJobTimeout) defer cancel() - slog.Info("starting embedding job (manual trigger)") + slog.Info("starting embedding job (manual trigger)", "timeout", w.embeddingsJobTimeout) count, err := w.engine.EmbedAll(ctx) if err != nil { slog.Error("embedding job failed", "error", err, "embedded", count) @@ -236,9 +241,9 @@ func (w *Worker) runEmbeddings() { } defer w.setRunning(JobEmbeddings, false) - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), w.embeddingsJobTimeout) defer cancel() - slog.Info("starting embedding job") + slog.Info("starting embedding job", "timeout", w.embeddingsJobTimeout) count, err := w.engine.EmbedAll(ctx) if err != nil { slog.Error("embedding job failed", "error", err, "embedded", count)