What I’d harden next in DocFlow

DocFlow is my reference implementation of an asynchronous document pipeline: upload a PDF, DOCX or image, ask for a summary, invoice fields or contract metadata, and get an answer back later. FastAPI takes the upload, MinIO stores the file, PostgreSQL tracks every request, and Celery workers on Redis do the OCR and model calls.

It already handles the failures you can see. This post is about the ones you can’t: the cases where a request stops moving and nothing ever tells you.

What already holds

• Failures reach Celery’s retry policy. The task re-raises, so autoretry_for with backoff gets up to three more attempts. Errors that retrying can’t fix (unreadable input, a request the model provider rejects) fail at once instead.
• Every attempt is a row. A processing_job records the attempt number, status and a failure reason that is safe to show the user.
• Cancellation can’t race the worker. The worker locks the request row before starting, so a cancel either lands first or sees PROCESSING.
• One result per request, enforced by a unique constraint, and invoice and contract output is parsed into Pydantic models.
• The owner gets an email when a request completes or finally fails, and worker logs carry IDs, never document text.

All of that assumes the worker lives long enough to record what happened. The gaps below are the cases where it doesn’t.

1. A worker that dies mid-job

By default Celery acknowledges a message when the worker receives it, before the task runs. DocFlow keeps that default:

@celery_app.task(
    bind=True,
    autoretry_for=(Exception,),
    dont_autoretry_for=PERMANENT_ERRORS,
    retry_backoff=True,
    retry_kwargs={"max_retries": 3},
)
def start_processing(self, processing_request_id: int): ...

If the worker is killed mid-task (out of memory on a large scan, a deploy, a node restart), the message is already gone. No exception runs, so no retry, no failure status and no email. The request says PROCESSING forever, and the user watches a spinner that never ends.

The textbook fix is task_acks_late with task_reject_on_worker_lost, so the broker redelivers unacknowledged work. Turning it on alone would make things worse, though, because the task is not safe to run twice. Its only guard skips cancelled requests:

if processing_request.status == Processing_status.CANCELLED:
    return

Suppose a worker saves the result and dies before acknowledging. The redelivered task runs the OCR and the paid model call again, then hits the unique constraint on the result. An IntegrityError isn’t on the permanent-error list, so the handler sets the request back to QUEUED and retries. After three more attempts, a request that had succeeded ends up FAILED, and the owner gets an email saying so.

So the first change is in the task, not the broker. A task should claim its request in one atomic statement and do nothing if it can’t:

claimed = db_session.execute(
    update(Processing_Request)
    .where(Processing_Request.id == processing_request_id)
    .where(Processing_Request.status == Processing_status.QUEUED)
    .values(status=Processing_status.PROCESSING)
    .returning(Processing_Request.id)
).scalar_one_or_none()
if claimed is None:
    return  # finished, cancelled, or another worker has it

Retries still work, because the failure handler already puts a retryable request back to QUEUED before re-raising. Duplicate messages become harmless: whichever arrives second finds nothing to claim.

Recovering from a dead worker then belongs to the database, not the broker. With Redis, Celery redelivers an unacknowledged message only after the visibility timeout (an hour by default), and a redelivered task can’t tell a dead worker from a slow one anyway. A worker can be recognised as dead from its time limit, which is the next gap.

2. No bound on time or memory

When a PDF has no text layer, DocFlow rasterises every page at once and runs OCR over the lot:

images = convert_from_bytes(file_stream.getvalue())
return "\n".join(
    pytesseract.image_to_string(ImageOps.grayscale(page)) for page in images
).strip()

The upload limit caps the file size, not the page count, and a few megabytes of scanned PDF can hold hundreds of pages. The model call has a timeout; the task as a whole has none. The newer document Q&A feature already caps pages (DOCUMENT_QA_MAX_PAGES); the processing path should too:

• A page cap checked before any OCR, failing with a message the user can act on.
• Page-by-page conversion (first_page/last_page), so memory holds one page image at a time instead of all of them.
• `soft_time_limit` and `time_limit` on the task. The soft limit raises inside the task so it can record a clear failure; the hard limit kills it if that doesn’t work. A document that timed out once will time out again, so the soft limit belongs on the permanent-error list.

The hard limit is also the number the sweeper below needs. A job still RUNNING well past it means the worker is gone.

3. The gap between commit and enqueue

The upload handler commits the request as QUEUED, then publishes the task:

processing_request_object.status = Processing_status.QUEUED
await db_session.commit()

try:
    start_processing.delay(processing_request_id=processing_request_object.id)
except Exception:
    await db_session.delete(document_object)
    await db_session.delete(file_object)
    await db_session.commit()
    raise

If Redis refuses the message, the compensation cleans up. If the API process dies between those two lines, nothing does: the database promises a job that the queue has never heard of. It’s the classic dual-write problem, and a transactional outbox is the textbook answer.

Here a smaller tool covers this gap and the first one together: a sweeper, a Celery beat task that runs every few minutes.

• Requests PROCESSING whose latest job started longer ago than the hard time limit (plus a margin): mark that job failed with “worker lost”, then re-queue the request, or fail it if it has used its attempts.
• Requests QUEUED for longer than the longest retry backoff: enqueue them again. If the original message was only waiting in a backlog, the claim makes the extra one a no-op.
• Every action logged with the request ID, so “how often do workers die?” becomes a number instead of a guess.

4. A retried upload is a second job

Uploads are slow by nature, and clients time out and retry. Today every POST creates a new file in MinIO, a new document, a new request and a new paid model call. Nothing breaks, but the user sees duplicates and pays twice.

The fix is an Idempotency-Key header. Store (user_id, key) under a unique constraint with the request it created and a hash of the uploaded bytes. A repeat with the same key and the same file gets the original response back; the same key with a different file is a 422. Keys can expire after a day; retries don’t take longer than that.

5. Clients still poll

The completion email is for people. Programmatic clients still poll the status endpoint and then fetch the result. That’s simple and works everywhere, but it adds latency and load in proportion to how long the jobs run.

I’d add webhooks for integrations: a registered URL, a payload signed with HMAC, and delivery as its own Celery task with retries. Delivery is at least once, so every event carries an ID the receiver can deduplicate on. For a browser that is waiting on a result, server-sent events would do, and DocFlow already streams them for chat.

Why this order

The first three are about the same promise: a request never sits in a state nobody will fix. That matters more than cost or convenience, because a stuck job is the one failure that generates no error, no alert and no email. The last two make the system cheaper and nicer to integrate with; they can wait until the first three are true.

DocFlow is open source. The code behind every snippet here is in the repository, and so is the work, as it lands.