Architecting Zero-Drop Asynchronous Document Pipelines with Celery, Redis, and FastAPI

The Concurrency Dilemma in Document Ingestion

When ingesting enterprise contracts and multi-page invoices, standard synchronous HTTP web servers quickly fail. A single 40-page PDF with embedded high-resolution scans can consume 8 seconds of intensive CPU time for image deskewing, binarization, and deep learning layout extraction. In a synchronous request-response model, a sudden burst of 50 concurrent uploads exhausts the server's thread pool, leading to 504 Gateway Timeouts and severed connections.

Deconstructing the Asynchronous Worker Topology

To achieve predictable sub-millisecond API response times regardless of document size, we decouple HTTP ingestion from computational inference using an asynchronous worker queue topology.

Client Upload (PDF)
        │
        ▼
[ FastAPI ASGI Gateway ]  ──>  Ack 202 Accepted + UUID
        │
        ▼  Push Job
[ Redis Message Broker ]
   ├── Task Queue (Normal Priority)
   └── High-Priority Queue (SLA Contracts)
        │
        ▼  Prefetch & Process
[ Distributed Celery Workers (Docker / ECS) ]
   ├── Page Slicing & Deskew (OpenCV)
   ├── Layout Segmentation (PyTorch)
   └── Text Serialization (PaddleOCR)
        │
        ▼  Store Validated Schema
[ Redis Result Store / PostgreSQL ]  ──>  Client Webhook Ping

1. The FastAPI Ingestion Gateway

The FastAPI gateway performs strict Pydantic v2 metadata validation, streams the uploaded binary to private S3-compatible object storage, and pushes an immutable task descriptor to Redis before returning an immediate HTTP 202 Accepted with a tracking UUID.

@app.post("/v1/documents/ingest", status_code=status.HTTP_202_ACCEPTED)
async def ingest_document(
    file: UploadFile = File(...),
    webhook_url: Optional[HttpUrl] = None,
):
    document_id = str(uuid.uuid4())
    s3_key = await upload_to_storage(file, document_id)
    
    task = extract_document_task.apply_async(
        kwargs={"document_id": document_id, "s3_key": s3_key, "webhook_url": str(webhook_url)},
        queue="documents_high_priority" if file.size < 5_000_000 else "documents_batch",
    )
    return {"status": "queued", "document_id": document_id, "task_id": task.id}

2. Preventing Worker Starvation with Celery Prefetch Tuning

By default, Celery prefetches multiple tasks per worker process (worker_prefetch_multiplier = 4). If one worker pulls a massive 80-page document alongside three quick 1-page receipts, the quick receipts sit idle waiting behind the heavy task while other workers starve.

Setting worker_prefetch_multiplier = 1 combined with task_acks_late = True guarantees that workers only reserve one document at a time and do not acknowledge completion until the structured JSON is written to the persistent database. If a worker pod gets killed mid-process by an EC2 spot interruption, Redis immediately re-queues the task for another worker without data loss.

3. Automated Dead-Letter Queue (DLQ) Recovery

When a corrupted PDF or malicious byte stream triggers an unexpected decoding crash, the task must not loop infinitely. We configure a dead-letter queue that moves failed payloads into a quarantined inspection store after three exponential backoff retries, alerting on-call engineers via Slack without blocking the primary ingestion queue.

Key Results in Production

• Average API Response Time: 18ms for initial upload acknowledgment.
• Queue Drop Rate: 0.00% across 2.4 million processed documents.
• Elastic Scale: Worker nodes auto-scale from 4 to 32 instances within 90 seconds of queue depth exceeding 200 pending tasks.