# Configuring Celery Queues for Spatial Workloads

A single default Celery queue that mixes a multi-minute polygon reprojection behind a sub-second metadata lookup guarantees head-of-line blocking: the cheap task waits on the expensive one, prefetch hoards messages a busy worker cannot start, and a runaway tiling job pins a core no time limit ever reclaims. This operation, part of Async Execution for Heavy Spatial Queries within the Federated Ownership & Routing Architecture, routes heavy spatial jobs — tiling, spatial joins, CRS reprojection — into dedicated queues with their own priority, prefetch, and hard time limits, and keeps every task idempotent by claiming a SHA-256(bbox+crs+params) key before doing any work, exactly the record sized in Tuning Redis Idempotency Cache TTL for Spatial Jobs.

Figure — spatial tasks are routed by kind into isolated queues, each drained by a worker pool tuned for its CPU or IO profile.

# Prerequisites

# Step-by-Step Implementation

Isolation is the whole design: each task kind names its queue, each queue gets a worker pool matched to its CPU or IO profile, and no heavy job can starve an interactive one. Every step is verifiable before the next.

# 1. Declare the queues and default routing

Define explicit queues rather than relying on the implicit celery default, and set the message priority ceiling so the priority queue can actually preempt.

from kombu import Queue
from celery import Celery

app = Celery("spatial", broker="amqp://mesh@rabbit:5672/mesh")
app.conf.task_queues = (
    Queue("cpu-heavy", routing_key="cpu.#", queue_arguments={"x-max-priority": 9}),
    Queue("io-light",  routing_key="io.#",  queue_arguments={"x-max-priority": 9}),
    Queue("priority",  routing_key="rt.#",  queue_arguments={"x-max-priority": 9}),
)
app.conf.task_default_queue = "io-light"
app.conf.task_default_priority = 3

Verify the queues were declared on the broker:

celery -A tasks inspect active_queues | grep -E 'cpu-heavy|io-light|priority'

# 2. Route each spatial task to its queue

task_routes maps task names to queues so callers never hard-code a queue. Heavy geometry work goes to cpu-heavy; catalog and metadata work goes to io-light.

app.conf.task_routes = {
    "spatial.tiling.*":       {"queue": "cpu-heavy", "priority": 5},
    "spatial.join.*":         {"queue": "cpu-heavy", "priority": 5},
    "spatial.reproject.*":    {"queue": "cpu-heavy", "priority": 5},
    "spatial.catalog.*":      {"queue": "io-light",  "priority": 3},
    "spatial.metadata.*":     {"queue": "io-light",  "priority": 3},
    "spatial.interactive.*":  {"queue": "priority",  "priority": 9},
}

Verify a submitted task lands on the expected queue:

python3 -c "from tasks import build_tiles; print(build_tiles.apply_async(kwargs={'bbox':[-74.5,40.1,-73.9,40.9],'crs':'EPSG:4326'}).id)"
celery -A tasks inspect reserved | grep cpu-heavy

# 3. Make heavy tasks idempotent and bounded

Every heavy task claims its SHA-256(bbox+crs+params) key before doing work, so a redelivery after a worker crash is a no-op. time_limit hard-kills a runaway; acks_late with reject_on_worker_lost requeues an interrupted job for a clean redelivery.

import redis, os
r = redis.from_url(os.environ["REDIS_URL"])

@app.task(bind=True, acks_late=True, reject_on_worker_lost=True,
          time_limit=600, soft_time_limit=540, max_retries=3)
def build_tiles(self, bbox, crs, **params):
    key = idem_key(bbox, crs, params)                 # from the TTL page
    if not r.set(key, self.request.id, nx=True, ex=1575):
        return {"status": "deduplicated", "key": key} # already done / in-flight
    return run_tiling(bbox, crs, params)               # EPSG:4326 in, MVT out

Verify a replayed identical task is deduplicated, not recomputed:

celery -A tasks call spatial.tiling.build_tiles \
  --kwargs '{"bbox":[-74.5,40.1,-73.9,40.9],"crs":"EPSG:4326"}'
redis-cli -u "$REDIS_URL" TTL "idem:$HASH"   # positive TTL confirms the claim

# 4. Tune prefetch and pool per queue class

CPU-bound workers must run --prefetch-multiplier=1 so a worker holds exactly one heavy message and the rest stay fair-dispatchable; IO-bound workers can prefetch more. Run one deployment per queue.

# CPU-heavy: solo pool, prefetch 1, one process reclaimed at the time limit
celery -A tasks worker -Q cpu-heavy --pool=prefork --concurrency=4 \
  --prefetch-multiplier=1 --max-tasks-per-child=50 -n cpu@%h

# IO-light: threaded, high prefetch for cheap catalog work
celery -A tasks worker -Q io-light --pool=threads --concurrency=16 \
  --prefetch-multiplier=8 -n io@%h

Verify the running pool and prefetch match the intent:

celery -A tasks inspect stats | jq '.[] | {pool: .pool.implementation, prefetch: .prefetch_count}'

# 5. Confirm isolation under load

Submit a burst of heavy jobs and one interactive job; the interactive job must complete on the priority queue without waiting on the heavy backlog.

for i in $(seq 1 20); do celery -A tasks call spatial.join.run --kwargs "{\"bbox\":[-74.5,40.1,-73.9,40.9],\"crs\":\"EPSG:4326\",\"n\":$i}"; done
celery -A tasks call spatial.interactive.lookup --kwargs '{"id":"abc"}'
celery -A tasks inspect active | jq '.[] | length'

# Configuration Reference

# Common Failure Modes & Fixes

• Symptom: cheap metadata calls stall behind tiling jobs. Root cause: both kinds share one queue, or the CPU worker prefetched a batch. Fix: route metadata to io-light via task_routes and set --prefetch-multiplier=1 on the CPU worker.
• Symptom: a job runs to completion twice after a worker restart. Root cause: acks_late requeued the task but it was not idempotent. Fix: claim the SHA-256 key before any side effect, as in Step 3, and confirm the dedup branch returns early.
• Symptom: worker RSS grows until the pod is OOM-killed. Root cause: GDAL/GEOS native allocations accumulate across tasks. Fix: set --max-tasks-per-child=50 and cap MAX_HEAP_MB so each child recycles before it bloats.
• Symptom: the priority queue never preempts. Root cause: the broker queue was declared without x-max-priority. Fix: redeclare with queue_arguments={"x-max-priority": 9} — priority is set at queue creation and cannot be added later.
• Symptom: interrupted heavy tasks vanish silently. Root cause: acks_early (the default) acked before completion, so a crash lost the job. Fix: set acks_late=True with reject_on_worker_lost=True so an interrupted task is redelivered.

# FAQ

# Related

• Async Execution for Heavy Spatial Queries — parent reference for the async worker-pool pattern.
• Tuning Redis Idempotency Cache TTL for Spatial Jobs — how the SHA-256 claim key each task writes is sized.
• Optimizing Async Execution for Spatial Joins — the join workload that fills the cpu-heavy queue.
• Building Fallback Chains for Routing APIs — where synchronous requests offload heavy work to these queues.
• Federated Ownership & Routing Architecture — up one level to the routing architecture overview.