Backfilling a Stream Pipeline from Archived Scenes

Replay archived scenes through a dedicated backfill queue and a separate consumer alias, cap that consumer at 200 of the 1,000 regional Lambda concurrency slots, and tag every replayed record mode=backfill so its historical acquisition time is written to a versioned history table instead of the live per-tile pointer. A pipeline built for live events has no natural defence against a producer that can emit two years of work in ninety seconds: the archive is not rate-limited by physics the way a satellite revisit is. The two things that break are throughput — the replay starves the live path of concurrency — and state, because a 2024 acquisition replayed today looks, to a last-writer-wins upsert, exactly like the newest observation of that tile.

Context

The batch vs stream decision is made on two numbers: how often data arrives and how stale a result users tolerate. A backfill breaks both assumptions at once. Arrival frequency is no longer a property of the sensor — it is a property of your dispatch loop — and the staleness of a replayed record is measured in months, not seconds. That is why a backfill run through an unmodified live pipeline is one of the most reliable ways to take that pipeline down.

Make it concrete. A pipeline ingests Sentinel-2 L2A scenes as they land, and each scene fans out into per-window tiling jobs the way chunked I/O for large satellite imagery describes: a 10 m band is 10,980 × 10,980 pixels, so a 512-pixel window grid is 22 × 22 = 484 jobs per band set. Live traffic is roughly 120 scenes a day — about 58,000 messages, under one per second averaged out. The archive you have been asked to reprocess holds 9,200 scenes, which is 4.45 million of the same messages. The live pipeline has never seen more than a few hundred messages in flight; the backfill hands it four and a half million.

Three failure modes follow, and they arrive in this order. First, concurrency exhaustion: the tiling consumer scales to the 1,000 regional concurrency default, the live consumer gets throttled behind it, and the live queue’s ApproximateAgeOfOldestMessage climbs from seconds to hours. Second, downstream saturation — the same catalogue table, the same output bucket prefix, the same PostGIS connection pool. Third, and worst because it is silent, state corruption: every backfilled scene carries a 2024 or 2025 acquisition timestamp, and any component that keeps a “current” value per MGRS tile now records whichever historical scene happened to finish last.

The mechanism you cannot borrow from the live path is the watermark. A live windowed aggregator such as the one in windowed aggregation of AIS positions in Kinesis advances a watermark from observed event times and drops anything that falls behind it. Feed it two-year-old records and one of two things happens: either every replayed record is discarded as impossibly late and the backfill silently produces nothing, or the watermark is derived from a maximum and the replay writes into windows that were closed and published months ago. Neither is a bug you want to find after the run completes.

Prerequisites

• Runtime: Python 3.11+ on AWS Lambda for both consumers, or the same handlers behind a Cloud Run pull subscription on GCP. The backfill consumer needs its own function or its own alias — not a shared one.
• Dependencies: boto3>=1.34.0 for queue and inventory access; rasterio>=1.3.9 in the tiling worker itself, unchanged by the backfill.
• A durable enumeration. An S3 Inventory report (daily, Parquet) over the archive prefix, or a STAC search result written to object storage. A live ListObjectsV2 walk is not acceptable: it cannot be resumed after a failure at object 3.1 million, and it costs a full re-listing every time you retry.
• Restored objects. Scenes on S3 Glacier Flexible Retrieval must be restored before the workers touch them — bulk retrieval takes 5–12 hours, standard 3–5 hours. Issue restore_object with Days=7 as a distinct phase and gate dispatch on the restore completing, or every worker fails with InvalidObjectState.
• Concurrency headroom you have actually measured. Take the live path’s observed peak ConcurrentExecutions, not its average. The remainder of the 1,000 regional quota, minus a margin, is your entire budget.
• IAM: sqs:SendMessage and sqs:GetQueueAttributes on the backfill queue for the dispatcher; s3:GetObject and s3:RestoreObject on the archive prefix; dynamodb:UpdateItem on the history table only — the backfill role must not hold write access to the live pointer table.

Choosing the Replay Lane

The first decision is whether the replay enters the same transport the live pipeline reads. It is worth stating the case for “yes”, because it is real: one lane means one consumer, one set of metrics, and no risk that the backfill path diverges from production behaviour and validates something you are not actually running. That argument holds when the replay is small enough to vanish into existing headroom — a few thousand messages injected over an hour, where the worst case is a modest latency bump.

It stops holding at any scale where the replay needs its own throttle. A Kinesis stream or an SQS queue has no notion of message priority; once four million historical records are in the log, the live records behind them wait. Retention makes the same point from the other side: a Kinesis stream retains 24 hours by default and can be extended to 365 days, so a stream can sometimes replay itself, but that is a recovery mechanism bounded by retention, not a reprocessing mechanism bounded by the archive.

The separate lane also buys the one operational property that matters most on a multi-day run: you can stop it. Deleting a subscription filter or setting reserved concurrency to zero pauses a backfill instantly and leaves the live path untouched. There is no equivalent move when both workloads share a queue.

Rate Control That Cannot Starve the Live Path

The instinct is to throttle the producer — send N messages per second and the pipeline will consume N per second. It does not work, because the queue absorbs any producer rate and the consumer scales to whatever the concurrency quota allows. The dispatcher’s rate only controls how fast the backlog forms; the consumer’s concurrency controls how much of the account’s capacity it takes.

So rate-limit at the consumer. Reserved concurrency on the backfill function is a hard account-level cap: set it to 200 and the backfill can never occupy more than 200 of the 1,000 regional slots, regardless of how deep the queue gets. Drain time then falls out of arithmetic rather than guesswork. At an average tiling job of 40 seconds, 200 slots is 5 jobs per second, and 4.45 million windows take about 10.3 days.

Pick the row that leaves the live path its measured peak plus margin, and accept the drain time that comes with it. A backfill that finishes in 2.6 days by taking 800 slots has not been tuned; it has been aimed at the same quota the live pipeline depends on. The dispatcher still needs a modest rate cap — enough to keep the backlog from growing past the queue’s retention window — but it is a safety valve, not the throttle.

Implementation

The dispatcher below walks the inventory cursor, restores anything still in Glacier, and sends into the backfill queue in batches of ten. Every message carries the flags the consumer needs to keep historical and live work apart.

import json
import os
import time
from datetime import datetime, timezone

import boto3

s3 = boto3.client("s3")
sqs = boto3.client("sqs")
ddb = boto3.client("dynamodb")

BACKFILL_QUEUE_URL = os.environ["BACKFILL_QUEUE_URL"]   # dedicated queue, not the live one
CURSOR_TABLE = os.environ["CURSOR_TABLE"]               # resumable position in the manifest
ARCHIVE_BUCKET = os.environ["ARCHIVE_BUCKET"]
DISPATCH_BUDGET = int(os.environ.get("DISPATCH_BUDGET", "2000"))  # messages per tick

# The dispatcher runs on a one-minute schedule. It is a safety valve on backlog
# depth, not the throughput control — that is the consumer's reserved concurrency.


def _needs_restore(key: str) -> bool:
    head = s3.head_object(Bucket=ARCHIVE_BUCKET, Key=key)
    if head.get("StorageClass") not in ("GLACIER", "DEEP_ARCHIVE"):
        return False
    # Restore already in progress or complete reports via the x-amz-restore header.
    return 'ongoing-request="false"' not in head.get("Restore", "")


def _envelope(scene: dict, window: dict) -> dict:
    return {
        "job_id": f"{scene['scene_id']}:{window['col']}:{window['row']}",
        "source_uri": f"s3://{ARCHIVE_BUCKET}/{scene['key']}",
        "window": window,
        # event_time is the acquisition time, never the replay time. Downstream
        # partitioning must land the output in the month the scene was captured.
        "event_time": scene["acquired_at"],
        "replay_time": datetime.now(timezone.utc).isoformat(),
        # The single flag every downstream branch keys off. Its absence must be
        # treated as live, so a forgotten tag fails closed rather than silently.
        "mode": "backfill",
        "watermark_advance": False,
    }


def handler(event, context):
    cursor = ddb.get_item(
        TableName=CURSOR_TABLE, Key={"run_id": {"S": event["run_id"]}}
    ).get("Item", {})
    offset = int(cursor.get("offset", {}).get("N", "0"))

    sent, batch = 0, []
    for scene in read_manifest(event["manifest_uri"], start=offset):
        if _needs_restore(scene["key"]):
            s3.restore_object(
                Bucket=ARCHIVE_BUCKET,
                Key=scene["key"],
                RestoreRequest={"Days": 7, "GlacierJobParameters": {"Tier": "Bulk"}},
            )
            continue  # revisit on a later tick; do not block the dispatch loop

        for window in tile_windows(scene):          # 22 x 22 = 484 per band set
            batch.append({
                "Id": str(len(batch)),
                "MessageBody": json.dumps(_envelope(scene, window)),
            })
            if len(batch) == 10:                    # SQS SendMessageBatch maximum
                sqs.send_message_batch(QueueUrl=BACKFILL_QUEUE_URL, Entries=batch)
                sent += len(batch)
                batch = []

        offset += 1
        if sent >= DISPATCH_BUDGET:
            break

    if batch:
        sqs.send_message_batch(QueueUrl=BACKFILL_QUEUE_URL, Entries=batch)

    ddb.update_item(
        TableName=CURSOR_TABLE,
        Key={"run_id": {"S": event["run_id"]}},
        UpdateExpression="SET #o = :o, updated_at = :t",
        ExpressionAttributeNames={"#o": "offset"},
        ExpressionAttributeValues={
            ":o": {"N": str(offset)},
            ":t": {"S": datetime.now(timezone.utc).isoformat()},
        },
    )
    return {"dispatched": sent, "offset": offset}

The consumer shares the tiling code with the live path and diverges only where it writes. The pointer update is the guard that makes a replay safe to run twice:

def commit(result: dict, envelope: dict) -> None:
    """Write tiling output, then advance the per-tile pointer only if newer."""
    ddb.put_item(                       # versioned history: always safe to write
        TableName=HISTORY_TABLE,
        Item={
            "tile_id": {"S": result["tile_id"]},
            "event_time": {"S": envelope["event_time"]},   # sort key
            "output_uri": {"S": result["output_uri"]},
            "mode": {"S": envelope["mode"]},
        },
    )
    if envelope["mode"] == "backfill":
        return                          # a replay never touches the live pointer

    ddb.update_item(
        TableName=POINTER_TABLE,
        Key={"tile_id": {"S": result["tile_id"]}},
        UpdateExpression="SET latest_uri = :u, event_time = :t",
        # Belt and braces: even on the live path, only a strictly newer
        # acquisition may advance the pointer.
        ConditionExpression="attribute_not_exists(tile_id) OR event_time < :t",
        ExpressionAttributeValues={
            ":u": {"S": result["output_uri"]},
            ":t": {"S": envelope["event_time"]},
        },
    )

Because both paths write history rows keyed by (tile_id, event_time), a replayed message that is delivered twice produces one row, not two — the same idempotency property that deduplicating S3 event notifications relies on, applied to a deterministic key instead of a lock table.

Verification

The measurement that matters is not the backfill’s own progress. It is the live queue’s oldest-message age while the backfill runs, sampled together:

# Live lane: this number is the pass/fail criterion for the whole run
aws sqs get-queue-attributes --queue-url "$LIVE_QUEUE_URL" \
  --attribute-names ApproximateAgeOfOldestMessage ApproximateNumberOfMessagesVisible

# Backfill lane: depth tells you the drain rate, not the health of the system
aws sqs get-queue-attributes --queue-url "$BACKFILL_QUEUE_URL" \
  --attribute-names ApproximateNumberOfMessagesVisible ApproximateNumberOfMessagesNotVisible

Expected output roughly an hour into a healthy run — the live lane unchanged from its idle baseline, the backfill lane holding 200 messages in flight, which is exactly its reserved concurrency:

{ "Attributes": { "ApproximateAgeOfOldestMessage": "3",
                  "ApproximateNumberOfMessagesVisible": "11" } }
{ "Attributes": { "ApproximateNumberOfMessagesVisible": "1874320",
                  "ApproximateNumberOfMessagesNotVisible": "200" } }

If ApproximateNumberOfMessagesNotVisible on the backfill lane sits below 200, the dispatcher is the bottleneck and you can raise DISPATCH_BUDGET. If the live lane’s oldest-message age has moved off its baseline at all, lower the backfill’s reserved concurrency — the cap is too generous, and no amount of dispatcher tuning will fix it.

Confirm the state guard separately by replaying one scene you know is older than the current pointer, then reading the pointer row back. It must be unchanged, and the history table must have gained exactly one row.

Gotchas and Edge Cases

• A shared alias defeats reserved concurrency. Reserved concurrency is set per function or per alias. If the backfill invokes the same $LATEST the live event source mapping uses, the reservation applies to both and the cap you thought protected the live path is now the live path’s ceiling too. Publish a version, point a backfill alias at it, and attach the backfill event source mapping to that alias.
• Out-of-order historical data is normal, not an error. An S3 Inventory report is sorted by key, which for most archive layouts means sorted by tile then by date — so scenes arrive grouped by geography, wildly out of chronological order. Anything downstream that assumes monotonic event time will misbehave. Partition output by the acquisition month carried in event_time, and let the history table’s sort key do the ordering at read time.
• Retention is a deadline on the whole run. SQS keeps a message for at most 14 days. A backfill sized to drain in 10.3 days has four days of slack; one that drains in 20 days will lose its oldest messages before the consumer reaches them. Either dispatch in waves sized to the drain rate, or keep the manifest cursor as the source of truth so a lost message is simply re-dispatched.
• The DLQ must be separate too. Historical scenes fail in ways live scenes do not — missing .jp2 assets, superseded processing baselines, a CRS the current code no longer handles. Routing them into the live dead-letter queue buries the one live failure you needed to see under ten thousand historical ones.
• Restore charges are per object and per retrieval. Issuing restore_object on a scene that is already restored is harmless, but issuing it on 9,200 scenes twice because the dispatcher restarted without a cursor is not. Check the x-amz-restore header before every request, as the dispatcher above does.

Frequently Asked Questions

Can I replay archived scenes through the same stream the live pipeline reads?

Only when the replay is small enough to disappear into existing headroom. A stream is a shared ordered log: historical records injected into it consume the same throughput and the same consumer concurrency as live traffic, and they arrive interleaved with it. A separate backfill queue keeps the two independently throttleable and, more importantly, independently pausable — which is what you want on day three of a ten-day run.

How do I stop backfilled scenes from corrupting live state?

Never let a replayed record advance a latest-value pointer. Write backfill output to a versioned history keyed by scene id plus acquisition time, and guard every pointer update with a conditional write that only accepts a strictly newer event time. A last-writer-wins upsert is the specific mistake: it lets a 2024 acquisition replayed today overwrite the current composite for that tile, and nothing in the pipeline reports an error when it does.

What rate should a backfill run at?

Set it from spare concurrency, not from a messages-per-second target. Subtract the live path’s measured peak concurrent executions, plus a margin, from the 1,000-slot regional quota and give the remainder to the backfill consumer as reserved concurrency. Drain time then follows from the average job duration — 200 slots at a 40-second job is 5 jobs per second — and the live path is protected by an account-level guarantee rather than a producer-side estimate.

Does the backfill need the same idempotency guarantees as the live path?

It needs stronger ones. A live pipeline sees each object once plus the occasional at-least-once duplicate; a backfill is routinely run more than once, because runs get paused, resumed and partially redone. Deriving the output key deterministically from scene id, window and processing version means a second full run overwrites byte-identical outputs instead of producing a second copy under a new name.

Related

• Batch vs Stream Geospatial Processing — the arrival-frequency and staleness criteria a backfill temporarily suspends
• When to Use Batch vs Streaming for Real-Time AIS Tracking — the nightly reconciliation job that handles routinely late data, as opposed to a one-off replay
• Chunked I/O for Large Satellite Imagery — where the 484 windows per band set that dominate the replay volume come from
• Implementing Dead Letter Queues for Failed Vector Jobs — why the backfill lane needs a dead-letter queue of its own
• Chunking Raster Jobs to Fit the 15-Minute Lambda Ceiling — sizing the per-job duration that sets the backfill’s drain rate

Back to Batch vs Stream Geospatial Processing