Automating Backfill for IoT Gateway Reconnects

When an offline gateway reconnects and flushes hours of buffered telemetry in a single burst, you need to backfill it in bounded, time-ranged batches and re-materialize only the continuous aggregate buckets that burst invalidated — never in one unbounded transaction that locks chunks and stalls the scheduler. This page turns that recovery into a deterministic loop: detect the gap per device, size the batch window, ingest oldest-first, then call refresh_continuous_aggregate() over exactly the affected range. It builds on the ingestion mechanics from the parent guide on Python ETL orchestration.

# Input Profiling: What to Measure Before Backfilling

A reconnect burst is not a normal ingestion path, so treat it as an incident with measurable inputs rather than firing a blind bulk INSERT. Three numbers decide how the backfill runs.

• Gap boundaries per device — the last timestamp the hypertable already holds for the reconnecting gateway (MAX(time)) and the oldest timestamp in the buffer it is now flushing. The half-open interval between them is the affected window.
• Buffered volume — how many rows and roughly how many bytes the gateway accumulated while offline. This sets the batch window so no single transaction rewrites too many chunks at once.
• Aggregate start_offset — the start_offset of every continuous aggregate refresh policy that reads this hypertable. If the gap start falls before now() - start_offset, the buffered rows land behind the refresh watermark and the scheduled policy will never touch them on its own.

Gap detection is a single query. Run it per device so a fleet-wide flush does not collapse distinct gaps into one oversized range:

-- Last stored point per gateway vs. the earliest buffered row now arriving.
SELECT
    device_id,
    MAX(time)                       AS last_stored,
    NOW() - MAX(time)               AS gap_duration
FROM sensor_readings
WHERE device_id = ANY(%(reconnected)s)
GROUP BY device_id
ORDER BY gap_duration DESC;

Pair that with the aggregate’s own configuration so you know whether the scheduled policy can be trusted to catch up, or whether you must refresh the range explicitly:

-- start_offset / end_offset of every refresh policy on the hypertable's aggregates.
SELECT
    ca.view_name,
    (config->>'start_offset')   AS start_offset,
    (config->>'end_offset')     AS end_offset,
    schedule_interval
FROM timescaledb_information.jobs j
JOIN timescaledb_information.continuous_aggregates ca
     ON ca.view_name = j.hypertable_name
WHERE j.proc_name = 'policy_refresh_continuous_aggregate';

# The Bounded Backfill Algorithm

The core rule is that both work units — ingestion and materialization — must be bounded. Ingest the buffer in fixed time windows so each transaction rewrites a predictable number of chunks, then issue a single refresh_continuous_aggregate() scoped to the gap rather than letting the policy grind the whole hypertable.

The number of batches follows directly from the gap span and the chosen window:

$$N_{batches} = \left\lceil \frac{T_{gap}}{W_{batch}} \right\rceil = \left\lceil \frac{t_{end} - t_{start}}{W_{batch}} \right\rceil$$

where $T_{gap}$ is the affected span (gap end minus gap start), and $W_{batch}$ is the batch window. Pick $W_{batch}$ so that one window’s worth of rows keeps each transaction inside the same 100 MB–1 GB working set you would size a chunk against — small enough to commit quickly and release locks, large enough that you are not paying round-trip overhead thousands of times.

The loop below walks the gap oldest-first in psycopg v3. Each iteration copies exactly one window, commits, and advances the cursor; only after the whole buffer is drained does it refresh the aggregate once:

from datetime import datetime, timedelta

import psycopg
from psycopg.rows import dict_row


def backfill_gateway(
    conn_str: str,
    device_id: str,
    gap_start: datetime,
    gap_end: datetime,
    batch_window: timedelta = timedelta(minutes=30),
) -> int:
    """Ingest a reconnect buffer oldest-first in bounded windows, then
    re-materialize only the aggregate buckets the backfill invalidated."""
    rows_written = 0
    window_start = gap_start

    with psycopg.connect(conn_str, row_factory=dict_row) as conn:
        while window_start < gap_end:
            window_end = min(window_start + batch_window, gap_end)

            with conn.cursor() as cur:
                # Idempotent load: ON CONFLICT DO NOTHING tolerates a replayed
                # window if the loop is restarted after a mid-backfill crash.
                cur.execute(
                    """
                    INSERT INTO sensor_readings (time, device_id, metric_type, value)
                    SELECT time, device_id, metric_type, value
                    FROM gateway_buffer
                    WHERE device_id = %(dev)s
                      AND time >= %(w_start)s
                      AND time <  %(w_end)s
                    ON CONFLICT (device_id, time, metric_type) DO NOTHING;
                    """,
                    {"dev": device_id, "w_start": window_start, "w_end": window_end},
                )
                rows_written += cur.rowcount
            conn.commit()                    # release chunk locks each window
            window_start = window_end

        # One bounded refresh over the exact affected range. refresh_continuous_
        # aggregate cannot run inside a transaction block, so autocommit is on.
        conn.autocommit = True
        with conn.cursor() as cur:
            cur.execute(
                "CALL refresh_continuous_aggregate('sensor_readings_1h', %s, %s);",
                (gap_start, gap_end),
            )

    return rows_written

Two properties make this safe. First, committing every window keeps the lock footprint per transaction small, so background compression and other refresh jobs are not starved. Second, the refresh call names an explicit [gap_start, gap_end) window, so TimescaleDB re-materializes only the buckets the backfill actually changed instead of scanning from the aggregate’s high-water mark forward. If the ingest transaction is separate from the refresh — as here — the newly landed rows are already committed and visible when the refresh runs, which is exactly what makes the re-materialization correct. For the ingest side of the loop, the connection-pooling and copy patterns in async batch ingestion with psycopg3 connection pooling scale this to many gateways reconnecting at once.

# Worked Example: A Six-Hour Gap in Thirty-Minute Batches

A cellular gateway drops off the network at 02:00 and reconnects at 08:00, having buffered six hours of telemetry locally. Gap detection returns last_stored = 02:00:00 and the buffer’s newest row at 08:00:00, so the affected window is [02:00, 08:00) — a span of six hours. Choosing a 30-minute batch window:

$$N_{batches} = \left\lceil \frac{6\ \text{h}}{30\ \text{min}} \right\rceil = \left\lceil \frac{360}{30} \right\rceil = 12$$

The loop runs twelve iterations — [02:00, 02:30), [02:30, 03:00), … [07:30, 08:00) — each committing independently. At, say, 4 devices sharing this gateway emitting a reading every 10 seconds, one 30-minute window carries roughly $4 \times 180 \times 30 = 21{,}600$ rows, a comfortably small transaction that commits in milliseconds and releases its chunk locks before the next window starts.

Once the twelfth batch commits, a single refresh re-materializes the hourly aggregate over exactly the six affected hours:

-- Runs once, after all 12 windows are ingested and committed.
CALL refresh_continuous_aggregate('sensor_readings_1h', '2026-07-17 02:00+00', '2026-07-17 08:00+00');

That one call rebuilds the six hourly buckets spanning the gap. Contrast this with the failure mode of a naive recovery: a single INSERT of all 259,200 rows followed by an unbounded CALL refresh_continuous_aggregate('sensor_readings_1h', NULL, NULL), which locks the hypertable for the length of the bulk write and then forces the aggregate to re-scan its entire history. The bounded version does the same repair with a bounded, resumable footprint.

# Edge Cases & When to Deviate

• Data landing behind the watermark. If the gap start is older than now() - start_offset, the scheduled refresh policy will skip it forever, because the policy only advances forward from its watermark. The explicit refresh_continuous_aggregate() call over the gap range covers this — but if late reconnects are routine, widen the policy’s start_offset so the automatic refresh window reaches back far enough to catch them without manual intervention. The tradeoffs are covered in refresh policy design and scheduling.
• Never refresh unbounded. Calling refresh_continuous_aggregate(view, NULL, NULL) after a backfill re-materializes the whole aggregate and can run for minutes to hours on a large hypertable, blocking the very buckets your dashboards read. Always pass the concrete [gap_start, gap_end) you detected.
• Compressed target chunks. If the gap reaches into chunks the compression policy already converted, the backfill INSERT must decompress the affected chunks first (or rely on TimescaleDB’s insert-into-compressed-chunk support on 2.11+). Batch windows aligned to chunk boundaries limit how many compressed chunks a single window disturbs.
• Ordering with retention. If any part of the gap falls before the hypertable’s drop_after horizon, retention may already have removed the destination chunks — backfilling them just recreates chunks the next retention sweep will drop again. Check the retention window before ingesting, and clamp gap_start to now() - drop_after so you never write rows destined for immediate deletion.
• Per-device, not fleet-wide, gaps. A shared upstream outage produces many gateways with different last-stored timestamps. Detect and loop per device rather than taking one global MIN(gap_start), or you will re-ingest windows for gateways that were never actually offline.

# Verification

After the backfill, confirm both that the raw rows landed and that the aggregate’s freshness recovered across the repaired range. The watermark helper reports how far the aggregate has been materialized:

-- Aggregate freshness: the materialization high-water mark should now sit at
-- or past the gap end, and the repaired buckets should hold non-null values.
SELECT
    _timescaledb_internal.to_timestamp(
        _timescaledb_internal.cagg_watermark(mat_hypertable_id)
    )                                            AS materialized_through
FROM _timescaledb_catalog.continuous_agg
WHERE user_view_name = 'sensor_readings_1h';

-- Spot-check the previously-empty buckets across the gap are now populated.
SELECT bucket, device_id, count
FROM sensor_readings_1h
WHERE bucket >= '2026-07-17 02:00+00'
  AND bucket <  '2026-07-17 08:00+00'
  AND device_id = %(dev)s
ORDER BY bucket;

If materialized_through sits past the gap end and every hourly bucket in the range returns a row, the recovery is complete. If buckets in the range are still empty, the refresh window did not cover them — usually a sign the gap start slipped behind start_offset; re-run refresh_continuous_aggregate() with the corrected boundaries. Persistent staleness beyond that points at a policy-level problem, diagnosed in troubleshooting stale continuous aggregates in production.

# Related

• Async Batch Ingestion with psycopg3 Connection Pooling — the pooled ingest path that feeds each backfill window at fleet scale
• Troubleshooting Stale Continuous Aggregates in Production — diagnosing aggregates that stay stale after a refresh
• Refresh Policy Design & Scheduling — sizing start_offset so late reconnects fall inside the automatic refresh window
• Data Retention, Compression & Lifecycle Automation — how backfill interacts with compression and retention windows

← Back to Python ETL Orchestration