S3 event notifications: handle duplicates and event order

S3 event notifications let an application react to changes in an Amazon S3 bucket. A file upload can trigger a queue message, an image-processing job, or an index update. Delivery is designed to be at least once, so a handler must tolerate repeats. AWS describes the feature in its event notifications overview.

This guide tests the receiving side with synthetic object-created notifications and a local SQLite database. Its policy is specific: for each bucket and key, accept a creation event only when its sequencer exceeds the highest one already recorded. That can suit work where older arrivals may be skipped. It is unsuitable for an audit trail or a workflow that must process every object version.

We ran 40 checks with Python 3.12.14 and SQLite 3.53.1, including concurrent connections and rollback after an injected failure. No AWS resources were created. Live delivery, IAM, queue retries, and external work execution were not tested.

Choose a destination and event scope

Amazon S3 supports SNS topics, SQS queues, Lambda, and EventBridge. Direct SNS, SQS, and Lambda destinations must be in the bucket’s Region. Direct SQS FIFO delivery is unsupported; EventBridge provides a route to FIFO queues. See the destination reference.

For this example, choose object-created events only. Selecting s3:ObjectCreated:* covers PUT, POST, COPY, and completed multipart uploads. A rule limited to PUT would miss the other creation paths. The handler below deliberately rejects deletion, restoration, and lifecycle notifications instead of pretending that they have the same meaning.

Limit the input prefix, for example to incoming/, and keep generated output elsewhere. An output upload that matches the trigger can start another run. Prefix and suffix filters are literal matching rules, not wildcard expressions. AWS documents the restrictions on key filters and overlapping rules.

Keep the message envelope separate from the record

The parser accepts a native S3 notification with a Records array. For S3-to-SQS delivery, JSON-decode the SQS message body before calling it. It does not unwrap SNS messages or implement a Lambda SQS batch handler. EventBridge uses a different envelope, including detail; passing that document directly will fail.

In the native S3 event structure, keys use form-style URL encoding, and event names omit the s3: prefix. Decode a key once. A sequencer orders events only for the same object key; compare hexadecimal values, not unpadded strings. Accept compatible minor schema additions while checking the major version and required fields.

import re
from urllib.parse import unquote_plus

def parse_created(message):
    if not isinstance(message, dict):
        raise ValueError('Expected a native S3 notification object')
    if message.get('Event') == 's3:TestEvent':
        return []
    records = message.get('Records')
    if not isinstance(records, list) or not records:
        raise ValueError('Expected nonempty native S3 Records')
    parsed = []
    allowed = {'ObjectCreated:Put', 'ObjectCreated:Post',
               'ObjectCreated:Copy', 'ObjectCreated:CompleteMultipartUpload'}
    for record in records:
        if record.get('eventSource') != 'aws:s3':
            raise ValueError('Wrong event source')
        version = record.get('eventVersion', '')
        match = re.fullmatch(r'2\.(\d+)', version)
        if not match or int(match[1]) < 1:
            raise ValueError('Unsupported schema version')
        if record.get('eventName') not in allowed:
            raise ValueError('This example handles object-created events only')
        bucket = record['s3']['bucket']['name']
        obj = record['s3']['object']
        key, seq = obj['key'], obj.get('sequencer', '')
        if not isinstance(bucket, str) or not bucket:
            raise ValueError('Missing bucket name')
        if not isinstance(key, str) or re.search(r'%(?![0-9A-Fa-f]{2})', key):
            raise ValueError('Invalid encoded key')
        if not isinstance(seq, str) or not re.fullmatch(r'[0-9A-Fa-f]+', seq):
            raise ValueError('Missing or invalid sequencer')
        parsed.append({'bucket': bucket,
                       'key': unquote_plus(key, encoding='utf-8', errors='strict'),
                       'seq': format(int(seq, 16), 'x'),
                       'version_id': obj.get('versionId'),
                       'event': record['eventName']})
    return parsed

Our fixture’s encoded key is incoming%2FJune+report%2Bfinal.csv. The parser returns incoming/June report+final.csv: the space and literal plus remain distinct. Another test keeps a%252Fb as a%2Fb, rather than decoding it twice into a different key. A UTF-8 filename, malformed escapes, and invalid UTF-8 bytes are also covered.

The helper accepts schema 2.1 and later 2.x versions, ignores fields it does not use, and rejects an unknown major version. Its minimum is a compatibility floor, not a claim that 2.1 is the latest schema. It treats a test notification as having no object work. These are deliberate parser choices, verified with local fixtures.

This is a small application parser, not a complete schema validator for arbitrary input. Missing nested fields can raise KeyError, and unexpected types can raise other parsing errors. The receiving adapter must report failures and keep them available for retry or investigation; it must not acknowledge them as completed work.

Commit the event and pending work together

A separate “seen” write followed by a job enqueue has a failure window. If the process stops between them, the retry can look like a duplicate even though no job exists. Our example puts both writes in one transaction: a watermark stores the greatest accepted sequencer, and an outbox stores the work to dispatch later.

Both tables live in one SQLite file. The key is the bucket plus the decoded object key, so an unrelated object does not share an ordering comparison. Sequencers are stored as text and compared as Python integers, avoiding a fixed-width database integer limit. The following functions use events returned by the parser:

import json
import sqlite3

def connect(path):
    db = sqlite3.connect(path, timeout=10, isolation_level=None)
    db.execute('CREATE TABLE IF NOT EXISTS watermark (bucket TEXT, key TEXT, seq TEXT, PRIMARY KEY(bucket,key))')
    db.execute('CREATE TABLE IF NOT EXISTS outbox (bucket TEXT, key TEXT, seq TEXT, payload TEXT, PRIMARY KEY(bucket,key,seq))')
    return db

def record_event(db, event, fail=False):
    # Use only normalized events returned by parse_created.
    bucket, key, seq = event['bucket'], event['key'], event['seq']
    db.execute('BEGIN IMMEDIATE')
    try:
        previous = db.execute(
            'SELECT seq FROM watermark WHERE bucket=? AND key=?',
            (bucket, key)).fetchone()
        if previous and int(seq, 16) <= int(previous[0], 16):
            db.execute('COMMIT')
            return 'duplicate' if int(seq, 16) == int(previous[0], 16) else 'stale'
        db.execute('INSERT INTO watermark VALUES (?,?,?) '
                   'ON CONFLICT(bucket,key) DO UPDATE SET seq=excluded.seq',
                   (bucket, key, seq))
        if fail:
            raise RuntimeError('Injected failure before outbox insert')
        db.execute('INSERT INTO outbox VALUES (?,?,?,?)',
                   (bucket, key, seq, json.dumps(event, sort_keys=True)))
        db.execute('COMMIT')
        return 'queued'
    except Exception:
        db.execute('ROLLBACK')
        raise

The transaction takes a write lock before reading the watermark. A duplicate returns without another outbox row; an older arrival returns as stale. A newer event updates the watermark and adds its payload before commit. The optional fail argument exists only to exercise rollback in the test suite.

In our failure check, the exception occurs after the watermark update but before the outbox insert. Both changes roll back, and retrying the same event succeeds. Reopening the database retains the committed state. Two concurrent connections submitting the same event produce one queued result and one duplicate result.

This demonstrates local transaction behavior. It does not make a SQLite file a shared store across independent Lambda instances. A deployed service needs durable storage shared by its workers, equivalent atomic writes, backup and recovery, and a retention policy for deduplication state.

Run the local example

Save the two code blocks above together as event_handler.py in a writable folder, then append the following fixture and demonstration. Run python event_handler.py with Python 3.12. It uses only the standard library and creates a disposable local database beside the script:

def fixture(seq='f', key='incoming%2FJune+report%2Bfinal.csv', bucket='example-input', version='2.1'):
    return {'Records':[{'eventVersion':version, 'eventSource':'aws:s3',
        'eventName':'ObjectCreated:Put', 's3':{'bucket':{'name':bucket},
        'object':{'key':key,'sequencer':seq,'versionId':'demo-version-1'}}}]}

def event(seq='f', **kwargs):
    return parse_created(fixture(seq, **kwargs))[0]

if __name__ == '__main__':
    from tempfile import TemporaryDirectory
    from pathlib import Path
    sample = event()
    print('Decoded key:', sample['key'])
    print('Hex order:', int('10', 16) > int('f', 16))
    with TemporaryDirectory(dir=Path(__file__).parent) as folder:
        path = Path(folder) / 'events.sqlite'
        db = connect(path)
        for seq in ['f', '000F', 'e', '10']:
            print(seq, record_event(db, event(seq)))
        db.close()
        db = connect(path)
        print('After reconnect:', record_event(db, event('10')))
        print('Outbox rows:', db.execute('SELECT count(*) FROM outbox').fetchone()[0])
        db.close()

The demonstration prints:

Decoded key: incoming/June report+final.csv
Hex order: True
f queued
000F duplicate
e stale
10 queued
After reconnect: duplicate
Outbox rows: 2

f and 000F both represent 15, so the second arrival is a duplicate. e represents 14 and is stale. 10 represents 16 and is accepted. There are two outbox rows because two events were accepted, not one row for every delivery attempt.

The tests also verify separate buckets and keys, an unusually long sequencer, an absent version ID, unsupported event types, malformed envelopes, and concurrent arrivals with different sequence values. The final watermark remains the greater value regardless of which connection commits first. The number of accepted intermediate events can differ with arrival order; that follows the chosen policy.

An outbox does not finish the job

The example stops after recording pending work. It does not send requests, acknowledge SQS messages, download objects, or mark outbox rows complete. A dispatcher still needs retries, monitoring, and recovery after a crash.

If an external action succeeds and the dispatcher crashes before recording completion, the action can repeat. Use a stable idempotency identity at the destination or a destination-side conditional update where supported. A local “sent” flag alone cannot close that gap. AWS also discusses ordering and duplicate-event handling.

Already queued work can finish out of order. Recording a newer watermark does not cancel an older outbox row, and checking the watermark before an external write leaves another race. If the destination must retain only the latest result, it needs its own atomic per-key version comparison or another design that prevents older work from overwriting newer output. This package does not implement that dispatcher.

Keep the exact object version ID when present. Reading only the key later can retrieve content different from the upload that produced the event. The payload retains the identifier but does not retrieve any object. Our S3 versioning guide explains version-specific recovery and access.

Verify the AWS path before deployment

Configure destination permissions separately from object-read permissions. S3 needs permission to send to the queue or topic, or invoke the function. An encrypted destination may also need key-policy permissions. Follow AWS’s destination permission guidance; a bucket policy alone does not establish the entire delivery path. Our bucket policy guide covers the object-access side.

Read the current notification configuration before changing it. PutBucketNotificationConfiguration replaces the existing configuration, so preserve unrelated rules. AWS validates destinations during this operation; an invalid destination can fail the whole update. We did not apply a configuration in these tests.

In a disposable AWS environment, verify a real matching upload reaches the intended destination, an excluded key does not match the rule, and the consumer can retrieve the intended version when required. Exercise retries, failed-message handling, and recovery after interrupted work. Record which account, Region, role, filter, and destination were tested.

This guide covers Amazon S3 general purpose buckets. Do not infer notification support or identical delivery behavior for Hivenet or another S3-compatible service from storage API compatibility. Confirm the provider’s event feature and contract before adapting the receiver.