Designing Celery Task Queues for Lease Portfolio Batches

Once lease extraction is distributed, the hard decisions stop being “which broker” and become “what is a task, and which queue does it belong to.” This page resolves the topology questions that surface the first time you abstract a whole portfolio on Async Batch Processing — how finely to slice the work, how to fan a portfolio out and fan the results back in, how to keep one landlord’s overnight bulk load from starving another tenant’s single urgent renewal, and how to end the run with a single trustworthy answer rather than thousands of scattered task states. Getting the shape wrong shows up as broker floods, orphaned subtasks, and portfolio jobs that never quite report “done.”

Architectural context

Task and queue design sits one layer above the durability concerns of a Celery deployment, inside the broader Parsing & Extraction Workflows domain. Upstream, the OCR preprocessing workflows clean scanned pages; this layer decides how those documents are grouped into tasks and routed across queues; downstream, validated abstractions flow through field mapping strategies into the canonical lease data models, while low-confidence results divert through fallback routing logic to review. This page is deliberately narrow: it is about orchestration shape, not connection health. Broker memory policy, acks_late durability, result-backend isolation, and content-hash idempotency are the sibling’s territory — configure those first in Scaling Async Lease Parsing with Celery and Redis, then use the topology here on top of them.

Choosing task granularity

The first decision is what a single Celery task represents. Finer granularity gives you per-lease retry, per-lease idempotency, and clean isolation when one document is corrupt; coarser granularity slashes the number of messages the broker must hold and dispatch. For a 40,000-lease portfolio the difference is 40,000 queued messages versus a few hundred.

The default is one task per document — it is the granularity at which the sibling’s content-hash idempotency and error handling & retry logic work cleanly, because retrying a task re-processes exactly one lease. Reach for chunking only when message count itself becomes the bottleneck, and keep the chunk body loop-idempotent so a mid-chunk crash that re-runs the whole chunk does not double-post rent.

Celery canvas primitives

Celery’s canvas gives you the composition primitives to express a portfolio as a workflow rather than a loop of .delay() calls. Two matter most here: group for the parallel fan-out and chord for the fan-in that produces a single portfolio-level result.

A portfolio abstraction run is almost always a chord: the header is a group of extract tasks over the portfolio’s document keys, and the body is a callback that receives every subtask’s return value and folds it into one report. That is exactly the fan-out/fan-in the diagram shows.

Recommended implementation

The core pattern is a chord. The header fans one extract_one task out per document key; the callback, aggregate_portfolio, runs once — and only once — after the last subtask completes, receiving the list of per-lease results as its first argument. Every result is re-validated with a pydantic v2 model on the way in, so a malformed subtask return can never silently corrupt the tally.

from decimal import Decimal
from typing import Any

from celery import chord, group
from celery.utils.log import get_task_logger
from pydantic import BaseModel, Field

logger = get_task_logger(__name__)


class LeaseResult(BaseModel):
    doc_key: str
    tenant_id: str
    status: str  # "committed" | "review" | "failed"
    base_rent: Decimal | None = None
    confidence: float = Field(ge=0.0, le=1.0, default=0.0)


class PortfolioReport(BaseModel):
    portfolio_id: str
    total: int
    committed: int
    routed_to_review: int
    failed: int
    committed_rent: Decimal


@app.task(bind=True, name="lease_parser.tasks.extract_one", acks_late=True, max_retries=4)
def extract_one(self, doc_key: str, tenant_id: str) -> dict[str, Any]:
    """Abstract one lease. Durability (acks_late, retries, idempotency) is
    configured per the Celery + Redis guide; here we only shape the result."""
    fields = _run_parser_engine(doc_key)  # OCR + clause extraction + normalization
    result = LeaseResult(
        doc_key=doc_key,
        tenant_id=tenant_id,
        status="committed" if fields["confidence"] >= 0.80 else "review",
        base_rent=Decimal(str(fields["base_rent"])),
        confidence=fields["confidence"],
    )
    return result.model_dump(mode="json")


@app.task(name="lease_parser.tasks.aggregate_portfolio")
def aggregate_portfolio(results: list[dict[str, Any]], portfolio_id: str) -> dict[str, Any]:
    """Chord callback: fold every per-lease result into one portfolio report."""
    parsed = [LeaseResult(**r) for r in results]
    committed = [r for r in parsed if r.status == "committed"]
    report = PortfolioReport(
        portfolio_id=portfolio_id,
        total=len(parsed),
        committed=len(committed),
        routed_to_review=sum(1 for r in parsed if r.status == "review"),
        failed=sum(1 for r in parsed if r.status == "failed"),
        committed_rent=sum((r.base_rent or Decimal("0")) for r in committed),
    )
    logger.info("Portfolio %s complete: %s", portfolio_id, report.model_dump(mode="json"))
    return report.model_dump(mode="json")


def dispatch_portfolio(portfolio_id: str, doc_keys: list[str], tenant_id: str, priority: int = 5):
    """Fan a portfolio out as a chord; the callback fires once, at the end."""
    header = group(
        extract_one.s(key, tenant_id).set(
            queue=f"tenant.{tenant_id}.extraction",
            priority=priority,  # 0 = highest; bulk loads run lower
        )
        for key in doc_keys
    )
    callback = aggregate_portfolio.s(portfolio_id=portfolio_id)
    return chord(header)(callback)

The per-lease task returns a plain dict so the chord’s result backend stays small; the callback re-hydrates each into a LeaseResult before tallying. Money is summed as Decimal, never float, so the reported committed_rent is exact.

Per-tenant and priority routing

SLA isolation is a routing problem. Give each stage its own queue so slow OCR never blocks cheap normalization, and give latency-sensitive work a priority lane so a single urgent renewal is not stuck behind a 40,000-lease overnight load. Routing keys map task names and priorities onto those queues declaratively.

from kombu import Queue

app.conf.task_queues = (
    Queue("lease_extraction", routing_key="extract.batch"),
    Queue("lease_ocr", routing_key="ocr.fallback"),
    Queue("lease_normalization", routing_key="normalize.fields"),
    Queue("lease_priority", routing_key="priority.#"),   # urgent single leases
)
app.conf.task_default_queue = "lease_extraction"
app.conf.task_routes = {
    "lease_parser.tasks.extract_one": {"queue": "lease_extraction", "routing_key": "extract.batch"},
    "lease_parser.tasks.run_ocr_fallback": {"queue": "lease_ocr", "routing_key": "ocr.fallback"},
    "lease_parser.tasks.normalize_financials": {"queue": "lease_normalization", "routing_key": "normalize.fields"},
}
# Dedicate worker fleets: many small workers on lease_normalization,
# fewer memory-heavy workers on lease_ocr, a reserved pool on lease_priority.

Dedicating a queue per tenant (tenant.<id>.extraction) or reserving lease_priority for interactive requests lets you scale and rate-limit each independently — the routing key on extract_one.s(...).set(queue=...) above sends bulk portfolio work down the tenant’s own lane, keeping one landlord’s backlog off everyone else’s critical path.

Edge cases specific to lease portfolios

• Giant portfolios flooding the broker. A 40,000-lease group is 40,000 queued messages. Wrap the fan-out in extract_batch.chunks(zip(doc_keys, tenants), 200) so the broker holds a few hundred envelopes instead, and make the per-chunk loop idempotent so a re-delivered chunk re-posts nothing.
• The one straggler lease. A chord blocks its callback until every member finishes, so a single 90-page scanned lease can hold the entire portfolio report hostage. Set a soft_time_limit on extract_one and route timed-out documents to review, so the straggler resolves to a failed/review result rather than stalling the fan-in forever.
• Partial batch failure. If any header task errors out, the chord callback never fires by default and the whole portfolio silently hangs. Attach an error callback — chord(header)(callback.on_error(alert_ops.s())) — and have extract_one catch permanent failures and return a status="failed" result instead of raising, so one corrupt lease degrades the report by one row rather than killing it.
• Per-tenant starvation. Without dedicated queues, a tenant who dumps a full portfolio monopolizes every worker. Per-tenant queues plus message priority keep a small tenant’s urgent renewal moving; cap bulk concurrency so no single tenant claims the whole fleet.
• Result-backend bloat. A large group writes one result row per subtask, and returning full abstractions balloons the backend. Return compact dicts (keys, status, confidence, rent) from extract_one, set result_expires so finished portfolios are reaped, and persist the canonical record to the database — not the Celery backend.

When to escalate

• Queue depth outruns workers. When any queue’s backlog exceeds worker_concurrency * 10 for a sustained window, the topology is sound but under-provisioned — trigger autoscaling on that specific queue (a per-queue Kubernetes HPA or autoscaling worker group) before SLAs slip.
• A lease exhausts its retries. Repeated permanent failures do not belong in the portfolio’s happy path; route them to dedicated handling per dead-letter queue handling for failed lease extractions so the completion report stays honest and ops can triage offline.
• Confidence stays low after fallback. When aggregate confidence holds below 0.80 even after the OCR fallback queue, divert through fallback routing logic to human review rather than committing a guessed base rent.

Frequently asked questions

When should I use a chord instead of just a group of tasks? Use a chord whenever you need a single result after the whole portfolio finishes — a completion report, a “notify the property manager” step, or a downstream sync trigger. A bare group fans work out but gives you no one place that fires exactly once at the end; the chord’s callback is that place.

How do I keep one tenant’s bulk load from starving another tenant? Give each tenant (or at least each priority tier) its own queue and set message priority on dispatched tasks, then dedicate or cap worker fleets per queue. Bulk portfolio work runs at low priority on the tenant’s own lane, so a small tenant’s urgent single lease is routed to a reserved priority queue and never waits behind a 40,000-lease backlog.

One slow lease is holding up my whole portfolio report — why? A chord callback waits for every header task to complete, so a single straggler blocks the fan-in. Put a soft_time_limit on the extraction task and have it return a review or failed result on timeout instead of running unbounded, so the straggler resolves to a row in the report rather than freezing it.

How fine should my task granularity be for a large portfolio? Default to one task per document for clean per-lease retry and idempotency. Switch to chunked batches only when the raw message count floods the broker — then loop over the chunk inside the task and keep that loop idempotent so a re-delivered chunk re-posts nothing.

Related

• Scaling Async Lease Parsing with Celery and Redis — the broker config, acks_late durability, and content-hash idempotency this topology assumes underneath it.
• Idempotency Patterns for S3 Event-Triggered Lease Ingestion — making the per-lease tasks these queues carry safe to re-deliver and re-dispatch.
• Dead-Letter Queue Handling for Failed Lease Extractions — where the failed rows in a portfolio report go for triage instead of vanishing.
• Async Batch Processing — the parent stage this fan-out/fan-in topology plugs into.

← Back to Async Batch Processing

← Back to Async Batch Processing