Jobs and Messaging

Jobs and Messaging

Harpyx offloads slow, failure-prone work (document extraction, chunking, embedding, indexing) from HTTP request handlers into a dedicated background process. The messaging fabric is RabbitMQ; the background process is Harpyx.Worker.

Why asynchronous jobs

• Latency: the WebApp returns immediately after writing metadata and enqueuing — uploads feel instantaneous even for large files.
• Resilience: jobs can be retried independently of the originating HTTP session. A transient LLM provider outage no longer fails an upload, it just delays the document's transition to Ready.
• Scale: Workers can be scaled horizontally without touching the WebApp; the WebApp can be scaled horizontally without duplicating background work.

Components

Component	Location	Role
IJobQueue	Harpyx.Application.Interfaces.IJobQueue	Abstraction producers/consumers depend on
RabbitMqJobQueue	Harpyx.Infrastructure.Services.RabbitMqJobQueue	RabbitMQ.Client implementation
JobConsumerService	Harpyx.Worker.Services.JobConsumerService	BackgroundService that drains the queue
ExpiredProjectsCleanupService	Harpyx.Worker.Services.ExpiredProjectsCleanupService	Periodic cleanup of projects past LifetimeExpiresAtUtc
Job entity	Harpyx.Domain.Entities.Job	Persisted execution record (type, state, attempts, last error)

Job lifecycle

Pending (implicit)  ─enqueue─▶  Processing  ─success─▶  Completed
                                      │
                                      ├─ exception ─▶  Failed
                                      │
                                      └─ doc quarantined ─▶  Skipped (audited)

The JobState enum (src/Harpyx.Domain/Enums/JobState.cs) captures these transitions. Every transition is persisted so the admin UI can show the current state and last error without needing to replay the broker.

ParseDocumentJob

The only job type currently in flight. Implemented in src/Harpyx.Worker/Services/JobConsumerService.cs:

1. Receive documentId from RabbitMQ.
2. Insert a Job row with State = Processing, AttemptCount = 1.
3. Record a job_start audit event.
4. Load the document; if its state is Quarantined or Rejected, short-circuit with job_skipped_security.
5. Otherwise transition the document to Processing / Extracting and call IRagIngestionService.IngestDocumentAsync.
6. On success, mark the job Completed and record job_end. On exception, mark both the job and the document Failed, record job_error, and rethrow so RabbitMQ's retry / dead-letter machinery takes over.

Each outcome increments a counter on HarpyxObservability.JobProcessedCounter with tags job_type and result ∈ {completed, failed, skipped_security} — see Observability.

Retries and dead-letter

RabbitMqJobQueue declares the main queue with a dead-letter exchange. The WebApp and Worker negotiate the same queue topology at startup, so either process can bootstrap it. Permanently failing messages end up on the dead-letter queue for operator inspection; transient failures are redelivered with the broker's default behavior.

Rejecting a message raises its x-death count; exceeding the configured threshold routes it to dead-letter.

Idempotency

IIdempotencyService / RedisIdempotencyService guards against duplicate processing: Redis records the id of each in-flight / recently completed job for a short TTL, so a broker redelivery after a consumer crash cannot double-count ingestion. Idempotency is keyed by job payload, not delivery tag, so it also covers publisher retries.

Periodic work

ExpiredProjectsCleanupService runs on a schedule and purges projects whose LifetimeExpiresAtUtc has passed and AutoExtendLifetimeOnActivity = false. This keeps tenant storage bounded for trial / time-boxed workspaces.

Publishing a new job type

1. Add a DTO to Harpyx.Application.DTOs.
2. Expose a producer method on IJobQueue and implement it in RabbitMqJobQueue (routing key + payload serialization).
3. Add a consumer in the Worker — either extend JobConsumerService (if it belongs in the ingestion pipeline) or create a new BackgroundService.
4. Persist the executing Job row, record audit events on start/end/error, increment the appropriate OTel counter.

Keep producer and consumer in the same queue topology declaration so partial deployments don't corrupt the broker state.