Deduplicating S3 Event Notifications for Idempotent Shapefile Ingestion

Amazon S3 event notifications are delivered at-least-once, so the same ObjectCreated:Put can invoke your handler two or more times. Make ingestion idempotent by deriving a deterministic job ID as sha256(bucket + "\n" + urldecode(key) + "\n" + etag), then claiming it with a DynamoDB PutItem guarded by ConditionExpression="attribute_not_exists(job_id)". Only the first delivery wins the conditional write and runs GDAL; every duplicate raises ConditionalCheckFailedException and returns early. A TTL attribute on the row reclaims stale locks automatically, so the table never grows unbounded.

Context

Amazon S3 promises at-least-once delivery for event notifications, not exactly-once. When S3 retries a notification during a partition failover, or when an SQS redrive replays a message, or when a Lambda invocation fails partway and is retried, your ingestion handler sees the same object twice. For a stateless thumbnail generator that is harmless. For shapefile ingestion it is not: reprocessing means duplicate rows in PostGIS, double-counted features in an aggregation, or two competing GDAL writes to the same Cloud Optimized GeoTIFF output.

Two at-least-once deliveries of one S3 event racing for the same dedup lockS3 delivers the same ObjectCreated:Put notification for county.shp twice. The first delivery derives a job ID from bucket, key and ETag, wins the conditional PutItem guarded by attribute_not_exists(job_id), and runs the GDAL ingestion. The redelivery derives the identical job ID, fails the same conditional put with ConditionalCheckFailedException, and returns without touching the object.The same notification, delivered twiceS3 notificationsIngest LambdaDynamoDB dedupGDAL / FionaObjectCreated:Putcounty.shp, eTag 9b2cf5PutItem job_idattribute_not_exists(job_id)claim wonstatus = processingrun ingestion oncedownload, fiona.open, loadthe same event againidentical bucket, key, eTagPutItem same job_idthe row already existsConditionalCheckFailedskipped, GDAL never runs
Both deliveries are legitimate — S3 promises at-least-once, not exactly-once. The conditional put is what turns two invocations into one ingestion.

The S3 and GCS Event Triggers for Shapefiles overview covers the multi-file atomicity problem — waiting for .shp, .shx, and .dbf to all arrive before dispatch, which the sibling page on aggregating multipart shapefile uploads treats in depth. This page solves the orthogonal problem: even once the bundle is complete, the trigger for that bundle can fire more than once. The fix is a dedup lock keyed by content identity, not by wall-clock time. It composes cleanly with any queue in front of the compute layer, including the SQS and Pub/Sub routing strategies that themselves add their own at-least-once semantics.

Prerequisites

Before wiring the dedup lock, confirm:

  • Runtime: Python 3.11 on AWS Lambda (Amazon Linux 2023). AWS Lambda gives you 15 minutes of timeout, up to 10,240 MB of memory, and 10 GB of /tmp — ample headroom for a single shapefile bundle.
  • DynamoDB table: on-demand billing, partition key job_id (String), a Number attribute expires_at registered as the table’s TTL attribute. No sort key is needed.
  • IAM — the Lambda execution role needs:
    • dynamodb:PutItem and dynamodb:UpdateItem on the dedup table ARN (scope to the table, never dynamodb:* on *)
    • s3:GetObject on the source bucket prefix
    • the managed AWSLambdaBasicExecutionRole for CloudWatch Logs
  • Environment variables (set in the function config; GDAL data paths point at the bundled layer):
    code 
    DEDUP_TABLE=shapefile-ingest-dedup
DEDUP_TTL_SECONDS=604800
GDAL_DATA=/opt/python/lib/python3.11/site-packages/rasterio/gdal_data
PROJ_LIB=/opt/python/lib/python3.11/site-packages/pyproj/proj_dir/share/proj
LD_LIBRARY_PATH=/opt/python/lib/python3.11/site-packages/rasterio.libs
  • Python dependencies: boto3 (bundled in the Lambda runtime), plus fiona==1.10.* for geometry validation in your ingestion step.

Implementation

The handler below derives the job ID, claims it with a conditional put, and only then runs ingestion. The conditional write is the entire concurrency-control mechanism: DynamoDB serializes competing puts on the same partition key, so exactly one duplicate wins the race even if two deliveries land in the same millisecond on two warm Lambda containers.

The five steps of a content-addressed dedup claimFive handler steps in order: read bucket, URL-decoded key and unquoted ETag from the notification record; derive a deterministic SHA-256 job ID over the three newline-joined fields; claim the job with a conditional PutItem guarded by attribute_not_exists on job_id; run the ingestion and mark the row done only inside a won claim; and let the TTL attribute reclaim the row seven days later.Five steps, and only step four decides anything1Read the event recordS3 URL-encodes keys and quotes the ETag; decode and strip both firstunquote_plus(key)2Derive the job IDNewline-joined so one field cannot blur into the nextsha256 hexdigest3Claim it with a conditional putDynamoDB serializes competing puts on the partition keyattribute_not_exists4Process, then mark doneIngestion runs only inside a claim this invocation wonstatus = done5Let the TTL reclaim the rowGarbage collection only; correctness never depends on it604800 s
Every step before the conditional put is pure derivation — the same event always produces the same job_id. The put is the only place where two deliveries can diverge.
python 
"""dedup_ingest.py — idempotent S3-triggered shapefile ingestion.

The DynamoDB conditional PutItem is the exactly-once gate. Everything
downstream of a successful claim runs at most once per (bucket, key, etag).
"""
import os
import time
import json
import hashlib
import logging
import urllib.parse

import boto3
from botocore.exceptions import ClientError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

TABLE_NAME = os.environ["DEDUP_TABLE"]
TTL_SECONDS = int(os.environ.get("DEDUP_TTL_SECONDS", "604800"))  # 7 days

# Reuse clients across warm invocations to avoid re-handshaking TLS.
_ddb = boto3.client("dynamodb")
_s3 = boto3.client("s3")


def _job_id(bucket: str, key: str, etag: str) -> str:
    """Deterministic content-identity hash.

    Including the ETag means a genuine re-upload of the same key (new
    content, new ETag) produces a NEW job_id and is reprocessed, while a
    duplicate notification for one upload (identical ETag) collapses onto
    the same job_id and is skipped. Newlines prevent boundary ambiguity
    between fields (e.g. a key ending in the bucket name of the next).
    """
    material = "\n".join((bucket, key, etag)).encode("utf-8")
    return hashlib.sha256(material).hexdigest()


def _claim(job_id: str, bucket: str, key: str) -> bool:
    """Attempt to claim the job. Returns True if this invocation won."""
    now = int(time.time())
    try:
        _ddb.put_item(
            TableName=TABLE_NAME,
            Item={
                "job_id": {"S": job_id},
                "bucket": {"S": bucket},
                "key": {"S": key},
                "status": {"S": "processing"},
                "claimed_at": {"N": str(now)},
                # TTL attribute: DynamoDB deletes the row after this epoch.
                "expires_at": {"N": str(now + TTL_SECONDS)},
            },
            # The gate: succeeds only if no row for this job_id exists yet.
            ConditionExpression="attribute_not_exists(job_id)",
        )
        return True
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
            # A prior (or concurrent) delivery already claimed this job.
            return False
        raise  # Throttling or a real fault — let Lambda retry.


def _mark_done(job_id: str) -> None:
    """Flip status to done. The row still expires via TTL later."""
    _ddb.update_item(
        TableName=TABLE_NAME,
        Key={"job_id": {"S": job_id}},
        UpdateExpression="SET #s = :d, finished_at = :t",
        ExpressionAttributeNames={"#s": "status"},
        ExpressionAttributeValues={
            ":d": {"S": "done"},
            ":t": {"N": str(int(time.time()))},
        },
    )


def _ingest(bucket: str, key: str) -> int:
    """Placeholder for the real GDAL/Fiona ingestion work.

    Download the bundle to /tmp, validate with Fiona, load to PostGIS or
    write a COG. Returns a feature count for logging. Kept small here so
    the dedup mechanics stay in focus.
    """
    head = _s3.head_object(Bucket=bucket, Key=key)
    logger.info("Ingesting s3://%s/%s (%d bytes)", bucket, key, head["ContentLength"])
    # ... download, fiona.open(...), load features ...
    return head["ContentLength"]


def handler(event, _context):
    processed, skipped = 0, 0
    for record in event.get("Records", []):
        s3 = record["s3"]
        bucket = s3["bucket"]["name"]
        # S3 URL-encodes keys in notifications; spaces become '+'.
        key = urllib.parse.unquote_plus(s3["object"]["key"])
        # ETag is quoted in the notification payload; strip the quotes.
        etag = s3["object"].get("eTag", "").strip('"')

        job_id = _job_id(bucket, key, etag)

        if not _claim(job_id, bucket, key):
            logger.info("Duplicate — skipping job_id=%s key=%s", job_id[:12], key)
            skipped += 1
            continue

        try:
            _ingest(bucket, key)
            _mark_done(job_id)
            processed += 1
        except Exception:
            # On failure, leave the row as 'processing'. It will expire via
            # TTL so a later redrive can re-claim and retry the object.
            logger.exception("Ingestion failed for key=%s — leaving lock to expire", key)
            raise

    return {"processed": processed, "skipped": skipped}

Verification

Invoke the handler twice with an identical record to prove the second run is skipped without touching S3:

python 
# test_dedup.py — run locally with moto or against a scratch table.
import json
from dedup_ingest import handler

record = {"Records": [{
    "s3": {
        "bucket": {"name": "geo-uploads"},
        "object": {"key": "parcels%2Fcounty.shp", "eTag": '"9b2cf5..."'},
    }
}]}

print(handler(record, None))   # first delivery claims and processes
print(handler(record, None))   # duplicate delivery is skipped

Expected output — the first call processes, the second is a no-op:

code 
{'processed': 1, 'skipped': 0}
{'processed': 0, 'skipped': 1}

Confirm the lock row and its TTL landed in DynamoDB:

bash 
aws dynamodb get-item \
  --table-name shapefile-ingest-dedup \
  --key '{"job_id": {"S": "<hash-from-logs>"}}' \
  --query 'Item.{status: status.S, expires_at: expires_at.N}'
# Expected: {"status": "done", "expires_at": "<epoch ~7 days out>"}

Gotchas and Edge Cases

  • URL-decode the key before hashing. S3 notification payloads URL-encode object keys and use + for spaces. If you hash the raw object.key, a key with a space produces a different job_id than the s3:GetObject call resolves, and dedup silently fails for those objects. Always run urllib.parse.unquote_plus first — and hash the decoded form.

  • Multipart uploads change ETag semantics. For a single-part PutObject, the ETag is the MD5 of the content. For a multipart upload it is a hash-of-hashes with a -N part suffix. That is fine for dedup — it is still stable per upload — but do not assume the ETag equals the object’s MD5 when validating content integrity elsewhere.

  • A crashed handler leaves a processing row. If ingestion throws after the claim, the row stays as processing and blocks retries until its TTL expires. For fast recovery, either delete the row in the exception path (accepting a tiny duplicate-processing window on a concurrent retry) or use a shorter claimed_at-based lease that a redrive can steal. Pair this with a dead-letter queue for failed vector jobs so poison objects do not loop forever.

  • DynamoDB TTL deletion is not instant. DynamoDB expires items within roughly 48 hours of the TTL timestamp, not exactly at it. That lag is harmless here because a live lock still exists during the whole retry window; TTL is only a garbage-collection convenience, never a correctness dependency. Never set a TTL short enough that a lock could vanish while duplicates might still arrive.

State machine of a DynamoDB dedup lock rowA dedup lock row has three states. No lock exists until a first delivery claims the job ID. In the processing state every duplicate delivery fails the conditional put and is skipped. A successful ingestion moves the row to done, and the TTL attribute eventually deletes it, returning the job ID to the no-lock state.What a dedup lock row can be, and how it leavesNo lockjob_id unseenProcessingclaimed by one deliveryDoneingestion finishedfirst delivery claimsduplicate fails the conditioningestion succeedsTTL deletes the rowA handler that throws after the claim leaves the row stuck in processing, and DynamoDB applies the TTL within roughly 48 hours of expires_atrather than exactly at it — which is why the lock is sized for the whole retry window, not for the delete.
The self-loop is the whole point of the pattern: a duplicate is a normal, expected event with a normal, expected outcome.

Frequently Asked Questions

Why does S3 deliver the same event notification more than once?

Amazon S3 event notifications are designed for at-least-once delivery. During internal retries, partition failovers, or when a destination (SQS, SNS, Lambda, or EventBridge) briefly rejects a message, S3 re-sends the notification. The documentation explicitly states notifications are typically delivered in seconds but can occasionally arrive more than once, so every consumer must be built to tolerate duplicates rather than assume exactly-once.

Should the ETag be part of the dedup key?

Yes. The ETag distinguishes a genuine re-upload of the same key (new content, new ETag) from a duplicate notification for the same upload (identical ETag). Hashing bucket, key, and ETag together means a real overwrite earns a fresh job_id and is reprocessed, while a duplicate delivery collapses onto the existing lock and is skipped. Keying on bucket and key alone would wrongly suppress reprocessing of a legitimately updated file.

What TTL should I set on the DynamoDB dedup lock?

Set the TTL comfortably longer than the maximum window in which duplicates can arrive, plus your processing time. S3 duplicates almost always land within minutes, but SQS redrives and Lambda retries can stretch that. A 24-hour to 7-day TTL is common: long enough to absorb every retry window, short enough to keep the table small and cheap. The lock’s correctness never depends on the TTL — it only prevents the table from growing without bound.

Can I use SQS FIFO deduplication instead of DynamoDB?

SQS FIFO offers a 5-minute deduplication window, which is far too short for shapefile pipelines where a redrive or a delayed retry can exceed it. FIFO is the right tool for guaranteeing order on sequential tile jobs, but for content-addressed idempotency over hours or days, the DynamoDB conditional-put lock is the durable choice.

Back to S3 and GCS Event Triggers for Shapefiles