From d6fe7c38d5d432b08d7de8c22c83af3f90ad94ca Mon Sep 17 00:00:00 2001 From: GreenEclipse Date: Sun, 16 Aug 2026 18:59:29 +0200 Subject: [PATCH] fix --- apps/api/cmd/worker/main.go | 20 +-- apps/api/internal/processing/openai.go | 125 ++++++++++++++++-- apps/api/internal/processing/openai_test.go | 84 ++++++++++++ apps/api/internal/processing/pipeline.go | 8 +- apps/api/internal/processing/stuck_cleanup.go | 5 +- 5 files changed, 215 insertions(+), 27 deletions(-) diff --git a/apps/api/cmd/worker/main.go b/apps/api/cmd/worker/main.go index cd22585..bc2f5f1 100644 --- a/apps/api/cmd/worker/main.go +++ b/apps/api/cmd/worker/main.go @@ -161,16 +161,16 @@ func main() { log.Printf("worker WARNING: ALLOW_INSECURE_LOCAL_PRODUCTION=1 with loopback WEB_ORIGIN=%s — not for public deploy", cfg.WebOrigin) } - // Before claiming a fresh heartbeat, reclaim running orphans left by a prior - // crash/SIGTERM (ClaimNext only selects pending). Skip when another worker is live. + // This process cannot resume prior in-memory work. Always reclaim running + // orphans before touching heartbeat so a fast restart (deploy/rb within the + // heartbeat stale window) does not leave ClaimNext-invisible status=running + // jobs. Single-worker architecture only — never run two processing workers. probe := jobs.ProbeWorkerReadiness(ctx, pool, jobs.DefaultHeartbeatStaleAfter) - if probe.WorkerCheck == "missing" || probe.WorkerCheck == "stale" { - if res, err := processing.ReclaimOrphanedRunning(ctx, pool); err != nil { - log.Printf("worker orphan reclaim: %v", err) - } else if res.JobsRequeued > 0 || res.ProductsReset > 0 || res.SyncJobsRequeued > 0 { - log.Printf("worker orphan reclaim jobs_requeued=%d products_reset=%d sync_requeued=%d (prior worker %s)", - res.JobsRequeued, res.ProductsReset, res.SyncJobsRequeued, probe.WorkerCheck) - } + if res, err := processing.ReclaimOrphanedRunning(ctx, pool); err != nil { + log.Printf("worker orphan reclaim: %v", err) + } else if res.JobsRequeued > 0 || res.ProductsReset > 0 || res.SyncJobsRequeued > 0 { + log.Printf("worker orphan reclaim jobs_requeued=%d products_reset=%d sync_requeued=%d (heartbeat was %s)", + res.JobsRequeued, res.ProductsReset, res.SyncJobsRequeued, probe.WorkerCheck) } if err := jobs.TouchHeartbeat(ctx, pool, jobs.ProcessingWorkerID); err != nil { @@ -224,7 +224,7 @@ func main() { log.Printf("support auto AI jobs processed=%d", n) } if _, err := jobSlots.Fill(ctx, pipeline.ClaimNext, func(jobCtx context.Context, jobID uuid.UUID) error { - log.Printf("processing job %s", jobID) + log.Printf("processing: claimed job=%s", jobID) return pipeline.ProcessJob(jobCtx, jobID) }, markJobFailed); err != nil && !errors.Is(err, pgx.ErrNoRows) { log.Printf("claim error: %v", err) diff --git a/apps/api/internal/processing/openai.go b/apps/api/internal/processing/openai.go index 97b2c99..007a5cf 100644 --- a/apps/api/internal/processing/openai.go +++ b/apps/api/internal/processing/openai.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "log" "math" "math/rand" "net" @@ -39,6 +40,16 @@ type OpenAIClient struct { const maxOpenAIRetries = 8 +// openAIHTTPTimeout is the full-request limit for OpenAI-compatible chat/embeddings. +// Reasoning models (e.g. code-fast) often need 60–180s; 240s reduces false timeouts. +// Timeouts are not retried (see classifyOpenAITransportErr) so a hung upstream fails +// the item in ~4m instead of MaxRetries×240s of silent PROCESSING. +const openAIHTTPTimeout = 240 * time.Second + +// openAIWaitLogInterval emits progress while HTTPClient.Do blocks (cancel is only +// checked between job items, so mid-call silence looks like a stuck job). +const openAIWaitLogInterval = 30 * time.Second + func NewOpenAIClient(apiKey, baseURL, model string, rpm, maxRetries int) *OpenAIClient { if baseURL == "" { baseURL = "https://api.openai.com/v1" @@ -57,14 +68,12 @@ func NewOpenAIClient(apiKey, baseURL, model string, rpm, maxRetries int) *OpenAI } // Dial-time SSRF. Loopback when base URL is local; RFC1918 only in non-prod. policy := openAIDialPolicy(baseURL) - // Header/body timeout: code-fast often exceeds 180s; 240s reduces false timeouts. - // Tradeoff: slower fail on hung upstream. Prefer synthesize fallback over infinite wait - // (RunSteps always synthesizes on timeout/error/empty — do not raise unboundedly). + // Prefer synthesize fallback on timeout/error/empty (RunSteps) over raising this unboundedly. return &OpenAIClient{ APIKey: apiKey, BaseURL: strings.TrimRight(baseURL, "/"), Model: model, - HTTPClient: security.SafeHTTPClientPolicy(240*time.Second, policy), + HTTPClient: security.SafeHTTPClientPolicy(openAIHTTPTimeout, policy), MinInterval: interval, MaxRetries: maxRetries, ModeLabel: AIProviderInternal, @@ -141,6 +150,79 @@ func IsMockOrLoopbackBaseURL(baseURL string) bool { return strings.Contains(lower, "mock-llm") } +// startWaitLogger logs request start and periodic progress while Do blocks. +// Call the returned stop with the Do error (or nil) when the request finishes. +func (c *OpenAIClient) startWaitLogger(op string, maxTokens int) func(error) { + if c == nil { + return func(error) {} + } + start := time.Now() + log.Printf("openai: request start op=%s model=%s base=%s max_tokens=%d timeout=%s", + op, c.Model, c.BaseURL, maxTokens, openAIHTTPTimeout) + done := make(chan struct{}) + var once sync.Once + go func() { + t := time.NewTicker(openAIWaitLogInterval) + defer t.Stop() + for { + select { + case <-done: + return + case <-t.C: + log.Printf("openai: still waiting op=%s model=%s base=%s elapsed=%s timeout=%s", + op, c.Model, c.BaseURL, time.Since(start).Round(time.Second), openAIHTTPTimeout) + } + } + }() + return func(err error) { + once.Do(func() { + close(done) + elapsed := time.Since(start).Round(time.Millisecond) + if err != nil { + log.Printf("openai: request done op=%s model=%s elapsed=%s err=%s", + op, c.Model, elapsed, TruncateError(err)) + return + } + log.Printf("openai: request done op=%s model=%s elapsed=%s ok=1", op, c.Model, elapsed) + }) + } +} + +// isOpenAITimeoutErr reports HTTP client / context deadline timeouts. +// These must not be retried: each attempt costs openAIHTTPTimeout and leaves the job looking stuck. +func isOpenAITimeoutErr(err error) bool { + if err == nil { + return false + } + if errors.Is(err, context.DeadlineExceeded) { + return true + } + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + return true + } + s := err.Error() + return strings.Contains(s, "Client.Timeout exceeded") || + strings.Contains(s, "context deadline exceeded") +} + +// classifyOpenAITransportErr maps Do/Read errors to (wrappedErr, retryable). +// Timeouts and cancel are fatal for this attempt chain; other network errors may retry. +func classifyOpenAITransportErr(model string, err error) (error, bool) { + if err == nil { + return nil, false + } + if errors.Is(err, context.Canceled) { + return err, false + } + if isOpenAITimeoutErr(err) { + return fmt.Errorf( + "openai timed out waiting for model %q after %s (not retrying; use a faster model or fix upstream latency): %w", + model, openAIHTTPTimeout, err), false + } + return err, true +} + type chatRequest struct { Model string `json:"model"` Messages []chatMessage `json:"messages"` @@ -218,9 +300,18 @@ func (c *OpenAIClient) CompleteWithOptions(ctx context.Context, system, user str return comp, nil } lastErr = err - if errors.Is(err, errEmptyModelResponse) && maxTok > 0 && maxTok < maxTokensReasoningBudget { - maxTok = maxTokensReasoningBudget - retryable = true + if errors.Is(err, errEmptyModelResponse) { + if maxTok <= 0 || maxTok < maxTokensReasoningBudget { + if maxTok < maxTokensReasoningBudget { + maxTok = maxTokensReasoningBudget + } + retryable = true + } else { + // Already at reasoning budget — another 240s call will not help. + return Completion{}, fmt.Errorf( + "openai empty response at max_tokens=%d for model %q (try a faster non-reasoning model): %w", + maxTok, c.Model, err) + } } if !retryable { return Completion{}, err @@ -325,14 +416,18 @@ func (c *OpenAIClient) doEmbed(ctx context.Context, texts []string) ([][]float32 req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+c.APIKey) + stopWait := c.startWaitLogger("embeddings", 0) res, err := c.HTTPClient.Do(req) + stopWait(err) if err != nil { - return nil, true, err + wrapped, retryable := classifyOpenAITransportErr(c.Model, err) + return nil, retryable, wrapped } defer res.Body.Close() raw, err := io.ReadAll(io.LimitReader(res.Body, 4<<20)) if err != nil { - return nil, true, err + wrapped, retryable := classifyOpenAITransportErr(c.Model, err) + return nil, retryable, wrapped } var parsed embeddingResponse if err := json.Unmarshal(raw, &parsed); err != nil { @@ -391,14 +486,18 @@ func (c *OpenAIClient) doComplete(ctx context.Context, system, user string, temp req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+c.APIKey) + stopWait := c.startWaitLogger("chat/completions", maxTokens) res, err := c.HTTPClient.Do(req) + stopWait(err) if err != nil { - return Completion{}, true, err + wrapped, retryable := classifyOpenAITransportErr(c.Model, err) + return Completion{}, retryable, wrapped } defer res.Body.Close() raw, err := io.ReadAll(io.LimitReader(res.Body, 1<<20)) if err != nil { - return Completion{}, true, err + wrapped, retryable := classifyOpenAITransportErr(c.Model, err) + return Completion{}, retryable, wrapped } var parsed chatResponse if err := json.Unmarshal(raw, &parsed); err != nil { @@ -426,7 +525,9 @@ func (c *OpenAIClient) doComplete(ctx context.Context, system, user string, temp } if text == "" { // Reasoning models often return empty content when max_tokens cuts mid-thought. - retryable := strings.EqualFold(finishReason, "length") + // Only retry when CompleteWithOptions can still raise max_tokens. + canBump := maxTokens <= 0 || maxTokens < maxTokensReasoningBudget + retryable := strings.EqualFold(finishReason, "length") && canBump return Completion{}, retryable, errEmptyModelResponse } return Completion{ diff --git a/apps/api/internal/processing/openai_test.go b/apps/api/internal/processing/openai_test.go index f43d388..edd1258 100644 --- a/apps/api/internal/processing/openai_test.go +++ b/apps/api/internal/processing/openai_test.go @@ -3,6 +3,7 @@ package processing import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -226,3 +227,86 @@ func TestOpenAIClient_doComplete_emptyContentLengthRetriesWithBudget(t *testing. t.Fatalf("text=%q", comp.Text) } } + +func TestOpenAIClient_Complete_timeoutNotRetried(t *testing.T) { + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + deadline := time.After(300 * time.Millisecond) + for { + select { + case <-r.Context().Done(): + return + case <-deadline: + _ = json.NewEncoder(w).Encode(map[string]any{ + "choices": []map[string]any{ + {"message": map[string]any{"content": "late"}}, + }, + }) + return + case <-time.After(10 * time.Millisecond): + } + } + })) + defer srv.Close() + c := NewOpenAIClient("test-key", srv.URL, "code-fast", 0, 3) + c.MaxRetries = 3 + c.HTTPClient = &http.Client{Timeout: 60 * time.Millisecond} + start := time.Now() + _, err := c.Complete(context.Background(), "sys", "user") + elapsed := time.Since(start) + if err == nil { + t.Fatal("expected timeout error") + } + if !strings.Contains(err.Error(), "timed out waiting for model") { + t.Fatalf("want clear timeout message, got: %v", err) + } + if calls != 1 { + t.Fatalf("calls=%d want 1 (timeout must not retry)", calls) + } + if elapsed > time.Second { + t.Fatalf("timeout too slow: %s", elapsed) + } +} + +func TestOpenAIClient_Complete_emptyAtReasoningBudgetNotRetried(t *testing.T) { + t.Parallel() + calls := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + _ = json.NewEncoder(w).Encode(map[string]any{ + "model": "code-fast", + "choices": []map[string]any{{ + "finish_reason": "length", + "message": map[string]any{"role": "assistant", "content": "", "reasoning_content": "still thinking"}, + }}, + "usage": map[string]int{"prompt_tokens": 1, "completion_tokens": 4096, "total_tokens": 4097}, + }) + })) + defer srv.Close() + c := NewOpenAIClient("test-key", srv.URL, "code-fast", 0, 3) + c.HTTPClient = srv.Client() + _, err := c.CompleteWithOptions(context.Background(), "sys", "user", CompleteOptions{MaxTokens: maxTokensReasoningBudget}) + if err == nil { + t.Fatal("expected empty response error") + } + if !strings.Contains(err.Error(), "empty response") { + t.Fatalf("err=%v", err) + } + if calls != 1 { + t.Fatalf("calls=%d want 1 (no retry at reasoning budget)", calls) + } +} + +func TestIsOpenAITimeoutErr(t *testing.T) { + t.Parallel() + if !isOpenAITimeoutErr(context.DeadlineExceeded) { + t.Fatal("DeadlineExceeded") + } + if !isOpenAITimeoutErr(fmt.Errorf("Get \"http://x\": context deadline exceeded (Client.Timeout exceeded while awaiting headers)")) { + t.Fatal("Client.Timeout message") + } + if isOpenAITimeoutErr(fmt.Errorf("connection refused")) { + t.Fatal("connection refused should not be timeout") + } +} diff --git a/apps/api/internal/processing/pipeline.go b/apps/api/internal/processing/pipeline.go index db4b258..8071abb 100644 --- a/apps/api/internal/processing/pipeline.go +++ b/apps/api/internal/processing/pipeline.go @@ -602,6 +602,10 @@ func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error { return nil } + // Log before AI resolve so a hung provider lookup still leaves a start line. + log.Printf("processing: start job=%s company=%s type=%s prior_processed=%d", + jobID, companyID, processingType, alreadyProcessed) + modeLabel := AIProviderInternal usingBYOK := false jobEngine := p.Engine @@ -625,9 +629,7 @@ func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error { } else if label := jobEngine.EngineProviderMode(); label != "" { modeLabel = label } - - log.Printf("processing: start job=%s company=%s type=%s mode=%s prior_processed=%d", - jobID, companyID, processingType, modeLabel, alreadyProcessed) + log.Printf("processing: provider job=%s mode=%s byok=%v", jobID, modeLabel, usingBYOK) progress := InitialStepProgress(processingType) if len(progress) > 0 { diff --git a/apps/api/internal/processing/stuck_cleanup.go b/apps/api/internal/processing/stuck_cleanup.go index 483aaf9..ad84a52 100644 --- a/apps/api/internal/processing/stuck_cleanup.go +++ b/apps/api/internal/processing/stuck_cleanup.go @@ -26,8 +26,9 @@ type OrphanReclaimResult struct { } // ReclaimOrphanedRunning resets in-flight work left by a dead worker process so -// ClaimNext / feed sync claim can pick it up again. Call only when the processing -// worker heartbeat is missing or stale (single-worker architecture) — never while +// ClaimNext / feed sync claim can pick it up again. Call on every worker process +// start (before TouchHeartbeat) under the single-worker architecture — a live +// heartbeat from a just-killed process must not skip reclaim. Never call while // another live worker may own the rows. // // Unlike CleanupStuck (age-gated fail), this requeues immediately: running → pending