Idempotency Patterns for S3 Event-Triggered Lease Ingestion

When a lease lands in object storage and an S3 event notification kicks off extraction — s3:ObjectCreated:* fanning through SNS or SQS into a worker — the trigger itself is the source of your duplication problem. S3 event notifications are delivered at least once, occasionally more than once, and with no ordering guarantee. This page resolves one narrow question inside Async Batch Processing: how to make event-driven lease ingestion produce an exactly-once effect — one abstraction committed per real document — even though the same event may arrive two, three, or ten times, and even though a later event may reach a worker before an earlier one.

The stakes are financial. A redelivered ObjectCreated event that re-runs a full extraction is wasteful; a redelivered event that commits a second rent row is a data-integrity incident. The fix is not “process faster” but a deliberate identity discipline: choose an idempotency key that actually identifies the content you are ingesting, claim it with a conditional write before doing any expensive work, and key the final commit on the content hash so a lost claim still cannot double-post. The broker and queue topology this rides on — Redis configuration, per-stage queues, acks_late — belongs to the sibling guides; here we stay strictly on the S3-to-worker identity boundary.

The problem: S3 events are at-least-once and unordered

Three properties of S3 notifications drive every design decision below. First, duplicate delivery: S3 may emit the same ObjectCreated event more than once, and if SNS or SQS sits in the path, each hop adds its own at-least-once semantics. Second, out-of-order arrival: two uploads to the same object key — a base lease at 09:00 and an amendment re-upload at 09:05 — can surface to workers in reverse order. Third, replay after visibility timeout: an SQS message whose extraction outruns the queue’s visibility window is re-delivered while the first attempt is still running, so the same document is processed twice concurrently.

None of this is a bug to be filed; it is the contract. A worker that assumes “one event equals one new document” will double-count rent the first time any of these fires. The defense is to derive a stable identity from the event payload and to serialize the effect through a deduplication store, not to wish the delivery guarantees were stronger.

Choosing an idempotency key

The event record gives you several candidate identifiers. They are not interchangeable — each guarantees something different and fails differently on commercial lease workloads, where the same filename is overwritten by amendments and large scanned leases trigger multipart uploads.

The practical answer is a composite claim key plus a content-hash commit key, used at two different moments. The claim key — bucket + key + versionId + etag — is cheap: every field is present in the event record, so you can deduplicate redelivered events before spending a byte of extraction. On a versioned bucket, versionId cleanly separates a new amendment (new version, new id) from a redelivery of the same version (identical id). The etag is a belt-and-suspenders content signal that also covers unversioned buckets where versionId is null. Then the commit is keyed on the SHA-256 content hash computed after fetch, so even if two different events somehow claim and both reach the database — a duplicate that slipped a null versionId, say — the downstream write still collapses to one row. Never key the claim on the S3 object key alone: it is a filename, and filenames are reused every time a landlord re-uploads a revised PDF.

Recommended implementation

The handler below runs per SQS record. It builds a claim key from the event, attempts a conditional put into a DynamoDB deduplication table (a Redis SET NX works identically), and only proceeds if it won the claim. The final commit is separately guarded by the content hash, so the two barriers are independent. This is the same typed-decimal, Decimal-for-money contract the metadata normalization standards require at the canonical boundary, enforced before anything reaches the books.

import hashlib
import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional

import boto3
from botocore.exceptions import ClientError
from pydantic import BaseModel, Field, ValidationError

logger = logging.getLogger(__name__)
s3 = boto3.client("s3")
dedup_table = boto3.resource("dynamodb").Table("lease_ingest_claims")


class LeaseAbstraction(BaseModel):
    property_id: str
    tenant_name: str
    base_rent: Decimal = Field(gt=0)
    lease_start: datetime
    lease_end: datetime
    extraction_confidence: float = Field(ge=0.0, le=1.0)


def build_claim_key(record: dict) -> str:
    """Identity from the event payload — never the object key alone."""
    s3info = record["s3"]
    bucket = s3info["bucket"]["name"]
    obj = s3info["object"]
    key = obj["key"]
    version_id = obj.get("versionId", "null")   # 'null' on unversioned buckets
    etag = obj.get("eTag", "")                    # opaque for multipart uploads
    return f"{bucket}/{key}@{version_id}#{etag}"


def claim(claim_key: str) -> bool:
    """Conditional put: True only for the first caller to win this claim."""
    try:
        dedup_table.put_item(
            Item={
                "claim_key": claim_key,
                "status": "in_progress",
                "claimed_at": datetime.now(timezone.utc).isoformat(),
            },
            ConditionExpression="attribute_not_exists(claim_key)",
        )
        return True
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
            return False        # a prior delivery already claimed it
        raise                   # store unavailable — do NOT swallow


def handle_record(record: dict) -> dict:
    claim_key = build_claim_key(record)
    if not claim(claim_key):
        logger.info("Duplicate S3 event, skipping claim %s", claim_key)
        return {"status": "skipped_duplicate", "claim_key": claim_key}

    s3info = record["s3"]
    bucket, key = s3info["bucket"]["name"], s3info["object"]["key"]
    body = s3.get_object(Bucket=bucket, Key=key)["Body"].read()
    content_hash = hashlib.sha256(body).hexdigest()

    try:
        fields = _run_extraction(body)               # OCR / clause parsing
        abstraction = LeaseAbstraction(**fields)
    except ValidationError as exc:
        logger.warning("Schema-invalid lease %s: %s", key, exc)
        return {"status": "validation_error", "content_hash": content_hash}

    return _commit_idempotent(content_hash, abstraction)


def _commit_idempotent(content_hash: str, abstraction: LeaseAbstraction) -> dict:
    """Upsert keyed on content hash — a residual duplicate is a no-op."""
    # INSERT ... ON CONFLICT (content_hash) DO NOTHING, or an upsert with a
    # unique constraint on content_hash. Two events, same bytes -> one row.
    logger.info("Committed lease on hash %s", content_hash)
    return {"status": "committed", "content_hash": content_hash}

The division of labour is deliberate: claim() stops redelivered events cheaply, and _commit_idempotent() stops duplicate content durably. Neither alone is sufficient — the claim can be lost if the dedup store is flushed, and the content-hash commit alone would still pay for a full extraction on every redelivery. Transient-versus-permanent classification of the get_object and extraction failures — what retries, what dead-letters — is out of scope here and covered by error handling & retry logic.

Edge cases specific to commercial leases

• Overwrite with identical content. A tenant re-sends the exact same PDF and it is PUT again. On a versioned bucket this is a new versionId, so the claim key differs and the claim succeeds — but the SHA-256 is identical, so the content-hash commit collapses it to a no-op. The content hash is what saves you when the version metadata says “new.”
• Genuine amendment re-upload. A landlord uploads a revised lease under the same object key. New versionId, new bytes, new hash — correctly treated as a distinct document. Do not resolve precedence here; hand the new version to amendment versioning & ledgers so latest-effective-wins is decided downstream against effective dates, never by overwriting the base lease in place.
• Out-of-order event delivery. The amendment’s event arrives before the base lease’s. Because each is claimed and committed on its own content hash independently, ingestion order is irrelevant — ordering is re-established downstream by effective date, not by arrival time.
• Missing versionId on an unversioned bucket. The field is absent or the literal string null, so every overwrite of a key would share a claim key. Fall back on the etag, which changes with content on single-part PUTs, and lean on the content-hash commit as the real guarantee. Better still, enable bucket versioning for lease buckets.
• ETag is not the MD5 for multipart uploads. Large scanned leases uploaded in parts carry an ETag like "<md5>-<partcount>", which is not the object’s MD5. Treat the ETag as opaque — fine as a claim-key component, useless as a content checksum. Compute your own SHA-256 after fetch; never assume ETag equals the content digest.
• Replayed SQS message after visibility timeout. A 90-page scanned lease outruns the queue’s visibility window and the message is re-delivered while the first attempt still runs. Both attempts build the same claim key; the conditional put lets exactly one proceed. Set the visibility timeout above worst-case extraction latency so this is the exception, and keep idempotency as the safety net rather than the primary control.

When to escalate

Idempotency has failure modes of its own that must degrade safely rather than silently:

• Dedup store unavailable. If the conditional put raises (not a condition failure — an actual outage), do not swallow it and proceed, and do not treat it as “already claimed.” Let the message return to the queue so the delivery is retried under backpressure. Proceeding blind here is exactly how a duplicate slips through; failing closed preserves exactly-once.
• Ambiguous version identity. When versionId is missing and the ETag is a multipart placeholder and two events collide on the same key within one window, route the object to review through fallback routing logic rather than guessing which upload is authoritative.
• Repeated claim-without-commit. A claim marked in_progress that never reaches committed after its lease window signals a worker dying mid-parse; reclaim it on a TTL and, after repeated failures, dead-letter it with full context for ops triage.

Frequently asked questions

Should I key idempotency on the S3 object key? No. The object key is a filename, and lease workflows reuse filenames constantly — an amended lease is re-uploaded under the same key. Keying on the key alone would either treat a real amendment as a duplicate or, worse, let a redelivery overwrite a distinct version. Use bucket + key + versionId + etag for the claim and the content hash for the commit.

How do I tell a redelivered event apart from a new lease version? On a versioned bucket the versionId distinguishes them precisely: a redelivery carries the same versionId, a new upload carries a new one. The content hash confirms it — identical bytes collapse to a no-op at commit even when the version metadata differs, and different bytes are correctly ingested as a new document.

Why compute a content hash if the claim already deduplicates events? Because the two barriers guard different failures. The claim stops redelivered events cheaply but can be lost if the dedup store is flushed or a null versionId collides. The content-hash commit is the durable backstop that makes a double-post impossible even when the claim fails. Belt and suspenders is the point.

What should happen if the deduplication store is down? Fail closed. Let the SQS message return to the queue and retry under backpressure rather than proceeding without a claim. Assuming “not claimed, so process it” would double-count; assuming “already claimed, so skip” would drop a lease. Neither is safe, so do not commit until the store answers.

Related

• Scaling Async Lease Parsing with Celery + Redis — the broker, acks_late, and Redis dedup-set configuration this handler rides on once work is distributed.
• Designing Celery Task Queues for Lease Portfolio Batches — per-stage queue topology and routing for the tasks these claimed events enqueue.
• Dead-Letter Queue Handling for Failed Lease Extractions — where claim-without-commit orphans and repeated failures land for ops triage.
• Async Batch Processing — the parent stage: bounded concurrency, the pydantic gate, and idempotent commits this page specializes for S3 event triggers.

← Back to Async Batch Processing

← Back to Async Batch Processing