Notifications S3 : gérer les doublons et l’ordre

Les notifications d’événements S3 permettent à une application de réagir aux changements dans un compartiment Amazon S3. Un téléversement peut déclencher un message en file d’attente, un traitement d’image ou une mise à jour d’index. La livraison est conçue pour se produire au moins une fois : le destinataire doit donc tolérer les répétitions. AWS présente cette fonction dans sa documentation sur les notifications.

Ce guide teste la réception avec des événements synthétiques de création d’objet et une base SQLite locale. La règle choisie est précise : pour chaque couple compartiment et clé, accepter une création uniquement si son séquenceur dépasse la plus grande valeur déjà enregistrée. Elle convient lorsque les arrivées anciennes peuvent être ignorées. Elle ne convient pas à un journal d’audit ni à un traitement devant couvrir chaque version d’un objet.

Nous avons exécuté 40 contrôles avec Python 3.12.14 et SQLite 3.53.1, dont des connexions concurrentes et une annulation de transaction après une panne simulée. Aucune ressource AWS n’a été créée. La livraison réelle, IAM, les nouvelles tentatives en file d’attente et l’exécution d’actions externes n’ont pas été testés.

Choisir la destination et les événements

Amazon S3 prend en charge SNS, SQS, Lambda et EventBridge. Les destinations directes SNS, SQS et Lambda doivent se trouver dans la région du compartiment. La livraison directe vers SQS FIFO n’est pas prise en charge ; EventBridge permet d’acheminer ces événements vers une file FIFO. Consultez la référence des destinations.

Pour cet exemple, choisissez uniquement les créations d’objet. s3:ObjectCreated:* couvre PUT, POST, COPY et la fin d’un téléversement multipartie. Une règle limitée à PUT manquerait les autres modes de création. Le destinataire ci-dessous rejette volontairement les suppressions, restaurations et événements de cycle de vie.

Limitez le préfixe d’entrée, par exemple à incoming/, et écrivez les résultats ailleurs. Un résultat téléversé qui correspond au déclencheur peut provoquer une nouvelle exécution. Les filtres de préfixe et de suffixe n’acceptent pas les caractères génériques. AWS explique les restrictions sur les filtres de clé et les règles qui se chevauchent.

Distinguer l’enveloppe du message et son contenu

L’analyseur accepte une notification S3 native contenant un tableau Records. Pour une livraison S3 vers SQS, décodez le corps JSON du message SQS avant de l’appeler. Il ne retire pas les enveloppes SNS et ne constitue pas un gestionnaire de lots SQS pour Lambda. EventBridge utilise une autre enveloppe, notamment detail ; la lui transmettre directement échouera.

Dans la structure native S3, les clés utilisent l’encodage URL des formulaires et les noms d’événements omettent s3:. Décodez la clé une seule fois. Le séquenceur ordonne uniquement les événements d’une même clé ; comparez les valeurs hexadécimales, pas les chaînes de longueurs différentes. Acceptez les ajouts mineurs compatibles, tout en vérifiant la version majeure et les champs requis.

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

La clé de notre exemple est incoming%2FJune+report%2Bfinal.csv. L’analyseur renvoie incoming/June report+final.csv : l’espace et le signe plus restent distincts. Un autre contrôle conserve a%252Fb sous la forme a%2Fb, sans le décoder deux fois. Les tests couvrent aussi un nom UTF-8, des séquences d’échappement incorrectes et des octets UTF-8 invalides.

La fonction accepte les versions 2.1 et les versions 2.x ultérieures, ignore les champs inutilisés et rejette une version majeure inconnue. Ce minimum est un seuil de compatibilité, pas l’affirmation que 2.1 est la dernière version. Une notification de test ne crée aucun travail sur un objet. Ces choix d’analyse ont été vérifiés avec des données locales.

Il s’agit d’un petit analyseur applicatif, pas d’un validateur exhaustif pour des entrées arbitraires. Des champs imbriqués absents peuvent provoquer KeyError ; des types inattendus peuvent produire d’autres erreurs. L’adaptateur de réception doit signaler ces échecs et permettre une nouvelle tentative ou une investigation, sans les acquitter comme du travail terminé.

Enregistrer l’événement et le travail dans une transaction

Enregistrer séparément « déjà vu », puis créer une tâche, laisse une fenêtre de panne. Si le processus s’arrête entre les deux, la nouvelle tentative peut sembler être un doublon alors qu’aucune tâche n’existe. Notre exemple regroupe les deux écritures : une table conserve le plus grand séquenceur accepté, et une table d’envoi, ou outbox, conserve le travail à transmettre ensuite.

Les deux tables résident dans un même fichier SQLite. Leur identité associe le compartiment à la clé décodée : deux objets distincts ne partagent pas leur comparaison d’ordre. Les séquenceurs sont stockés comme texte et comparés comme entiers Python, sans limitation à un entier de taille fixe dans la base. Ces fonctions utilisent les événements renvoyés par l’analyseur :

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

La transaction prend un verrou d’écriture avant de lire le séquenceur enregistré. Un doublon ne crée aucune nouvelle ligne ; une arrivée plus ancienne est signalée comme périmée. Un événement plus récent met à jour le séquenceur et ajoute son contenu avant la validation. L’argument facultatif fail sert uniquement à tester l’annulation de la transaction.

Dans notre contrôle de panne, l’exception survient après la mise à jour du séquenceur, mais avant l’insertion du travail. Les deux changements sont annulés et une nouvelle tentative réussit. L’état validé persiste après réouverture de la base. Deux connexions concurrentes soumettant le même événement produisent un résultat accepté et un doublon.

Ce test démontre le comportement d’une transaction locale. Il ne transforme pas un fichier SQLite en stockage partagé entre instances Lambda indépendantes. Un service déployé doit utiliser un stockage durable commun aux travailleurs, des écritures atomiques équivalentes, une stratégie de sauvegarde et de reprise, et une durée de conservation de l’état de déduplication.

Exécuter l’exemple local

Enregistrez les deux blocs de code précédents ensemble dans event_handler.py, dans un dossier accessible en écriture, puis ajoutez les données et la démonstration ci-dessous. Lancez python event_handler.py avec Python 3.12. Seule la bibliothèque standard est nécessaire. Le script crée une base temporaire à côté de son fichier :

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()

La démonstration affiche :

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 et 000F représentent tous deux 15 : la deuxième arrivée est un doublon. e représente 14 et est périmé. 10 représente 16 et est accepté. Deux lignes restent dans l’outbox parce que deux événements ont été acceptés, et non une ligne par tentative de livraison.

Les contrôles portent aussi sur des compartiments et clés distincts, un séquenceur très long, l’absence d’identifiant de version, des types d’événements exclus, des enveloppes incorrectes et des arrivées concurrentes de valeurs différentes. Le séquenceur final reste le plus grand, quelle que soit la première connexion à valider. Le nombre d’événements intermédiaires acceptés peut varier selon leur ordre d’arrivée ; c’est une conséquence de la règle choisie.

L’outbox ne termine pas le travail

L’exemple s’arrête après l’enregistrement du travail en attente. Il n’envoie aucune requête, n’acquitte aucun message SQS, ne télécharge aucun objet et ne marque aucune ligne comme terminée. Le processus d’envoi doit encore gérer les nouvelles tentatives, la surveillance et la reprise après une panne.

Si une action externe réussit, puis que le processus s’arrête avant d’enregistrer sa réussite, elle peut se répéter. Utilisez une identité d’idempotence stable à destination, ou une mise à jour conditionnelle côté destinataire lorsqu’elle est disponible. Un indicateur local « envoyé » ne supprime pas cette fenêtre. AWS traite aussi de l’ordre et des doublons d’événements.

Des tâches déjà enregistrées peuvent se terminer dans le désordre. Un séquenceur plus récent n’annule pas une ancienne ligne de l’outbox ; le vérifier avant une écriture externe laisse encore une condition de concurrence. Si la destination doit conserver uniquement le dernier résultat, elle doit effectuer sa propre comparaison atomique par clé, ou empêcher autrement un ancien traitement d’écraser le résultat plus récent. Ce package n’implémente pas ce processus d’envoi.

Conservez l’identifiant exact de version lorsqu’il est présent. Lire seulement la clé plus tard peut renvoyer un contenu différent de celui qui a produit l’événement. Le contenu enregistré conserve cet identifiant, sans récupérer l’objet. Notre guide du versioning S3, en anglais explique la récupération et l’accès à une version précise.

Vérifier le parcours AWS avant le déploiement

Configurez les permissions de destination séparément de celles de lecture des objets. S3 doit pouvoir envoyer vers la file ou le sujet, ou appeler la fonction. Une destination chiffrée peut aussi nécessiter des permissions dans la politique de clé. Suivez la documentation AWS des permissions de destination : une politique de compartiment ne suffit pas à établir tout le parcours. Notre guide des politiques S3, en anglais couvre l’accès aux objets.

Lisez la configuration actuelle avant de la modifier. PutBucketNotificationConfiguration remplace la configuration existante : préservez les règles sans rapport avec votre changement. AWS valide les destinations pendant cette opération ; une destination incorrecte peut faire échouer toute la mise à jour. Nous n’avons appliqué aucune configuration dans ces tests.

Dans un environnement AWS jetable, vérifiez qu’un vrai téléversement correspondant à la règle atteint la destination, qu’une clé exclue ne correspond pas au filtre et que le consommateur peut lire la version voulue si nécessaire. Testez les nouvelles tentatives, les messages en échec et la reprise après interruption. Consignez le compte, la région, le rôle, le filtre et la destination testés.

Ce guide concerne les compartiments Amazon S3 à usage général. La compatibilité avec l’API de stockage ne prouve pas la prise en charge des notifications ni un comportement de livraison identique chez Hivenet ou un autre service compatible S3. Vérifiez la fonction d’événements et ses garanties avant d’adapter le destinataire.