Backfilling Historian Gaps with Celery Task Chains

When a poller crashes, a plant network partitions, or a historian is taken down for maintenance, the readings for that window never arrive — and a silent hole in the time series will quietly distort every averaging calculation that touches it. This page sits inside the parent Async Batch Processing Setup section and solves one narrow, high-stakes job: replaying a bounded stretch of missing historian data through Celery so that the recovered readings land exactly once, never overwhelm the source, and automatically trigger a recomputation of any compliance window they touch. Where the sibling page on configuring Celery for async water-quality batches establishes the broker, worker, and serialization baseline, this one assumes that baseline exists and concentrates on the fan-out/fan-in orchestration that makes a gap backfill safe to run — and safe to re-run — against a live plant historian.

The mechanism is a Celery chord: split the gap into fixed-width intervals, dispatch each interval as an idempotent task in a group, and attach a callback that fires only after every interval succeeds. That callback is where you re-derive the affected running averages consumed by the MCL Exceedance Logic Implementation, so a backfill never leaves the rule engine reasoning over a stale window.

Prerequisites & Environment Setup

You need a working Celery application with a result backend, because a chord cannot exist without one — the callback has to know when the last member of the group has finished, and that bookkeeping lives in the backend. Redis is the pragmatic default for both the broker and the result store on a compliance pipeline of this scale. The gap detector that feeds this workflow is documented under Monitoring Gap Detection Algorithms; here we assume it has already handed you a start and end timestamp for the hole.

python3 -m venv .venv && source .venv/bin/activate
pip install "celery[redis]==5.4.*" "redis==5.0.*" "pydantic==2.7.*"
# The historian client and DB driver as fits your deployment:
pip install "influxdb-client==1.43.*" "psycopg[binary]==3.2.*"

Point the app at your broker and backend, and enable late acknowledgement so an interval task that dies mid-flight is redelivered rather than lost:

from celery import Celery

app = Celery("historian_backfill", broker="redis://localhost:6379/0",
             backend="redis://localhost:6379/1")
app.conf.update(
    task_acks_late=True,            # ack only after the task returns
    task_reject_on_worker_lost=True,
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    result_expires=86400,           # keep chord state a day for auditing
)

Step-by-Step Implementation

Step 1 — Chunk the gap window into fixed intervals

A backfill window is described by a start instant, an end instant, and a chunk width $\Delta$ . Splitting it into evenly sized intervals bounds the work per task, caps memory, and lets a single failed interval retry without replaying the whole window. The number of tasks the chord will fan out to is

N = \left\lceil \frac{t_\text{end} - t_\text{start}}{\Delta} \right\rceil

Choose $\Delta$ so that each interval fits comfortably inside one historian query and one write batch — one hour is a sensible default for second-resolution tags. The generator below yields half-open [start, end) intervals so adjacent chunks never overlap on a boundary sample.

from datetime import datetime, timedelta, timezone
from typing import Iterator, NamedTuple


class Interval(NamedTuple):
    start: datetime
    end: datetime


def chunk_window(gap_start: datetime, gap_end: datetime,
                 delta: timedelta = timedelta(hours=1)) -> Iterator[Interval]:
    """Yield half-open [start, end) intervals covering the gap exactly once."""
    if gap_start.tzinfo is None or gap_end.tzinfo is None:
        raise ValueError("Backfill boundaries must be timezone-aware (UTC).")
    if gap_end <= gap_start:
        raise ValueError("gap_end must be strictly after gap_start.")
    cursor = gap_start
    while cursor < gap_end:
        nxt = min(cursor + delta, gap_end)
        yield Interval(cursor, nxt)
        cursor = nxt

Step 2 — Derive a stable idempotency key per interval

Re-running a backfill must not double-write. The defence is a deterministic idempotency key derived only from the identity of the work — the tag, the interval bounds, and the chunk width — so the same interval always hashes to the same key regardless of when or how many times it is dispatched. The historian write is then made conditional on that key: if the key has already been recorded as committed, the task is a no-op.

import hashlib


def idempotency_key(tag_id: str, interval: Interval, delta: timedelta) -> str:
    """Deterministic key: identical inputs always produce the identical digest."""
    raw = "|".join([
        tag_id,
        interval.start.astimezone(timezone.utc).isoformat(),
        interval.end.astimezone(timezone.utc).isoformat(),
        str(int(delta.total_seconds())),
    ])
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

Because the key is content-addressed, two operators kicking off the same backfill, or a retry landing beside its own original, all converge on one committed row per interval. Store committed keys in a small backfill_ledger table (or a Redis set) and consult it before every write.

Step 3 — Write the idempotent per-interval task with backoff

Each interval is one Celery task. It reads the source historian for its slice, validates the readings, and writes them under the idempotency key. The task is registered with automatic retry and exponential backoff so a transient historian timeout does not fail the whole chord, and a per-task rate_limit throttles how fast workers hammer the source — essential when you are replaying weeks of data against a production historian that is still serving live queries. The related patterns for durable historian reads and writes live in the Historian Integration Patterns for Compliance Pipelines section.

from celery import chord, group, chain
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)


@app.task(
    bind=True,
    autoretry_for=(TimeoutError, ConnectionError),
    retry_backoff=2,          # 2s, 4s, 8s, ...
    retry_backoff_max=120,
    retry_jitter=True,
    max_retries=6,
    rate_limit="30/m",        # protect the source historian
    acks_late=True,
)
def backfill_interval(self, tag_id: str, start_iso: str, end_iso: str,
                      delta_seconds: int) -> dict:
    interval = Interval(datetime.fromisoformat(start_iso),
                        datetime.fromisoformat(end_iso))
    key = idempotency_key(tag_id, interval, timedelta(seconds=delta_seconds))

    if ledger_contains(key):                      # already committed
        logger.info("Skip committed interval %s", key[:12])
        return {"key": key, "written": 0, "skipped": True}

    readings = read_source_historian(tag_id, interval.start, interval.end)
    validated = [r for r in readings if r.quality == "GOOD"]

    # Conditional, all-or-nothing commit tied to the idempotency key.
    written = write_readings_once(tag_id, validated, key)
    logger.info("Backfilled %d readings for %s", written, key[:12])
    return {"key": key, "written": written, "skipped": False}

The write_readings_once helper is where “exactly once” is enforced: it opens a transaction, inserts the readings, and inserts the key into the ledger in the same commit, so a crash between the two can never leave a written slice without its key or a key without its slice.

def write_readings_once(tag_id: str, readings: list, key: str) -> int:
    with db_transaction() as tx:
        if tx.ledger_contains(key):     # re-check inside the transaction
            return 0
        tx.bulk_insert_readings(tag_id, readings)
        tx.insert_ledger_key(key)
        return len(readings)

Step 4 — Fan out with a chord and finalize in the callback

Now assemble the canvas. Every interval becomes a signature in a group; the group becomes the header of a chord; and the chord’s body is a single callback that runs only after all intervals report success. That callback recomputes the compliance windows the backfill disturbed — the running annual averages and rolling means that the rule engine reads — so recovered data is reflected in violation logic without a manual trigger.

def launch_backfill(tag_id: str, gap_start: datetime, gap_end: datetime,
                    delta: timedelta = timedelta(hours=1)):
    header = group(
        backfill_interval.s(tag_id, iv.start.isoformat(),
                            iv.end.isoformat(), int(delta.total_seconds()))
        for iv in chunk_window(gap_start, gap_end, delta)
    )
    callback = recompute_compliance_windows.s(
        tag_id=tag_id,
        window_start=gap_start.isoformat(),
        window_end=gap_end.isoformat(),
    )
    return chord(header)(callback)      # returns the callback's AsyncResult


@app.task(bind=True, max_retries=3, retry_backoff=True)
def recompute_compliance_windows(self, interval_results: list, *,
                                 tag_id: str, window_start: str,
                                 window_end: str) -> dict:
    total = sum(r["written"] for r in interval_results)
    logger.info("Backfill complete: %d readings across %d intervals",
                total, len(interval_results))
    # Re-derive every averaging window overlapping the recovered range.
    affected = recompute_running_averages(tag_id, window_start, window_end)
    return {"tag_id": tag_id, "written": total,
            "windows_recomputed": affected}

Celery passes the list of every header task’s return value as the first positional argument to the chord body, which is why recompute_compliance_windows receives interval_results — that is how the finalizer knows the totals and can confirm no interval was skipped unexpectedly. If any single interval exhausts its retries and fails, the chord body does not run at all, so a partial backfill can never silently mark itself finished.

Configuration Reference

Treat every value below as versioned deployment configuration, tuned to your broker capacity and the load ceiling of the source historian — never as inline literals scattered through the code.

Parameter	Where	Typical value	Effect
`delta`	`chunk_window`	`1h`	Interval width $\Delta$ ; smaller means more tasks, finer retry granularity
`rate_limit`	`backfill_interval`	`30/m`	Caps per-worker dispatch rate to protect the historian
`retry_backoff`	`backfill_interval`	`2`	Base seconds for exponential backoff between retries
`retry_backoff_max`	`backfill_interval`	`120`	Ceiling on any single backoff interval
`max_retries`	`backfill_interval`	`6`	Attempts before an interval fails and aborts the chord
`task_acks_late`	app config	`True`	Redelivers a task if the worker dies mid-execution
`result_expires`	app config	`86400`	How long chord/result state is retained for auditing
`backfill_ledger`	database	—	Committed idempotency keys; the exactly-once guarantee

Verification & Testing

The property that matters most is idempotency: running the same interval twice must write the readings once. The test below asserts that the key is deterministic and that a second commit under an already-recorded key is a no-op.

from datetime import datetime, timedelta, timezone


def test_idempotency_key_is_deterministic():
    iv = Interval(datetime(2026, 3, 1, tzinfo=timezone.utc),
                  datetime(2026, 3, 1, 1, tzinfo=timezone.utc))
    k1 = idempotency_key("plant-a/turbidity", iv, timedelta(hours=1))
    k2 = idempotency_key("plant-a/turbidity", iv, timedelta(hours=1))
    assert k1 == k2 and len(k1) == 64


def test_second_write_under_same_key_is_noop(fake_db):
    iv = Interval(datetime(2026, 3, 1, tzinfo=timezone.utc),
                  datetime(2026, 3, 1, 1, tzinfo=timezone.utc))
    key = idempotency_key("plant-a/ph", iv, timedelta(hours=1))
    readings = [Reading(quality="GOOD")] * 5
    assert write_readings_once("plant-a/ph", readings, key) == 5
    assert write_readings_once("plant-a/ph", readings, key) == 0  # no double-write


def test_chunk_window_covers_gap_exactly():
    start = datetime(2026, 3, 1, tzinfo=timezone.utc)
    end = datetime(2026, 3, 1, 3, 30, tzinfo=timezone.utc)
    chunks = list(chunk_window(start, end, timedelta(hours=1)))
    assert len(chunks) == 4                     # ceil(3.5) == 4
    assert chunks[0].start == start and chunks[-1].end == end
    for a, b in zip(chunks, chunks[1:]):
        assert a.end == b.start                 # no overlap, no gap

Acceptance criteria before running a backfill against production:

chunk_window covers the gap with no overlapping or missing intervals, and rejects naive (non-UTC) timestamps.
The idempotency key depends only on tag, interval bounds, and chunk width — nothing time-of-run.
A replayed interval writes zero additional rows; the ledger holds exactly one key per interval.
write_readings_once commits readings and the ledger key in a single transaction.
A forced interval failure aborts the chord so the finalizer never runs on a partial backfill.
rate_limit holds worker dispatch below the historian’s measured safe ceiling.
The finalize callback recomputes every averaging window overlapping the recovered range.

Troubleshooting & Gotchas

The chord callback never fires. A chord requires a result backend that supports it; a misconfigured or missing backend leaves the header tasks finishing while the body waits forever. Confirm app.conf.backend is set and reachable, and that result_expires is long enough that state is not evicted before the slowest interval completes.
Duplicate rows appear after a re-run. The idempotency key is leaking run-time state — a timestamp, a UUID, or datetime.now() — so each run produces a new key. Derive the key only from the work’s identity, and verify with test_idempotency_key_is_deterministic.
The historian buckles under the backfill. Fan-out is fast; the source may not be. Lower rate_limit, widen delta so there are fewer, larger reads, or route the group to a dedicated low-concurrency queue so backfill traffic never competes with live ingestion.
Compliance numbers look wrong right after a backfill. The readings landed but the finalizer did not recompute, or it recomputed a window narrower than the one the recovered data actually affects. A running annual average is influenced by samples far outside the gap, so recompute every window that overlaps the recovered range — coordinate the boundaries with the MCL Exceedance Logic Implementation.
A retried task is redelivered but its slice was already written. Expected and safe: late acknowledgement means a worker that died after committing can have its task redelivered, and the ledger check turns the redelivery into a no-op. This is exactly why the write and the key share one transaction.

Async Batch Processing Setup — parent section and the async orchestration baseline
Configuring Celery for Async Water-Quality Batches — the broker, worker, and serialization setup this workflow builds on
Historian Integration Patterns for Compliance Pipelines — durable historian read and write patterns for the interval tasks
Monitoring Gap Detection Algorithms — the detector that supplies the gap boundaries this backfill replays
MCL Exceedance Logic Implementation — the compliance windows the finalize callback recomputes

Backfilling Historian Gaps with Celery Task Chains

Related pages