InferenceWorker

Location: iris/inference_worker/ ⧉ Language: Python Role: Sidecar process per replica; consumes requests, calls model, reports metrics

InferenceWorker is a Python daemon that runs alongside the model container (e.g., vLLM) in each replica's docker-compose environment. It implements the competing consumers pattern by pulling requests from RabbitMQ, forwarding them to the local model, writing results to Redis, and reporting health and metrics data.

Architecture

docker-compose replica (per deployment replica)
┌────────────────────────────────────────────────┐
│                                                │
│   InferenceWorker          ModelContainer      │
│   ┌──────────────┐         ┌───────────────┐   │
│   │ listener     │  HTTP   │ vLLM / LLM    │   │
│   │ health_check │────────>│ :8000         │   │
│   │ metrics      │<────────│               │   │
│   └──────────────┘         └───────────────┘   │
│        │    ▲                                  │
└────────┼────┼──────────────────────────────────┘
         │    │
         │    └─── Redis (write results)
         │
         └──────── RabbitMQ (consume requests)

Three Concurrent Tasks

InferenceWorker runs three async tasks simultaneously:

1. RabbitMQ Listener (`rmq_listener.py`)

Consumes messages from the deployment's queue and processes inference requests.

Features:

Custom connection lifecycle (not RobustQueue) for full QoS control
Exponential backoff reconnection: 1s → 2s → ... → 30s
Consumer tag tracking to avoid zombie channels
Prefetch control (default 1 in-flight message)
Can be suspended/resumed based on health status

2. Health Checker (`health_checker.py`)

Periodically polls the model container's /health endpoint.

Behavior:

Polls every HEALTH_CHECK_INTERVAL_S (default 10s)
On unhealthy: triggers listener suspension (stops consuming messages)
On healthy again: triggers listener resumption
Prevents message processing when model is unavailable

3. Metrics Reporter (`metrics_reporter.py`)

Flushes collected metrics to MetricServer and Croesus (billing).

Flushes every METRICS_FLUSH_INTERVAL_S (default 30s):

Inference request records (latency, tokens, status codes)
Health snapshot (healthy bool, timestamp)
GPU/KV-cache metrics from scraper
RabbitMQ queue depth
Token usage to Croesus as CloudEvents

Best-effort: Failures logged and dropped, doesn't block message processing.

Data Flow

RabbitMQ: infer_req.<deployment_id>
      │
      ▼
  RabbitMQListener (rmq_listener.py)
      │
      │  (consumer suspended while unhealthy)
      │◄──── suspend_consuming() / resume_consuming() ◄──── HealthChecker
      │                                                      (polls /health)
      │
      ▼
  MessageProcessor (message_processor.py)
      │
      ├─ Parse error  ──► Redis: {status: failed, 400}  ──► nack
      ├─ Expired      ──► Redis: {status: failed, 408}  ──► ack
      │
      ▼
  ModelClient (model_client.py)
      │
      │  POST http://model:8000/v1/chat/completions
      │
      ├─ 2xx  ──► Redis: {status: completed, result}  ──► ack
      ├─ 4xx/5xx ──► Redis: {status: failed, error}   ──► ack
      │
      └─ Network/timeout exception
              ├─ retry < max  ──► republish with x-retry-count+1  ──► ack
              └─ max retries  ──► Redis: {status: failed, 500}    ──► nack → DLQ

  MetricsBuffer (metrics_buffer.py)
      ▲
      │ InferenceRequestRecord appended at disposition
      │
      ▼ (every 30s)
  MetricsReporter
      ├─ POST /metrics → MetricServer
      └─ POST /events → Croesus (billing)

Message Processing

Request Validation

Parse InferenceRequest:

{
  "request_id": "uuid",
  "received_at": 1234567890.123,  # unix timestamp from proxy
  "timeout_s": 60.0,
  "payload": { /* OpenAI chat completions request */ }
}

Expiration Check:

if (now() - received_at) > timeout_s:
    write_redis(request_id, {"status": "failed", "status_code": 408})
    ack_message()
    return  # skip processing

Model Invocation

response = model_client.infer(
    url=MODEL_CONTAINER_URL + MODEL_INFER_PATH,
    timeout=MODEL_INFER_TIMEOUT_S,
    payload=request.payload
)

Captures: - status_code - HTTP status from model - response_body - Full response JSON - ttft (time-to-first-token) - If model returns header - tbt (time-between-tokens) - If model returns header

Result Writing

Success (2xx):

redis.set(
    f"infer_resp:{request_id}",
    json.dumps({
        "status": "completed",
        "status_code": 200,
        "result": response_body
    }),
    ex=REDIS_RESPONSE_TTL_S
)

Failure (4xx/5xx):

redis.set(
    f"infer_resp:{request_id}",
    json.dumps({
        "status": "failed",
        "status_code": status_code,
        "error": error_message
    }),
    ex=REDIS_RESPONSE_TTL_S
)

Retry Logic

Retriable Errors: - Network errors calling model (connection refused, timeout) - Temporary model unavailability

Non-Retriable Errors: - Parse errors (malformed JSON) - Expired requests - Model returned 4xx/5xx (application error)

Retry Flow:

if retry_count < RABBITMQ_MAX_RETRIES:
    # Republish with incremented retry count
    republish(message, retry_count + 1)
    ack_original()
else:
    # Max retries exceeded, send to DLQ
    nack()  # RabbitMQ moves to dead-letter queue

Headers: - x-retry-count - Incremented on each retry

Redis Write Failure Handling

If Redis write fails: 1. Catch RedisError exception 2. Nack message with requeue=True 3. Sleep REDIS_RETRY_DELAY_S (default 10s) 4. RabbitMQ redelivers to any available worker

Prevents: - Consumer crash loops - Message loss when Redis is temporarily down

Configuration

Environment Variables

Variable	Default	Description
RabbitMQ
`RABBITMQ_HOST`	`localhost`	Server hostname
`RABBITMQ_PORT`	`5672`	Server port
`RABBITMQ_USERNAME`	`guest`	Auth username
`RABBITMQ_PASSWORD`	`guest`	Auth password
`RABBITMQ_VHOST`	`/`	Virtual host
`RABBITMQ_HEARTBEAT_S`	`120`	Connection heartbeat interval (seconds)
`RABBITMQ_PREFETCH`	`1`	In-flight messages per worker
`RABBITMQ_MAX_RETRIES`	`1`	Retry attempts before DLQ
Dead-Letter
`INFERENCE_DLX`	`infer.dlx`	Dead-letter exchange name
`INFERENCE_DLQ`	`infer.dlq`	Dead-letter queue name
`INFERENCE_DLX_ROUTING_KEY`	`dead`	DLX routing key
Model
`MODEL_CONTAINER_URL`	`http://localhost:8000`	Model container base URL
`MODEL_INFER_PATH`	`/v1/chat/completions`	Inference endpoint path
`MODEL_INFER_TIMEOUT_S`	`300`	HTTP timeout for inference
Health
`HEALTH_CHECK_INTERVAL_S`	`10`	Polling interval for /health
Redis
`REDIS_URL`	`redis://localhost:6379`	Connection URL
`REDIS_PASSWORD`	(empty)	AUTH password
`REDIS_RESPONSE_TTL_S`	`3600`	TTL for response keys (1 hour)
`REDIS_RETRY_DELAY_S`	`10`	Delay before requeue on write failure
Metrics
`METRIC_SERVER_URL`	required	MetricServer endpoint
`METRIC_SERVER_SERVICE_TOKEN`	required	Auth token for MetricServer
`METRICS_FLUSH_INTERVAL_S`	`30`	How often to flush metrics
`VLLM_METRICS_URL`	(derived from `MODEL_CONTAINER_URL`)	vLLM Prometheus metrics endpoint for KV cache and GPU metrics
`DCGM_METRICS_URL`	(empty)	DCGM GPU metrics endpoint on the host
Billing
`CROESUS_HOST`	required	Croesus (billing) server URL
Identity
`DEPLOYMENT_ID`	required	UUID of deployment
`REPLICA_ID`	required	Unique ID for this replica
`USER_ID`	required	Owner user ID
`ORG_ID`	(empty)	Organization ID
`HF_MODEL_ID`	required	HuggingFace model identifier

Error Scenarios

Model Container Down

Health checker marks unhealthy
Listener suspends consuming (stops pulling from queue)
Other replicas handle queue
When model recovers: health checker marks healthy, listener resumes

Redis Temporarily Unavailable

Write to infer_resp:<request_id> fails
Catch RedisError, nack message with requeue
Sleep 10s (backpressure)
RabbitMQ redelivers message
When Redis recovers: message processed successfully

RabbitMQ Connection Lost

Listener detects connection drop
Exponential backoff reconnection: 1s, 2s, 4s, ..., 30s (max)
On reconnect: re-declare infrastructure (queues, DLX, bindings)
Resume consuming from last acked message

MetricServer Unavailable

Metrics flush fails (HTTP error)
Log warning, drop metrics batch
Continue processing requests normally
Next flush interval retries

Rationale: Metrics are important but not critical for inference pipeline.

Performance Characteristics

Throughput

Controlled by: - RABBITMQ_PREFETCH - In-flight messages per worker - Model inference latency - Health check overhead (minimal, async)

Typical Setup:

Prefetch = 1: Sequential processing, no queue buildup in worker
Prefetch > 1: Parallel processing if model supports concurrency

Latency

Components:

RabbitMQ delivery: <5ms (local network)
Message processing overhead: <10ms (parse, validate)
Model inference: 100ms–10s (model-dependent)
Redis write: <5ms
Total overhead: ~20ms per request

2026-06-042026-06-10