08. Transactions, erreurs & câblage

1. La frontière transactionnelle#

La règle, correctement formulée#

Qui décide du commit / rollback ? Pas le repository. Sinon un use case qui fait deux écritures ne pourrait pas les rendre atomiques.

La formulation qu'on lit souvent — « la transaction, c'est la requête HTTP » — est trompeuse : elle est vraie de l'API, et fausse dès qu'un autre canal existe. La bonne formulation est :

La transaction entoure le use case. Chaque adaptateur entrant fournit cette frontière avec les moyens de son canal.

Les trois implémentations de ce projet#

Même principe, trois déclencheurs :

# 1. API — presentation/api/dependencies/session.py : une transaction par requête
async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    session_factory = request.app.state.session_factory
    async with session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# 2. CLI — presentation/cli/app.py : une transaction par commande
@asynccontextmanager
async def transactional_session() -> AsyncIterator[AsyncSession]:
    ...
    async with session_factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

# 3. Worker — presentation/worker/__main__.py : une transaction par message
async with session_factory() as session:
    use_case = NotifyTaskShared(...)
    try:
        await process_message(incoming, use_case, dlx)
        await session.commit()
    except Exception:
        await session.rollback()
        raise

Ces trois canaux fournissent une session au traitement, mais le moment de l’accusé de succès diffère. L’API actuelle committe après la réponse ; le CLI après la commande ; le worker acquitte dans process_message avant le commit du wrapper. Ce dernier ne fait actuellement que des lectures SQL. Pour un consumer qui écrit, le commit doit réussir avant l’ACK, puis le traitement doit tolérer une redélivrance.

Ce que ce choix te donne#

Un use case peut enchaîner plusieurs écritures dans une seule transaction atomique, sans le savoir. Il ne connaît ni commit, ni rollback, ni même AsyncSession — il ne voit que des ports.

C'est une forme légère d'Unit of Work : le pattern où une unité de travail regroupe les modifications et les valide d'un bloc. D'autres modèles existent — Unit of Work explicite passé au use case, décorateur @transactional, transaction script — et sont tout aussi défendables. Ce qui compte est que la responsabilité transactionnelle soit située et unique, pas la forme exacte.

Quand committer : avant ou après la réponse ?#

Les trois implémentations ci-dessus committent toutes dans la fermeture de la frontière. Pour la CLI et le worker, c'est sans ambiguïté : la commande ou le message est fini. Pour une requête HTTP, c'est un vrai embranchement.

Dans FastAPI, une dépendance yield exécute son code de sortie après l'envoi de la réponse par défaut : c'est la stratégie B. FastAPI documente un paramètre scope sur Depends — Depends(get_session, scope="function") fait tourner la sortie juste après la fonction de route, donc avant l'envoi : c'est la stratégie A. Vérifie le comportement de la version que tu utilises plutôt que de le supposer, en t'appuyant sur les scopes FastAPI.

Ce projet choisit A : presentation/api/dependencies/session.py expose Depends(get_session, scope="function"), donc la fermeture et son commit s'exécutent avant l'envoi de la réponse. Les providers qui dépendent de cette session doivent rester compatibles avec ce scope ; une réponse streamée utilise un cycle de vie séparé adapté à la durée du flux.

Pour les commandes, garde cette séquence :

POST · PUT · PATCH · DELETE
  use case
    └── flush là où une contrainte doit être traduite
  COMMIT
    └── puis construction et envoi du 2xx

Un 201 ou un 204 est une promesse : l'écriture est durable. Le flush ne la tient pas à lui seul, parce qu'un flush n'est pas un commit. Il fait remonter les violations que tu avais anticipées, à l'endroit qui sait encore les traduire — c'est son rôle, et il reste utile sous les deux stratégies (ch. 02). Mais une contrainte différée, un échec de sérialisation, un trigger, un timeout ou une connexion coupée échouent quand même au COMMIT, après un flush pourtant réussi.

B reste défendable pour les chemins en lecture seule, et pour les écritures dont la confirmation est explicitement provisoire — un 202 Accepted promet l'acceptation, pas la durabilité. Ce que B ne doit jamais être, c'est le défaut que tu n'as pas choisi.

Les limites à connaître#

• Effets non transactionnels. Un e-mail envoyé ou un message publié pendant la transaction ne se « rollback » pas. Le sujet est assez important pour avoir sa propre section : voir plus bas.
• Transactions longues. Une frontière calée sur la requête tient tant que les requêtes sont courtes. Un traitement de plusieurs minutes doit être découpé — et ce dépôt en contient un, l'import en masse, qui committe pendant le use case via un UnitOfWorkPort. C'est la seule entorse, elle est explicite, et elle est traitée au chapitre 10. La règle « un repository ne committe jamais », elle, ne bouge pas : le pouvoir de committer se demande par un port, il ne s'obtient pas par effet de bord.
• Un échec au commit peut arriver trop tard. L'endroit où vit le commit le décide. Ce projet choisit la stratégie A : l'échec remonte avant l'envoi. Un flush explicite reste utile pour déclencher et traduire une contrainte à l'endroit prévu — voir Quand committer. Le streaming, lui, doit conserver une session distincte ouverte pendant le flux.
• Il n'existe pas d'« après le commit » dans le use case. La frontière l'entoure : il rend la main avant le commit. Tout ce qui ne doit tourner qu'une fois les données persistées se raccroche donc à la frontière, ce qui suppose que celle-ci soit un objet auprès duquel on peut s'enregistrer — pas une simple dépendance yield. Et pour un effet qui sort du processus, on ne construit pas ce crochet du tout : on écrit une ligne d'outbox dans la transaction. Les trois cas sont détaillés au chapitre 11.
• PgBouncer en mode transaction. L'infrastructure de ce projet impose ses contraintes (pas de LISTEN/NOTIFY, pas de curseur WITH HOLD, cache de prepared statements désactivé).

2. Les effets externes : le problème du double write#

Tout ce qui précède suppose un monde simple : une transaction, une base, un commit. Le monde réel en sort dès qu'une opération doit aussi envoyer un e-mail, publier un message ou appeler une API tierce.

Le problème, posé précisément#

Une seule opération métier, deux systèmes qui ne partagent aucune transaction :

« créer et notifier » =  écrire en base        (PostgreSQL)
                        + publier un message    (RabbitMQ)

On ne peut pas les rendre atomiques. Il n'existe que deux ordres possibles, et les deux sont cassés :

C'est le dual write problem, et le point à comprendre est qu'il n'a aucune solution par réordonnancement. Déplacer la publication ne fait que choisir quel type de panne on préfère. Il faut changer de mécanisme.

L'outbox : rendre l'intention transactionnelle#

L'idée tient en une phrase : on n'écrit pas dans le broker, on écrit dans la base — dans la même transaction que la donnée métier.

CREATE TABLE outbox (
    id            uuid PRIMARY KEY,
    name          text        NOT NULL,   -- le type d'événement
    payload       jsonb       NOT NULL,
    created_at    timestamptz NOT NULL DEFAULT now(),
    published_at  timestamptz              -- NULL tant que non publié
);
CREATE INDEX ON outbox (created_at) WHERE published_at IS NULL;

# Le use case n'appelle plus le broker : il écrit une ligne.
async def execute(self, ...) -> None:
    task = await self._tasks.get(task_id)
    event = ShareTaskNotification(task_id=..., user_ids=..., ...)
    await self._outbox.add(event)      # même session, même transaction
    # le commit de la frontière transactionnelle rend les deux atomiques

Un relais séparé lit ensuite les lignes non publiées, les envoie au broker, et les marque publiées. La donnée et l'intention de publier deviennent atomiques : si la transaction échoue, il n'y a ni tâche partagée, ni ligne d'outbox.

Le coût que l'outbox déplace — il ne le supprime pas#

C'est le point que la plupart des présentations escamotent. Le relais peut mourir après avoir publié et avant d'avoir marqué la ligne. Au redémarrage, il republie.

Autrement dit : adopter l'outbox sans rendre le consommateur idempotent, c'est échanger une perte silencieuse contre des doublons silencieux. On n'a pas résolu le problème, on l'a déplacé chez le voisin.

Rendre un consommateur idempotent#

Idempotent signifie que rejouer le traitement n’ajoute pas d’effet métier. Il faut préciser aussi la réponse au doublon : Task.complete() lève TaskAlreadyCompleted au deuxième appel. Un consumer ne peut l’acquitter comme un succès que si ce comportement est explicitement voulu.

Pour des effets dans une même base, enregistrer l’identifiant du message et appliquer la mutation dans la même transaction :

# Illustratif : ces ports ne sont pas implémentés dans ce dépôt.
async with uow:
    inserted = await processed.try_insert(message_id)  # UNIQUE + ON CONFLICT DO NOTHING
    if inserted:
        await apply_database_change(event)
    await uow.commit()
# L'adaptateur acquitte après le commit.

En cas de crash avant commit, les deux écritures sont annulées ; après commit mais avant ACK, la redélivrance retrouve la clé et n’applique pas l’effet une seconde fois. La clé doit être stable et conservée assez longtemps.

Une clé métier est possible, mais doit identifier une même demande : (task_id, user_id, jour) supprimerait aussi deux partages légitimes le même jour. Le bon choix dépend du besoin, pas du transport.

Comment le relais lit-il l'outbox ?#

Un relais peut interroger périodiquement l’outbox ou utiliser LISTEN sur une connexion PostgreSQL directe et dédiée pour être réveillé. NOTIFY peut être émis depuis une transaction passant par PgBouncer. Les notifications ne sont pas un journal durable : le relais relit toujours l’outbox et prévoit un rattrapage après déconnexion.

Et si plusieurs agrégats sont modifiés ?#

C'est l'autre moitié de la question, et elle a une réponse plus simple.

Dans un monolithe sur une base unique, modifier deux agrégats dans une même transaction fonctionne, sans mécanisme particulier. C'est une dérogation assumée à « un agrégat par transaction » (ch. 04), et elle est parfaitement tenable tant que la base est partagée.

Dès que les agrégats vivent dans deux services, la transaction commune disparaît. Deux familles de réponses :

Une saga coordonne des transactions locales et leurs compensations, lesquelles peuvent elles-mêmes échouer et demander une reprise. L’outbox est un moyen de fiabiliser les publications d’une saga ; ce n’est pas une dépendance obligatoire de toute implémentation de saga.

Quant au 2PC / XA (validation en deux phases) : techniquement il résout le problème, en pratique il est écarté presque partout. Les brokers modernes le supportent mal, il maintient des verrous ouverts à travers plusieurs systèmes, et un coordinateur qui tombe laisse des transactions en suspens qu'il faut résoudre à la main.

En a-t-on vraiment besoin ?#

Deux questions suffisent à trancher, et elles ne portent pas sur la technique :

1. Que se passe-t-il si ce message est perdu ?
2. Que se passe-t-il si ce message arrive deux fois ?

Pour une notification d'e-mail sur le partage d'une tâche, les deux réponses sont « tolérable » : la ligne 1 s'applique, et l'absence d'outbox est un choix défendable. Pour un débit bancaire, la ligne 4 s'impose.

Une troisième option, souvent oubliée et parfois la meilleure : la réconciliation périodique. Un travail nocturne qui compare les deux systèmes et rattrape les écarts coûte bien moins cher qu'un outbox, et suffit quand la latence de correction est acceptable.

3. Deux familles d'erreurs#

Métier ≠ technique#

Le domaine lève des exceptions métier ; il ignore HTTP. Mais tout n'est pas métier : une base injoignable, un timeout SMTP, un broker down sont des échecs techniques. Les confondre conduit à renvoyer un 400 pour une panne réseau — et à faire croire au client que sa requête était fautive.

Ce projet maintient donc deux hiérarchies parallèles :

Le critère de tri est simple :

Le client peut-il corriger sa requête ? Oui → erreur métier (4xx). Non, c'est une dépendance qui est en panne → erreur technique (5xx).

Le mapping, une fois pour toutes#

L'astuce qui évite un handler par exception : mapper les bases sémantiques.

# presentation/api/error_handlers.py
@app.exception_handler(NotFoundError)
async def _not_found(_, exc): return _error(404, str(exc))

@app.exception_handler(ConflictError)
async def _conflict(_, exc): return _error(409, str(exc))

# + ValidationError → 422, DomainError → 400 (repli)

@app.exception_handler(InfrastructureError)
async def _service_unavailable(_, exc):
    logger.error("Dépendance technique indisponible : %s", exc, exc_info=exc)
    return _error(503, "Une dépendance technique est indisponible.")

Quatre détails qui comptent :

1. Un nouveau bounded context est couvert sans toucher ce fichier. Il suffit que ses exceptions héritent de la bonne base : FastAPI remonte la hiérarchie de classes pour trouver le handler.
2. 503, pas 500. Une dépendance externe indisponible est un problème transitoire et extérieur, pas une anomalie du service.
3. On journalise les erreurs techniques, pas les erreurs métier. Un 404 est un fonctionnement normal ; polluer les logs avec ferait perdre les vrais incidents.
4. On ne renvoie pas le détail technique au client. Le message d'une InfrastructureError peut contenir un hôte, un port, une trace — il reste dans les logs.

Et une règle de discipline qui découle du reste : jamais de HTTPException en dehors de presentation/. Lever un code HTTP dans un use case, c'est décider du protocole depuis une couche qui n'est pas censée le connaître — et casser immédiatement le CLI et le worker.

Où logger ?#

Dans ce projet, on évite les logs dans le domaine et on journalise aux frontières, où le contexte technique est disponible. Cela n’en fait pas une fonction pure au sens strict : les factories génèrent des identifiants et lisent l’horloge. Une horloge ou un générateur injectés seraient utiles pour maîtriser ces entrées dans des tests plus exigeants.

4. La composition root#

Quelque part, il faut relier le port abstrait (TaskRepository) à son implémentation (SqlAlchemyTaskRepository). Ce point de câblage s'appelle la composition root : le point où les implémentations concrètes sont assemblées avec les abstractions qu'elles réalisent.

Dans ce projet, c'est le package presentation/api/dependencies/, via l'injection de dépendances de FastAPI :

# dependencies/repositories.py — port ← adaptateur
def get_task_repository(session: SessionDep) -> TaskRepository:
    return SqlAlchemyTaskRepository(session)

# dependencies/task.py — use case ← ses ports
def get_create_task(tasks: TaskRepositoryDep, users: UserRepositoryDep) -> CreateTask:
    return CreateTask(tasks, users)

Son organisation est délibérée :

• session.py et repositories.py sont transverses ; task.py et user.py sont par contexte, et ne s'importent jamais entre eux ;
• l'__init__.py est une façade qui réexpose tout : les routers, l'app et les tests importent depuis ...api.dependencies. C'est légitime ici, parce que le rôle même d'une composition root est d'être un point d'entrée unique — ce n'est pas un motif à copier partout (ch. 12).

Ce n'est pas la seule organisation possible : d'autres projets placent la composition root dans un package bootstrap/ ou container/ à la racine, en particulier quand plusieurs adaptateurs driving doivent la partager. Ici, le CLI et le worker construisent leurs dépendances à la main — c'est acceptable à cette échelle, et ça deviendrait une duplication à surveiller si les use cases câblés se multipliaient.

C'est aussi ce qui rend les tests faciles : substituer un adaptateur ne touche à rien d'autre. Les tests d'intégration remplacent get_session par une dépendance équivalente pointant sur la base de test, et injectent un publisher inerte à la place de RabbitMQ — le reste du câblage reste authentique.

À retenir#

• La transaction entoure le use case ; chaque adaptateur entrant fournit la frontière (requête, commande CLI, message).
• Les repositories ne committent jamais et reçoivent leur session : c'est un Unit of Work léger, un choix parmi d'autres — mais la responsabilité doit être située quelque part, explicitement.
• Pour une commande HTTP, committe avant la réponse. Un 2xx est une promesse de durabilité, et un flush n'est pas un commit : il traduit les violations anticipées, il ne garantit pas que le commit final réussira.
• Les effets externes ne se rollback pas : c'est le dual write problem, et il n'a pas de solution par réordonnancement.
• L'outbox rend l'intention de publier transactionnelle, mais ne fournit qu'une publication at-least-once : elle rend l'idempotence du consommateur obligatoire.
• L'idempotence est une propriété du traitement, pas du message : opération naturellement rejouable, sinon déduplication garantie par une contrainte en base.
• Deux questions décident du besoin : que coûte une perte ? que coûte un doublon ?
• Deux hiérarchies d'erreurs : métier (4xx, non journalisées) et technique (503, journalisées, détail masqué).
• Jamais de HTTPException hors de presentation/ ; jamais de logger dans le domaine.
• La composition root concentre la décision d'association port ↔ adaptateur — pas la connaissance des deux mondes, que l'adaptateur a forcément.