Le contrat de livraison

Première version, écrite en une après-midi : une fonction déclenchée par le bucket appelle Gemini et insère une ligne dans BigQuery. Trois semaines plus tard, la comptable signale que la facture de carburant de 4 278,83 € apparaît deux fois dans le tableau de bord, et qu'une relance d'impayé n'y est jamais arrivée. Les deux défauts ont la même cause : le code supposait qu'un événement arrive exactement une fois. Pub/Sub promet autre chose, « au moins une fois », et tout le pipeline doit être écrit contre cette promesse.

Le mécanisme

Quand un objet est créé sous entrant/, Cloud Storage publie un message sur le sujet lab-pipeline-evt-docs. Le message porte des attributs (eventType=OBJECT_FINALIZE, bucketId, objectId, objectGeneration) et, en data, les métadonnées de l'objet en JSON. L'abonnement push transmet chaque message au worker par un POST HTTPS signé d'un jeton OIDC (OpenID Connect), puis attend la réponse :

• un code 2xx acquitte le message : Pub/Sub l'oublie ;
• tout autre code, ou pas de réponse avant le délai d'acquittement (ici 120 s), le laisse en attente : Pub/Sub le relivrera, avec un délai croissant, jusqu'au nombre maximal de tentatives, puis l'envoie dans la file de lettres mortes.

sequenceDiagram
  participant GCS as "Cloud Storage"
  participant PS as "Pub/Sub (push)"
  participant W as "Worker Cloud Run"
  GCS->>PS: OBJECT_FINALIZE (messageId 2220…603)
  PS->>W: POST / (tentative 1)
  W--xPS: 500 (quota Gemini)
  Note over PS: attente 10 s, puis 20 s, … jusqu'à 60 s
  PS->>W: POST / (tentative 2, même messageId)
  W-->>PS: 200 ok
  PS->>W: POST / (relivraison rare, même messageId)
  W-->>PS: 200 duplicate

Le code de réponse du worker est donc un contrat avec Pub/Sub. Celui du projet tient en quatre lignes :

Le point délicat est la deuxième ligne. Répondre 500 sur un document vide le ferait réessayer dix fois pour rien, puis l'enverrait en lettres mortes, où un humain le découvrirait sans explication. On l'acquitte et on écrit la raison dans la table : le rejet devient une donnée que le métier peut lire.

Notification Cloud Storage ou Eventarc ?

Eventarc sait aussi router les événements Cloud Storage vers Cloud Run. Les deux passent par Pub/Sub ; la différence est qui possède l'abonnement.

Ce projet choisit la notification parce que la politique d'échec (10 tentatives, délai de 10 à 60 s, sujet de lettres mortes) fait partie du livrable et doit être relue en revue de code, et parce que le filtre de préfixe évite une boucle décrite plus bas. Prenez Eventarc dès que le même worker doit réagir à plusieurs sources (un bucket, une table BigQuery, un journal d'audit) ou que l'équipe a déjà standardisé ses services sur le format CloudEvents.

La mise en œuvre

La lecture de l'enveloppe vit dans pipeline.py, sans dépendance HTTP, pour être testée seule :

def parse_envelope(body: dict) -> Event | None:
    """Enveloppe push -> Event. None = événement à ignorer (acquitté sans traitement)."""
    try:
        msg = body["message"]
        message_id = msg["messageId"]
        attrs = msg.get("attributes") or {}
    except (KeyError, TypeError) as exc:
        raise PoisonMessage(f"enveloppe Pub/Sub invalide : {exc!r}") from exc
    if attrs.get("eventType") != "OBJECT_FINALIZE":
        if "eventType" in attrs:
            return None  # suppression, métadonnées... : pas notre travail
        raise PoisonMessage("message sans attribut eventType : il ne vient pas de Cloud Storage")

Le point d'entrée HTTP traduit chaque issue en code de réponse :

    try:
        event = parse_envelope(await request.json())
    except (PoisonMessage, ValueError) as exc:
        log.warning("message empoisonné", extra={"fields": {**fields, "event": "message_empoisonne", "error": str(exc)}})
        return JSONResponse({"status": "poison"}, status_code=400)
    if event is None:
        return JSONResponse({"status": "ignored"})
    ...
    try:
        status = await run_in_threadpool(handle, event, *deps())  # clients bloquants : hors boucle asyncio
    except Exception as exc:  # passager ou inconnu : Pub/Sub réessaiera
        log.exception("échec du traitement", extra={"fields": {**fields, "event": "document_echec", "error": repr(exc)}})
        return JSONResponse({"status": "error"}, status_code=500)

run_in_threadpool compte : les clients BigQuery et Gemini sont bloquants. Appelés directement dans une fonction async, ils gèleraient la boucle d'événements, et les trois autres requêtes que l'instance accepte (--concurrency=4) attendraient.

Lancez les tests du contrat (tests/test_main.py, tests/test_pipeline.py) :

cd files && make setup && make test

Résultat obtenu (2026-10-16)

$ make test
................                                                         [100%]
16 passed in 0.97s

Parmi eux : test_message_empoisonne_400_pour_finir_en_lettres_mortes, test_erreur_passagere_500_pour_reessai, test_evenement_autre_que_finalize_est_ignore.

Ce qui casse en production

• La boucle infinie. Le worker écrit un fichier dans le bucket qu'il surveille (un résumé, une copie) ; cette écriture déclenche un nouvel événement, qui déclenche une nouvelle écriture. Symptôme : la facture Vertex AI monte en quelques minutes. Parade : notification limitée à entrant/, et rien d'autre ne s'écrit sous ce préfixe ; les sources de Cloud Build passent par _build/.
• Délai d'acquittement trop court. Avec le délai par défaut de 10 s et un appel Gemini de 12 s, Pub/Sub relivre le message alors que la première livraison travaille encore. Symptôme : des duplicate dans les journaux sans aucune erreur. Parade : --ack-deadline supérieur au --timeout du service, ici 120 s pour les deux.
• Un 500 pour une erreur définitive. Un PDF chiffré renvoie une erreur 400 de Gemini ; si le worker répond 500, Pub/Sub le réessaie dix fois, dix appels facturés. Parade : traduire les erreurs définitives en Rejected (leçon 3).
• Événements non désirés. Une notification sans --event-types reçoit aussi les suppressions et les mises à jour de métadonnées. Parade : filtrer à la source, et acquitter sans traitement ce qui n'est pas OBJECT_FINALIZE (réponse ignored).

La règle de décision

Écrivez la table « situation → code de réponse » avant le code. Toute erreur que réessayer ne corrigera pas est acquittée et tracée ; toute erreur passagère reçoit un 5xx ; une enveloppe illisible reçoit un 4xx et finit en lettres mortes. Choisissez la notification Cloud Storage pour une source unique dont vous voulez versionner la politique d'échec, Eventarc pour plusieurs sources ou un format d'événement commun.

Pratique

Exercice. Un collègue propose de répondre 200 à tout, y compris en cas d'erreur, « pour ne jamais bloquer la file », et de logguer les erreurs. Décrivez ce que devient un document dont l'analyse a échoué sur un 429 de Vertex AI, puis proposez la correction.

• J'écris la table « situation → code de réponse » d'un consommateur push.
• Je distingue erreur définitive (acquittée et tracée) et erreur passagère (5xx).
• Je règle le délai d'acquittement par rapport au délai du service.
• J'argumente notification Cloud Storage contre Eventarc pour un cas donné.