fix: prevent persistent 429 responses after model worker failures

[cyb-fox]

Summary

This PR fixes a failure mode where a single Infinity instance could begin
returning persistent HTTP 429 responses after a model worker failed.

HTTP 429 in Infinity is queue-based backpressure, not a per-client rate limit:
the server returns it when a model's pending input queue exceeds
INFINITY_QUEUE_SIZE (32,000 items by default).

Root Cause

Before this change, an uncaught exception in a preprocessing, inference, or
postprocessing worker only terminated that background task.

Requests already accepted by the server held futures that were never completed.
The instance continued accepting requests, the in-memory priority queue grew,
and once it exceeded INFINITY_QUEUE_SIZE, all subsequent requests for that
model received HTTP 429.

Restarting the instance appeared to fix the problem because it recreated the
workers and cleared the in-memory queue.

Changes

• Supervise the batch publisher, result subscriber, and every model worker stage.
• Mark the engine unhealthy when a background task exits unexpectedly.
• Fail all in-flight request futures when the engine becomes unhealthy.
• Return HTTP 503 for unhealthy engines instead of continuing to accumulate work.
• Make /health return HTTP 503 when any loaded engine is unhealthy, allowing
  deployment platforms to remove and restart the instance.
• Add optional INFINITY_REQUEST_TIMEOUT detection for workers that remain
  alive but stop producing results.
• Validate that worker result counts match the input batch before completing
  request futures.
• Document queue overload behavior, health checks, and timeout configuration.

INFINITY_REQUEST_TIMEOUT defaults to 0 to preserve existing behavior for
long-running inference requests. Deployments that need stalled-request
detection can opt in with a value appropriate for their latency budget.

Related Issue

N/A

Checklist

• I have read the CONTRIBUTING guidelines.
• I have added tests to cover worker failure, request timeout, and result-length mismatch.
• I have updated the documentation (docs folder) accordingly.

Verification

• python -m compileall -q libs/infinity_emb/infinity_emb passes.
• Full pytest execution was not run in the current environment because
  sentence_transformers is not installed.

[greptile-apps]

Greptile Summary

This PR supervises batching workers and exposes engine failures through request and health responses instead of allowing queues to grow indefinitely.

• Tracks and fails in-flight futures when background processing fails.
• Adds optional request-timeout monitoring and validates batch result cardinality.
• Returns HTTP 503 for unhealthy engines and adds focused worker-failure tests.

Confidence Score: 3/5

The PR should not merge until accepted requests cannot escape failure completion and the timeout distinguishes a stalled worker from a healthy but slow or backlogged engine.

Worker failure can race with in-flight registration and leave a request unresolved, while the optional watchdog can permanently disable an engine solely because one queued request exceeds a total-latency threshold.

Files Needing Attention: libs/infinity_emb/infinity_emb/inference/batch_handler.py

Important Files Changed

Filename	Overview
libs/infinity_emb/infinity_emb/inference/batch_handler.py	Adds worker supervision, in-flight failure, timeout monitoring, and result validation, but leaves a failure-registration race and conflates request latency with worker stalls.
libs/infinity_emb/infinity_emb/infinity_server.py	Propagates unhealthy engine state as HTTP 503 from model endpoints and the health check.
libs/infinity_emb/infinity_emb/engine.py	Exposes BatchHandler health through the asynchronous engine abstraction.
libs/infinity_emb/infinity_emb/env.py	Adds non-negative parsing for the optional request-timeout setting.
libs/infinity_emb/tests/unit_test/inference/test_batch_handler.py	Covers worker exceptions, permanently blocked inference, and mismatched result counts, but not concurrent registration failure or slow-progressing workers.

Sequence Diagram

sequenceDiagram
    participant R as Request
    participant B as BatchHandler
    participant W as Worker thread
    participant H as Health callback
    R->>B: schedule items
    B->>B: check is_healthy()
    W-->>H: stage exits
    H->>H: set unhealthy and shutdown
    H-->>B: fail current inflight snapshot
    B->>B: register new futures
    B->>B: enqueue after workers stopped
    B-->>R: future remains pending

Loading

_{Reviews (1): Last reviewed commit: "fix: fail requests when a model worker s..." | Re-trigger Greptile}

[greptile-apps]

In-flight registration race

When a publisher or model-stage thread fails after this health check but before the new futures enter _inflight, the one-shot failure callback misses those futures and the stopped workers cannot complete them, causing the accepted request to wait indefinitely.

[greptile-apps]

Request latency triggers engine failure

When INFINITY_REQUEST_TIMEOUT is enabled and a request exceeds the threshold because of legitimate inference time or queue backlog, this check marks the entire engine unhealthy without determining whether workers are still producing results, causing all in-flight and subsequent requests to receive HTTP 503 until restart.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: fd035c1d05

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Preserve root worker errors when marking unhealthy

When a worker stage or the subscriber wraps the real exception in a generic ValueError (for example _core_batch crashed. or _subscribe_to_model crashed), interpolating only that wrapper here drops the underlying cause. In the new failure scenarios, waiting requests receive a generic EngineUnhealthyError, so the added tests that match simulated core failure or results for a batch fail and operators lose the diagnostic reason in the 503 response; unwrap/preserve the original exception when constructing the unhealthy error.

Useful? React with 👍 / 👎.