Async Batch Processing Setup for Water Quality Compliance Pipelines

Async batch processing is the execution tier that lets a utility run compliance-grade validation over telemetry without ever stalling the live acquisition path. This topic sits within SCADA Data Ingestion & Time-Series Sync and is written for the municipal Python developers who build the automation, the SCADA operators who own the polling cadence, and the environmental compliance teams who sign the reports those batches feed. The core problem is one of decoupling: a treatment plant emits a continuous stream of readings that must never be delayed by a slow database write, a state-agency API throttle, or a cryptographic signing step, yet every one of those readings still has to pass deterministic Safe Drinking Water Act (SDWA) validation before it becomes a reported number. The pattern below routes discrete, time-bounded measurement windows through an asynchronous task queue so that historical backfill and real-time enrichment both run behind the ingestion edge — concurrent, retryable, and fully attributable — while the acquisition layer stays a strictly non-blocking producer.

Async batch processing end-to-end flow Both protocol streams converge on a broker that dispatches time-bounded batches to validation workers. Field sources (Modbus and OPC UA) send raw signals to an ingestion and normalization stage, which enqueues each time-bounded batch onto a message broker. The broker delivers batches to worker nodes, which run validation and compliance checks against themselves before persisting the validated result to an append-only compliance ledger. raw signals enqueue batch deliver batch persist result validate & compliance checks Field sources Modbus · OPC UA Ingestion & normalization Message broker Worker nodes validation Compliance ledger

Regulatory / Protocol Foundation

The asynchronous design is not an efficiency luxury; it is what makes the compliance arithmetic defensible. Under the SDWA and its implementing rules at 40 CFR Part 141, determinations for parameters such as turbidity, disinfectant residual, and disinfection byproducts are computed from time-averaged or percentile aggregations — a four-hour turbidity average, a locational running annual average, a 30-day mean residual. Those aggregations are only valid over a complete, correctly timestamped window, which means the processing tier must be able to hold a batch open until every reading that belongs to it has arrived, reprocess it if a late or corrected value appears, and prove afterward that the number it reported came from a specific, unaltered set of inputs. A synchronous pipeline that computes a window average inline with acquisition cannot satisfy any of those requirements without back-pressuring the sensors.

Two hard constraints therefore govern this tier. First, idempotent reprocessing: because a batch may be retried after a transient failure or re-queued when a corrected reading lands, running the same window twice must never produce a second compliance submission. Second, complete lineage: every result must trace back through the batch that produced it to the raw readings, their quality flags, and the rule version applied. Both constraints are satisfied by treating each batch as an immutable, content-addressed unit of work rather than a mutable job. Authoritative reporting requirements for these aggregations are published in the EPA Safe Drinking Water Act compliance resources; the pipeline’s job is to compute them without ever letting the execution mechanics compromise the underlying data.

Architecture & Design Decisions

The execution layer is built around a distributed message broker and an I/O-optimized worker pool. Readings enter already normalized: legacy RTU payloads arrive through Modbus TCP Parsing Workflows, which decode registers, apply scaling, and attach quality flags, while information-model sensor networks are handled by OPC UA Data Extraction, which preserves namespace metadata and sub-second timestamps. Both streams are aligned to a single monotonic UTC axis by Time-Series Alignment Strategies before any batch is cut, so the async tier never has to reason about protocol quirks or clock skew — it consumes one canonical reading shape carrying contaminant_id, method_code, sample_ts, quality_flag, and a 64-character source_hash.

The first design decision is where to draw the batch boundary. Batches are cut on the reporting window that regulation cares about — a rolling four-hour turbidity window, a calendar-day residual window — not on an arbitrary count, so that a completed batch maps one-to-one to a compliance calculation. The second decision is concurrency tuning: SCADA validation is I/O-bound, dominated by database round-trips, cryptographic signing, and external API calls, not CPU work. Worker concurrency is therefore sized to downstream connection limits and rate ceilings, not to core count, and a prefetch of one keeps any single worker from hoarding batches it cannot yet start. The third decision is priority routing: compliance-critical parameters such as lead, copper, and disinfectant residual travel on a dedicated high-priority queue ahead of historical backfill, so a large reprocessing job can never delay a live exceedance check. The concrete broker, worker, and result-backend wiring for this architecture is covered in depth in Configuring Celery for Async Water Quality Batches.

Async batch execution architecture Normalized readings enter a message broker that holds three named queues: compliance_priority for high-priority parameters, validation as the default, and backfill for bulk historical work. A horizontally-scaled worker pool configured with prefetch equal to one drains the queues in priority order, so a live exceedance check is never delayed by a large backfill job. Successfully committed batches are written to an append-only compliance ledger; batches that exhaust their retries are routed to an append-only dead-letter queue for operator disposition. Normalized readings Message broker compliance_priority high priority validation default backfill bulk Worker pool prefetch = 1 · I/O-bound worker worker worker Compliance ledger append-only Dead-letter queue append-only committed retries exhausted

Phase-by-Phase Implementation

Building the tier proceeds in four phases, each producing an artifact the next depends on: an immutable, content-addressed batch envelope; a tuned broker and worker pool; a retrying, idempotent validation task; and a dead-letter path that seals every failure into the audit trail.

Phase 1 — Define the batch envelope and its idempotency key

A batch is a frozen, time-bounded set of readings plus the metadata needed to reproduce its result. Its identity is a deterministic hash of its contents, so a re-queued or reconstructed batch carries the same key as its original and can be recognized as a duplicate before any work is committed.

Implementation steps:

  1. Freeze the batch envelope so no reading can be mutated after the window is cut.
  2. Derive the idempotency key from the sorted source_hash values of the readings plus the window bounds.
  3. Normalize the window bounds to UTC before hashing so identical windows always hash identically.
import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone


@dataclass(frozen=True)
class TelemetryBatch:
    """A time-bounded, immutable set of readings dispatched as one unit of work."""

    contaminant_id: str
    method_code: str
    window_start: datetime
    window_end: datetime
    primacy_id: str
    readings: tuple[dict, ...]

    def idempotency_key(self) -> str:
        """Content-address the batch so a re-queued window never double-submits."""
        payload = {
            "contaminant_id": self.contaminant_id,
            "method_code": self.method_code,
            "window_start": self.window_start.astimezone(timezone.utc).isoformat(),
            "window_end": self.window_end.astimezone(timezone.utc).isoformat(),
            "primacy_id": self.primacy_id,
            "readings": sorted(r["source_hash"] for r in self.readings),
        }
        blob = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(blob.encode("utf-8")).hexdigest()

Phase 2 — Configure the broker and the worker pool

The broker holds batches durably between the ingestion edge and the workers; the worker configuration decides how safely and how concurrently they drain. The two settings that matter most for compliance are late acknowledgement — a batch is acked only after it finishes, so a crashed worker returns its batch to the queue instead of losing it — and a prefetch of one, so priority routing is honored batch by batch rather than defeated by a worker that grabbed a dozen jobs up front.

Implementation steps:

  1. Point the broker and result backend at durable storage and enable late acks.
  2. Set prefetch to one and requeue on worker loss so no in-flight batch is dropped.
  3. Route compliance-critical parameters and bulk backfill onto separate named queues.
from celery import Celery

app = Celery(
    "wq_compliance",
    broker="redis://broker:6379/0",
    backend="redis://broker:6379/1",
)

app.conf.update(
    task_acks_late=True,             # ack only after the batch is committed
    task_reject_on_worker_lost=True, # requeue an in-flight batch if a worker dies
    worker_prefetch_multiplier=1,    # one batch per worker so priority routing holds
    task_default_queue="validation",
    task_routes={
        "tasks.validate_batch": {"queue": "compliance_priority"},
        "tasks.backfill_batch": {"queue": "backfill"},
    },
    # Redelivery window must exceed the slowest expected batch runtime.
    broker_transport_options={"visibility_timeout": 3600},
)

Phase 3 — Implement the idempotent validation task

The validation task is the unit of work. It reconstructs the batch, checks the result backend for its idempotency key, and short-circuits if the window was already committed. Transient failures — a dropped database connection, a throttled state API — trigger exponential backoff with jitter rather than an immediate retry, which prevents a broker-wide thundering herd when a dependency recovers.

Implementation steps:

  1. Reconstruct the frozen batch and compute its idempotency key.
  2. Skip immediately if the backend shows the window already committed.
  3. Retry transient errors with capped exponential backoff and jitter; commit exactly once on success.
from celery import shared_task
from celery.utils.log import get_task_logger

log = get_task_logger(__name__)


@shared_task(
    bind=True,
    acks_late=True,
    max_retries=5,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,       # 1s, 2s, 4s, 8s, 16s between attempts
    retry_backoff_max=600,
    retry_jitter=True,        # spread retries to avoid a thundering herd
)
def validate_batch(self, batch_payload: dict, result_backend, run_checks) -> dict:
    batch = TelemetryBatch(
        contaminant_id=batch_payload["contaminant_id"],
        method_code=batch_payload["method_code"],
        window_start=batch_payload["window_start"],
        window_end=batch_payload["window_end"],
        primacy_id=batch_payload["primacy_id"],
        readings=tuple(batch_payload["readings"]),
    )
    key = batch.idempotency_key()

    if result_backend.already_committed(key):
        log.info("batch %s already committed; skipping duplicate", key)
        return {"status": "duplicate", "key": key}

    result = run_checks(batch)              # deterministic compliance validation
    result_backend.commit(key, result)     # commit exactly once, keyed by content
    return {"status": "committed", "key": key}

Phase 4 — Seal failures into a dead-letter path

When a batch exhausts its retries, it must not vanish. The task’s failure handler writes the raw batch, the exhausted error, and its idempotency key to an append-only dead-letter store and alerts the compliance team. Because the store is append-only, a batch that is later corrected and re-queued produces a new record; the original failure remains on file for audit.

Implementation steps:

  1. On terminal failure, capture the raw payload, the error, and the idempotency key.
  2. Append the failure record to a write-once store — never overwrite an existing line.
  3. Emit a structured alert so the batch enters an operator disposition workflow.
import json


def route_to_dead_letter(batch_payload: dict, exc: Exception, ledger_path: str) -> dict:
    """Preserve a batch that exhausted its retries for operator disposition.

    The dead-letter store is append-only: a failed batch is never silently
    dropped and never overwritten, so every unresolved failure stays auditable.
    """
    record = {
        "idempotency_key": batch_payload["idempotency_key"],
        "contaminant_id": batch_payload["contaminant_id"],
        "error": repr(exc),
        "raw_payload": batch_payload,
    }
    with open(ledger_path, "a", encoding="utf-8") as dlq:
        dlq.write(json.dumps(record, sort_keys=True) + "\n")
    return record
Batch retry and dead-letter decision flow Failed batches retry with backoff; exhausted ones land in a dead-letter queue for manual review. A batch is processed and tested for success. On success it is persisted and acknowledged. On failure the flow checks whether retries remain: if they do, the worker waits an exponential backoff interval and re-processes the same batch; if they do not, the batch is routed to the dead-letter queue, which raises an alert requiring operator manual review. Process batch Success? Persist & ack Retries left? Wait exponential backoff Dead-letter queue Alert & manual review yes no yes no re-enqueue

Validation, Quality Flags & Edge Cases

Before a batch is committed, its readings are validated as a set, not merely one at a time. Only readings whose quality_flag is GOOD or INTERPOLATED may enter a compliance average; a SUSPECT or BAD value inside the window must either be excluded with its exclusion recorded, or, if too many are missing, force the whole window to a partial-data disposition rather than silently averaging over a gap. A declarative schema library rejects malformed batches at the boundary of the task so a bad payload fails fast instead of corrupting a reported number.

from datetime import datetime

from pydantic import BaseModel, Field, field_validator


class BatchReading(BaseModel):
    contaminant_id: str
    value: float
    sample_ts: datetime
    quality_flag: str
    source_hash: str = Field(min_length=64, max_length=64)

    @field_validator("sample_ts")
    @classmethod
    def _must_be_utc(cls, ts: datetime) -> datetime:
        if ts.tzinfo is None or ts.utcoffset() is None:
            raise ValueError("sample_ts must be timezone-aware UTC")
        return ts


def eligible_values(readings: list[BatchReading]) -> list[float]:
    """Only clean or interpolated readings may drive a compliance calculation."""
    return [r.value for r in readings if r.quality_flag in {"GOOD", "INTERPOLATED"}]

The compliance windows the batch computes are the same aggregations regulation defines. A rolling window mean over the eligible values in window qq of width nn is

Cˉq=1n∑i=q−n+1qCi\bar{C}_q = \frac{1}{n}\sum_{i=q-n+1}^{q} C_i

and a batch may only report Cˉq\bar{C}_q when the count of eligible readings meets the rule’s minimum data-availability threshold for that window. Several edge cases must be handled explicitly at this tier:

  • Partial windows. A batch cut before every reading arrives must stay open, or reopen on a late value, rather than committing an average over an incomplete set. The idempotency key changes when the reading set changes, so the reopened window is a distinct, traceable computation.
  • Timezone and DST drift. Windows are cut on a UTC axis; a batch keyed on local wall-clock time can appear to run backward across a fall-back transition and double-count an hour. All window bounds are normalized before hashing.
  • Duplicate delivery. At-least-once brokers redeliver. The idempotency check in Phase 3 makes a redelivered batch a no-op rather than a second submission.
  • Non-finite readings. NaN or Inf from a faulting analyzer is flagged BAD upstream and excluded here, never averaged into a phantom exceedance.
Batch lifecycle state machine A batch begins OPEN, accumulating readings as they arrive. When its reporting window is complete it becomes SEALED, then VALIDATING, then reaches the terminal COMMITTED state on a successful exactly-once commit. From VALIDATING a transient error transitions to RETRYING, which retries with backoff back into VALIDATING; when retries are exhausted the batch enters the terminal DEAD_LETTER state. If a late or corrected reading arrives while SEALED, the batch transitions to REOPENED and re-accumulates as OPEN, forming a distinct, traceable computation with a new idempotency key. OPEN accumulating SEALED window complete VALIDATING COMMITTED REOPENED distinct key RETRYING backoff DEAD_LETTER operator review seal dispatch commit once reading arrives late / corrected re-accumulate transient retry exhausted

Deployment & Integration Patterns

Workers deploy as small, single-purpose containers with a read-only root filesystem, so a compromised process cannot persist a foothold or rewrite its own code; the append-only ledger and dead-letter store are the only writable mounts. Scale is horizontal — add worker replicas against the priority queue to raise validation throughput — but concurrency per worker stays pinned to the database and external-API limits established in the architecture, because oversubscribing those dependencies converts throughput into timeouts and retries.

Backpressure is handled in the broker, never on the sensors. The ingestion edge is a strictly non-blocking producer: it enqueues a batch and returns immediately, so a burst of telemetry lengthens the queue rather than stalling acquisition, and workers drain at their own sustainable rate. This mirrors the passive-consumer discipline enforced at the Security Boundary Design perimeter, where nothing downstream may apply load back toward the control network. Threshold context for a committed batch is resolved against the SDWA MCL Reference Mapping in the Core Architecture & SDWA Compliance Taxonomy, the reporting window it belongs to is governed by Monitoring Frequency Scheduling, and once sealed the batch’s result feeds the Violation Detection Rule Engine, where exceedance and monitoring-gap logic turn attributable measurements into reportable compliance state. When integrating custom coroutine-based validation hooks inside a worker, the Python asyncio documentation covers event-loop tuning and coroutine lifecycle; keep those hooks inside the worker process and off the acquisition path.

Production Validation Checklist

Failure Modes & Gotchas

The single most consequential misconfiguration is treating the dead-letter queue as an error sink instead of a compliance backlog. A batch reaches the dead-letter store only after every retry has failed — which means, by definition, that a compliance window has not been evaluated. If the queue is unmonitored, those unevaluated windows accumulate silently, and each one may be an unreported SDWA event hiding behind a green ingestion dashboard. The failure is easy to miss precisely because live telemetry keeps flowing and the real-time pipeline looks healthy; only the validation result is missing. Catch it by making dead-letter depth a first-class signal: chart it beside sensor-health metrics, alert when it grows, and require an explicit operator disposition for every entry — corrected and re-queued, or escalated as a formal data gap — so a batch can never leave the backlog without a documented decision.

A close second is a broken idempotency guarantee. If the key is derived from anything non-deterministic — insertion order, a wall-clock timestamp, an unsorted reading list — a redelivered or reopened batch hashes differently, defeats the duplicate check, and submits the same window twice. Assert in tests that two envelopes built from the same readings in any order produce identical keys, and that the result backend commit is genuinely conditional on that key.

Related pages