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.

S3 event deduplication with a DynamoDB conditional-put lock Two identical S3 notifications produce the same deterministic job ID. The first conditional PutItem succeeds and processing runs; the second fails the condition and is skipped. A TTL attribute later expires the lock. S3 event #1 ObjectCreated:Put S3 event #2 (dup) same bucket/key/etag Derive job_id sha256(bucket + key + etag) deterministic DynamoDB PutItem attribute_not_exists (job_id) + ttl first writer wins Process run GDAL once Skip Conditional fail 200 OK 409 One deterministic job_id → one conditional claim → exactly-once processing

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.

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.

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.

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