Idempotence : un document, une ligne

Le contrat de la leçon précédente garantit qu'aucun document n'est perdu : tout ce qui échoue revient. Il garantit aussi l'inverse de ce que veut la comptable : un message peut revenir alors qu'il a déjà été traité. Un délai d'acquittement dépassé, une instance arrêtée entre l'écriture BigQuery et la réponse HTTP, et la facture apparaît deux fois, avec deux appels Gemini payés. Le worker doit donc être idempotent : traiter deux fois le même message produit le même état qu'une seule fois.

Le mécanisme

Pub/Sub attribue à chaque message publié un messageId unique sur le sujet, et le conserve à chaque relivraison. C'est la clé de déduplication de ce projet. Le worker fait deux vérifications, qui n'ont pas le même rôle :

1. seen(message_id) : une requête SELECT 1 … LIMIT 1 avant l'appel Gemini. Elle économise l'appel dans le cas courant, une relivraison quelques secondes après le succès.
2. record() : un MERGE … WHEN NOT MATCHED THEN INSERT. Il garantit l'unicité, y compris quand deux livraisons du même message arrivent en même temps sur deux instances et passent toutes les deux la première vérification.

flowchart TD
  M["Message reçu<br/>messageId = 2220…603"] --> S{"seen() ?"}
  S -- "oui" --> D["200 duplicate<br/>aucun appel Gemini"]
  S -- "non" --> C{"document valide ?"}
  C -- "non" --> R["MERGE status=rejected"]
  C -- "oui" --> G["Gemini"] --> O["MERGE status=ok"]
  O --> A{"ligne insérée ?"}
  R --> A
  A -- "1 ligne" --> OK["200 ok / rejected"]
  A -- "0 ligne : l'autre livraison a gagné" --> D

Pourquoi ne pas laisser Pub/Sub dédupliquer ? Sa livraison « exactement une fois » n'existe que pour les abonnements pull, et seulement dans une région ; un abonnement push, comme celui-ci, reste « au moins une fois ». Et même en pull, elle ne couvre pas la panne qui survient entre votre écriture et l'acquittement.

Quelle clé pour quel doublon

L'ID de message protège contre les relivraisons de Pub/Sub. Il ne protège pas contre deux publications du même événement : si deux notifications identiques existent sur le bucket, chaque dépôt produit deux messages, avec deux ID différents, et deux lignes ok.

La lecture par génération complète le dispositif : chaque écriture d'un objet Cloud Storage reçoit un numéro de génération, transmis dans l'événement. Le worker lit blob(name, generation=…), donc la version qui a déclenché l'événement, même si l'objet a été remplacé entre-temps.

La mise en œuvre

def handle(event: Event, store: Store, reader: Reader, analyze: Analyzer) -> str:
    if store.seen(event.message_id):
        return "duplicate"  # redélivrance : on n'appelle pas Gemini une seconde fois
    try:
        check(event)
        data = reader.read(event.bucket, event.name, event.generation)
    except Rejected as exc:
        return "rejected" if store.record(event, "rejected", reason=str(exc)) else "duplicate"
    analyse, usage = analyze(data, event.content_type)
    # Le seen() ci-dessus économise un appel Gemini ; c'est record() qui garantit l'unicité
    # (deux livraisons simultanées passent toutes les deux le seen()).
    return "ok" if store.record(event, "ok", analyse=analyse, usage=usage) else "duplicate"

Le MERGE côté BigQuery, paramétré (jamais de valeur concaténée dans le SQL) et plafonné en octets facturés :

        sql = (
            f"MERGE `{self.table}` T USING (SELECT @message_id AS message_id) S "
            f"ON T.message_id = S.message_id "
            f"WHEN NOT MATCHED THEN INSERT ({cols}) VALUES ({', '.join('@' + c for c in row)})"
        )
        ...
        job = self.client.query(sql, job_config=cfg)
        job.result()
        return (job.num_dml_affected_rows or 0) == 1

La table est regroupée (clustering) par message_id : la recherche ne lit que les blocs qui contiennent cet ID. Le test de la course simule deux livraisons qui ne se voient pas :

def test_livraisons_simultanees_un_seul_enregistrement():
    store, analyzer = FakeStore(), FakeAnalyzer()
    e = parse_envelope(envelope())
    store.seen = lambda _id: False  # simule la course : aucune ne voit l'autre
    assert sorted([handle(e, store, FakeReader(), analyzer), handle(e, store, FakeReader(), analyzer)]) == ["duplicate", "ok"]
    assert len(store.rows) == 1

Résultat obtenu (2026-10-16)

make test passe (16 tests). Sur Google Cloud, make e2e renvoie au worker une enveloppe portant le messageId d'un document déjà analysé, puis compte les lignes :

15:17:58 rejeu du message 22201064101151603 : {'http': 200, 'status': 'duplicate'}
15:23:20 OK   10 lignes, une par document
15:23:20 OK   aucun message_id en double
15:23:20 OK   rejeu acquitté comme doublon

Ce qui casse en production

• Notification créée deux fois. Un gcloud storage buckets notifications create relancé à la main, ou un script non idempotent : chaque document produit deux lignes ok et deux appels Gemini, sans un seul duplicate dans les journaux. Diagnostic : gcloud storage buckets notifications list gs://…. Parade : le test d'existence dans cloud-up.sh, et une alerte sur COUNT(*) > 1 GROUP BY object_uri.
• Conflits de DML concurrents. BigQuery limite les instructions qui modifient une même table en parallèle ; au-delà, elles attendent en file, puis échouent. Symptôme : des 500 avec concurrent update sous forte charge. Ici l'erreur remonte en 500 et Pub/Sub réessaie, ce qui suffit à quelques centaines de documents par jour. Au-delà de quelques documents par seconde, écrivez par l'API Storage Write en ajout seul et dédupliquez à la lecture (QUALIFY ROW_NUMBER() OVER (PARTITION BY message_id ORDER BY processed_at) = 1).
• Déduplication qui coûte plus que le traitement. Sans regroupement par message_id, chaque seen() lirait toute la table ; à 2 Go de table, 600 documents par jour, cela fait plus de 1 To lu par jour. Parade : clustering, et maximum_bytes_billed (100 Mo ici) qui transforme une dérive en erreur visible.
• Écrire avant d'analyser. Insérer une ligne « en cours » puis la compléter crée un état intermédiaire qu'une panne fige : la ligne existe, seen() répond oui, le document n'est jamais analysé. Parade : une seule écriture, à la fin.

La règle de décision

Une seule écriture, conditionnelle, à la fin du traitement ; une vérification préalable seulement si elle économise un appel coûteux. Choisissez la clé selon le doublon à absorber : messageId pour les relivraisons, bucket/objet#génération si plusieurs publications du même événement sont possibles. Surveillez le taux de duplicate : quelques pour mille est normal, quelques pour cent révèle un délai d'acquittement mal réglé.

Pratique

Exercice. Modifiez la déduplication pour qu'elle résiste aussi à une notification doublée, sans perdre les redépôts volontaires. Quelle clé, et quels fichiers changent ?

• J'explique pourquoi un abonnement push reste « au moins une fois ».
• Je sépare la vérification qui économise de l'écriture qui garantit.
• Je choisis la clé de déduplication selon l'origine des doublons.
• Je plafonne le coût de la déduplication (clustering, maximum_bytes_billed).