Setting Up Async Lineage Logs with Celery

Part of: Asynchronous Logging Strategies

Setting up async lineage logs with Celery requires decoupling provenance capture from your primary geospatial processing pipeline, routing metadata writes through a message broker, and implementing idempotent worker tasks that persist transformation records without blocking I/O. The core pattern emits a lightweight JSON payload containing dataset UUIDs, operation signatures, timestamps, and processor identities, then delegates persistence to a dedicated Celery worker. This architecture guarantees strict chain-of-custody tracking while keeping raster tiling, vector topology validation, and coordinate transformations running at peak throughput.

Why Asynchronous Provenance Capture Matters for GIS

Heavy spatial operations routinely exhaust database connection pools and saturate network I/O when audit writes execute synchronously. A single INSERT into a lineage table during a 50 GB raster reprojection can stall the entire ETL thread. Adopting Asynchronous Logging Strategies shifts audit persistence to background workers, isolating compliance overhead from compute-heavy steps.

Celery’s distributed queue automatically handles:

  • Retry backoff for transient database or broker outages
  • Dead-letter routing for malformed payloads that exceed retry limits
  • Rate limiting to prevent write storms during batch processing peaks
  • Late acknowledgment (task_acks_late=True) to improve delivery reliability on worker crashes

For agency tech teams, this aligns directly with federal data governance mandates requiring immutable, tamper-evident audit trails. By queuing lineage events, you maintain predictable processing SLAs while preserving PROV-compliant metadata.

Broker Architecture & Compliance Hardening

Broker selection dictates delivery guarantees and operational complexity:

  • Redis 7+: Optimal for low-latency spatial ETL. Use redis.conf persistence (appendonly yes) and TLS to meet state-level data sovereignty requirements.
  • RabbitMQ 4.0+: Preferred for compliance-heavy workflows. Provides publisher confirms, message TTL, and dead-letter exchanges out of the box.

Configure broker persistence, enforce TLS encryption, and tune visibility timeouts to prevent premature message redelivery during long-running geospatial jobs. Production hardening should follow the official Celery Configuration Guide, specifically enabling broker_use_ssl, task_acks_late=True, and task_reject_on_worker_lost=True. These settings integrate cleanly into broader Python Automation & Pipeline Integration architectures where multiple microservices share audit infrastructure.

Where the record can be lost in a Celery lineage path A geoprocessing task publishes to the broker; a separate lineage worker consumes and persists. The acknowledgement point after persistence is what makes worker loss survivable. Geoprocessing publishes, returns immediately Broker durable queue, persistent delivery Lineage worker validates, upserts by content key Store append-only ack ONLY after the write commits acks_late=True moves the acknowledgement after the task body — a killed worker redelivers rather than drops. Celery's default acknowledges on receipt, which loses every in-flight lineage record when a worker dies.

The default acknowledgement behaviour is the single most important thing to change, and it is easy to miss because it costs nothing in normal operation. Celery acknowledges a message when a worker receives it, not when the task finishes. A worker killed mid-task — by an OOM killer, a spot reclaim, or a rolling deploy — therefore loses that lineage record permanently, and nothing anywhere reports a loss. Setting acks_late=True moves the acknowledgement to after the task body returns, so an interrupted task is redelivered instead.

Late acknowledgement makes redelivery normal rather than exceptional, which is why the worker must be idempotent. Upserting on a content-derived key turns a duplicate delivery into a no-op, and that pairing — late acks plus idempotent write — is what makes the whole path survivable. Adopting one without the other either loses records or duplicates them.

Production Implementation

The following implementation demonstrates a production-ready Celery task for geospatial lineage capture. It uses deterministic SHA-256 hashing for idempotency, exponential backoff for transient failures, and PostgreSQL ON CONFLICT DO NOTHING to prevent duplicate audit entries during network partitions.

# lineage_tasks.py
import hashlib
import json
import logging
from datetime import datetime, timezone
from typing import Dict, Any

from celery import Celery
from celery.utils.log import get_task_logger
from sqlalchemy import create_engine, text

app = Celery(
    "geospatial_lineage",
    broker="redis://localhost:6379/1",
    backend="redis://localhost:6379/2",
    include=["lineage_tasks"]
)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    broker_transport_options={"visibility_timeout": 3600},
)

logger = get_task_logger(__name__)

@app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    acks_late=True
)
def record_lineage_event(self, payload: Dict[str, Any], db_uri: str) -> str:
    """
    Persist geospatial lineage metadata with idempotency guarantees.
    Uses PostgreSQL ON CONFLICT DO NOTHING to handle duplicate deliveries safely.
    """
    # 1. Generate deterministic idempotency key from sorted payload JSON
    payload_bytes = json.dumps(payload, sort_keys=True).encode("utf-8")
    event_id = hashlib.sha256(payload_bytes).hexdigest()

    # 2. Prepare SQL with ON CONFLICT DO NOTHING
    upsert_sql = text("""
        INSERT INTO lineage_events (
            event_id, dataset_uuid, operation, crs_from, crs_to,
            processor_id, timestamp, payload_json
        ) VALUES (
            :event_id, :dataset_uuid, :operation, :crs_from, :crs_to,
            :processor_id, :timestamp, :payload_json
        )
        ON CONFLICT (event_id) DO NOTHING;
    """)

    params = {
        "event_id": event_id,
        "dataset_uuid": payload.get("dataset_uuid"),
        "operation": payload.get("operation"),
        "crs_from": payload.get("crs_from"),
        "crs_to": payload.get("crs_to"),
        "processor_id": payload.get("processor_id"),
        "timestamp": datetime.now(timezone.utc),
        "payload_json": json.dumps(payload)
    }

    try:
        engine = create_engine(db_uri, pool_pre_ping=True, pool_size=5)
        with engine.connect() as conn:
            conn.execute(upsert_sql, params)
            conn.commit()
        logger.info("Lineage event persisted: %s", event_id)
        return f"SUCCESS:{event_id}"
    except Exception as exc:
        logger.warning("Lineage write failed, retrying: %s", exc)
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

Key Implementation Details

  • Idempotency: The event_id is derived from a sorted JSON hash. Identical payloads produce identical keys, making ON CONFLICT DO NOTHING safe for at-least-once delivery systems.
  • Retry Strategy: Exponential backoff (2^retries * 60s) prevents database thrashing during partial outages. task_acks_late=True ensures the broker requeues the task if the worker crashes mid-write.
  • Connection Safety: pool_pre_ping=True validates connections before execution, avoiding stale pool errors common in long-running GIS workers.
  • Schema Requirements: The target table requires a unique constraint on event_id for the ON CONFLICT clause to function. Refer to PostgreSQL’s official INSERT documentation for constraint syntax.
Separating compliance-critical events from routine updates Two queues fed from the same producer, each with its own worker pool and retry policy, so a backlog of routine updates cannot delay integrity-critical records. Producer routes by event type queue: lineage.critical seal failures, integrity queue: lineage.bulk routine step records dedicated pool · fast retry alarm on any backlog shared pool · batched backlog tolerated in minutes One queue means a nightly bulk load delays the integrity alert you most needed to see promptly.

Queue separation matters more here than in ordinary task processing because the two classes have genuinely different urgency. A routine step record arriving three minutes late costs nothing. An integrity-seal failure arriving three minutes late — behind fifty thousand queued bulk records from an overnight reprocessing job — is an alert that fires after the bad data has already propagated downstream. Route by event type at publish time, give the critical queue its own worker pool, and alarm on any sustained depth at all rather than on a threshold.

Operationalizing the Pipeline

Deploy workers with concurrency tuned to your database connection limits. For PostgreSQL, max_connections minus reserved overhead dictates safe worker counts. Use celery -A lineage_tasks worker --concurrency=4 --loglevel=info for baseline deployments, scaling horizontally via Docker or Kubernetes when batch volumes exceed 10k events/hour.

Monitor dead-letter queues and retry rates using Celery Flower or Prometheus exporters. High retry counts typically indicate broker TLS misconfigurations or database connection pool exhaustion. For compliance audits, export lineage tables to W3C PROV-JSON format using the PROV Data Model specification, ensuring interoperability with federal metadata catalogs.

By isolating audit persistence from spatial compute, you eliminate I/O contention, guarantee immutable provenance records, and maintain predictable pipeline throughput across government and enterprise GIS environments.

Configuration Reference

Setting Value Why
task_acks_late True Acknowledge after the body runs, so a killed worker redelivers
worker_prefetch_multiplier 1 Limits redelivery to the one in-flight message
task_reject_on_worker_lost True Requeues rather than silently discarding on worker death
task_serializer "json" Readable by non-Python consumers; avoids pickle on a shared broker
broker_transport_options.visibility_timeout > longest task duration Prevents premature redelivery of a still-running task
task_default_retry_delay / retry_backoff exponential with jitter Spreads correlated retries after a store outage
task_annotations.rate_limit tuned per queue Keeps bulk ingestion from starving the critical queue

visibility_timeout is the one that produces the most confusing incidents when left at its default. If a lineage task takes longer than the timeout — which happens when the store is slow, not when the task is complex — the broker assumes the worker died and hands the same message to a second worker. Both then write, and without an idempotent key you get duplicates that look like the pipeline ran twice. Set it comfortably above your p99 task duration and revisit it whenever the store’s latency profile changes.

Verification

Three failure injections that prove the configuration Each test names the fault to inject, the correct observable outcome, and the misconfiguration it exposes when the outcome differs. INJECT CORRECT OUTCOME IF NOT, YOU LACK… SIGKILL a worker mid-task Record appears after redelivery acks_late=True Stop the broker for 60s Pipeline keeps running; records land afterwards a local spill buffer Replay the same message Row count unchanged (upsert, not insert) a content-derived key

Run all three against a staging broker before trusting the deployment, and run the first one again after any Celery or broker upgrade — acknowledgement semantics are exactly the kind of behaviour that shifts between major versions. The second test is the one that most often surprises teams: publishing to an unreachable broker raises by default, which propagates back into the geoprocessing task and fails the pipeline for a reason that has nothing to do with the data.

Gotchas & Edge Cases

  • task_acks_late alone is not enough. Pair it with worker_prefetch_multiplier = 1, or a worker holding a large prefetch buffer will redeliver a batch of messages on death and reprocess far more than the one that was in flight.
  • Result backends are not lineage stores. Celery’s result backend expires entries and is designed for task return values, not for audit records. Write lineage to your own append-only store and leave the backend disabled or short-lived.
  • Serialisation defaults leak. Configure task_serializer = "json" explicitly. A pickle-serialised payload is both a security concern on a shared broker and unreadable to any non-Python consumer that later needs to read the queue.
  • Broker TLS failures look like retry storms. High retry counts with no error detail almost always trace to certificate or hostname verification rather than to load. Check the worker log at debug level before scaling anything.