Workers

À quoi servent les workers, et pourquoi des process séparés

L’API HTTP d’OrchPay ne parle (presque) jamais directement aux PSP ni à Titreo pendant une requête. Un appel comme « débite la carte cadeau » ou « capture le PSP » est lent, faillible et doit être rejouable : on ne veut pas qu’un timeout réseau pendant une requête HTTP laisse une session dans un état incohérent. La solution retenue est le pattern outbox : pendant la transaction de base de données, le use case écrit une intention (une ligne dans la table outbox, par ex. DEBIT_GIFT) en même temps que la modification de la session. Cette écriture est atomique. Ensuite, plus tard et hors requête, un processus séparé lit ces intentions et exécute les vrais appels externes.

Les workers sont précisément ces processus séparés. Il y en a deux :

Worker	Rôle métier	Process / commande
OutboxPoller	Draine la file outbox : prend les entrées prêtes, appelle la logique d’orchestration, marque succès ou échec, replanifie en cas de retry.	pnpm worker:outbox
CleanupScheduler	Expire les sessions actives dépassées (TTL) : annule les cartes cadeaux déjà débitées, libère les holds PSP, émet le webhook d’expiration.	pnpm worker:cleanup

Ils tournent comme des processus Node distincts de l’API HTTP. Cela permet de les redémarrer, scaler ou crasher indépendamment du serveur web, et garantit qu’aucune logique externe lente ne bloque une requête entrante.

Distinction clé : ordonnanceur vs logique

Cette page documente les ordonnanceurs (OutboxPoller, CleanupScheduler) : la boucle, le rythme, le verrouillage, le backoff. Elle ne documente pas ce que fait réellement chaque action (DEBIT_GIFT, AUTHORIZE_PSP, rollback…). Cette logique métier vit dans ProcessOutboxEntry et est décrite dans L’orchestrateur. Le poller ne sait pas débiter une carte ; il sait quand et comment réessayer, et délègue le quoi à l’orchestrateur. Le contexte conceptuel complet est dans Outbox & orchestration.

flowchart LR
  HTTP["API HTTP (use case)"] -->|"écrit une intention"| OUT[("table outbox")]
  POLL["OutboxPoller (worker)"] -->|"listReady + claim"| OUT
  POLL -->|"execute(entry)"| PROC["ProcessOutboxEntry (orchestrateur)"]
  PROC -->|"appels réels"| EXT["PSP / Titreo / webhook marchand"]
  POLL -->|"markCompleted / markFailed"| OUT

Fichiers de cette zone

Fichier	Rôle	Exports clés
api/src/workers/outbox.ts	Point d’entrée (composition root) du worker outbox : charge la config, construit le container, instancie et démarre le OutboxPoller.	main() (module exécutable)
api/src/workers/cleanup.ts	Point d’entrée du worker cleanup : même composition, instancie et démarre le CleanupScheduler.	main() (module exécutable)
api/src/infrastructure/workers/outbox-poller.ts	Implémentation de l’ordonnanceur outbox : boucle, claim atomique, backoff, dead-letter.	OutboxPoller, OutboxPollerOptions, DEFAULT_BACKOFF_SCHEDULE_MS
api/src/infrastructure/workers/cleanup-scheduler.ts	Implémentation de l’ordonnanceur cleanup : boucle à intervalle fixe.	CleanupScheduler, CleanupSchedulerOptions

Les classes sont dans infrastructure/workers/ car ce sont des adaptateurs pilotants (ils orchestrent le temps et le process). Les use cases qu’elles appellent (ProcessOutboxEntry, CleanupExpiredSessions) vivent dans application/workers/ et restent purs au sens hexagonal — voir Architecture hexagonale et Couche Application.

OutboxPoller — l’ordonnanceur de la file

Points d’entrée et câblage (api/src/workers/outbox.ts)

Le fichier outbox.ts est un composition root complet et autonome : il refait, pour le worker, le même assemblage de dépendances que le serveur HTTP. Il :

1. charge la config (loadConfig()), exige PSP_CONFIG_KEK ;
2. ouvre la connexion Firebird, le chiffrement AEAD, les registres PSP / gift card (Titreo si config.titreo est présent, sinon le MockGiftCardAdapter) ;
3. appelle buildContainer(...) pour obtenir container.useCases.processOutboxEntry ;
4. instancie le poller et le démarre.

L’instanciation lit ses paramètres dans l’environnement (avec valeurs par défaut) :

const poller = new OutboxPoller(uow, container.useCases.processOutboxEntry, clock, {
  intervalMs: Number(process.env.OUTBOX_INTERVAL_MS ?? 2000),
  batchSize: Number(process.env.OUTBOX_BATCH_SIZE ?? 10),
})

L’arrêt est propre : SIGINT / SIGTERM déclenchent poller.stop() puis db.dispose() avant process.exit(0).

Deux jeux de valeurs par défaut

Les défauts dans outbox.ts (intervalMs: 2000, batchSize: 10) et ceux codés dans le constructeur de OutboxPoller (?? 2000, ?? 10) coïncident — mais ce sont deux endroits distincts. Si vous changez l’un, vérifiez l’autre. En pratique, les variables d’environnement OUTBOX_INTERVAL_MS / OUTBOX_BATCH_SIZE priment sur les deux.

La boucle : start(), loop(), tickOnce()

start() met running = true et lance loop(). La boucle est un while (this.running) qui, à chaque tour :

1. appelle tickOnce() (entouré d’un try/catch : une erreur de tick est loguée, jamais fatale) ;
2. attend intervalMs via un setTimeout mémorisé dans this.timer.

stop() repasse running à false et clearTimeout(this.timer) pour ne pas laisser de timer pendant.

Le cœur est tickOnce(), qui retourne le nombre d’entrées traitées :

async tickOnce(): Promise<number> {
  // 1) lister les entrées prêtes (status PENDING|FAILED, next_attempt_at <= now, attempts < max)
  const candidates = await this.uow.run(({ outbox }) =>
    outbox.listReady(this.clock.now(), this.batchSize),
  )
  let processed = 0
  for (const candidate of candidates) {
    // 2) claim ATOMIQUE : UPDATE ... SET status='PROCESSING', attempts=attempts+1 ... RETURNING
    const claimed = await this.uow.run(({ outbox }) =>
      outbox.claim(candidate.id, this.workerId, this.clock.now()),
    )
    if (!claimed) continue          // perdu la course contre un autre worker → on passe
    await this.handle(claimed)      // 3) execute + markCompleted / markFailed
    processed += 1
  }
  return processed
}

Le déroulé d’un tick :

sequenceDiagram
  participant P as OutboxPoller
  participant DB as "outbox (Firebird)"
  participant O as "ProcessOutboxEntry"
  P->>DB: listReady(now, batchSize)
  DB-->>P: candidates[]
  loop pour chaque candidate
    P->>DB: claim(id, workerId, now)
    alt claim gagné
      DB-->>P: entry (PROCESSING, attempts+1)
      P->>O: execute(entry)
      alt COMPLETED
        O-->>P: { status: "COMPLETED" }
        P->>DB: markCompleted(id, now)
      else FAILED ou throw
        O-->>P: { status: "FAILED", errorCode, errorMessage }
        P->>DB: markFailed(id, lastError, nextAttempt)
      end
    else claim perdu
      DB-->>P: null
      P->>P: continue (skip)
    end
  end

Le claim atomique sécurise le multi-worker

listReady puis claim sont deux étapes : entre les deux, un autre poller peut avoir pris la même entrée. Le claim est un UPDATE ... WHERE id=? AND status IN ('PENDING','FAILED') AND attempts < max_attempts RETURNING ... : un seul worker gagne la ligne, les autres reçoivent null et passent. C’est ce qui rend OrchPay sûr à plusieurs instances de worker, sans verrou applicatif. Chaque poller a un workerId (poller-<uuid>) inscrit dans locked_by.

Traitement d’une entrée : handle()

handle(entry) appelle this.processor.execute(entry). Tout est protégé : si l’orchestrateur lève une exception, elle est convertie en { status: 'FAILED', errorCode: 'PROCESSOR_THROW', errorMessage } — le poller ne crashe jamais sur une entrée. Ensuite :

• COMPLETED → outbox.markCompleted(id, now), fin.
• FAILED → on compose lastError = "${errorCode}: ${errorMessage}", puis deux branches :
  • si entry.attempts >= entry.maxAttempts → dead-letter ;
  • sinon → markFailed(id, lastError, nextAttempt) avec nextAttempt calculé par le backoff. L’entrée repasse FAILED et sera reprise quand next_attempt_at <= now.

Schéma de backoff

Le rythme de retry suit une cadence prescrite par la spec (voir Outbox & orchestration), exportée comme DEFAULT_BACKOFF_SCHEDULE_MS :

Tentative échouée (n° d’attempt)	Délai avant prochaine tentative
1	1 minute
2	5 minutes
3	30 minutes
4	2 heures
5	12 heures
6 et au-delà	24 heures (dernier palier du barème)

computeNextAttempt(attempts) indexe ce barème par min(max(0, attempts - 1), schedule.length - 1) et plafonne à maxBackoffMs (24 h par défaut). Si aucun barème n’est fourni (cas configurable via backoffScheduleMs), il retombe sur un backoff exponentiel baseBackoffMs * 2^(attempts-1), lui aussi plafonné. Tous ces paramètres sont surchargeables via OutboxPollerOptions.

Dead-letter

Quand attempts >= maxAttempts, l’entrée est dead-lettée : on la marque FAILED avec un next_attempt_at projeté très loin dans le futur (maxBackoffMs * 365), ce qui la sort de fait du flux de reprise (listReady ne la sélectionnera plus avant des siècles). Le message est préfixé DEAD_LETTERED: .... Un log d’erreur structuré est émis (entryId, action, attempts, lastError), et un hook optionnel onDeadLetter(entry, lastError) peut être branché via les options (best-effort : s’il jette, c’est logué sans interrompre le poller).

Pas de table dead-letter dédiée

Le dead-letter n’est pas une table séparée : c’est une entrée FAILED dont le next_attempt_at est repoussé à l’infini. Pour rejouer une entrée dead-lettée, il faut remettre son next_attempt_at <= now (et éventuellement réinitialiser attempts) directement en base. Aucun mécanisme de rejeu automatique n’existe en Phase 1.

Options (OutboxPollerOptions)

Option	Défaut	Rôle
intervalMs	2000	Pause entre deux ticks de la boucle.
batchSize	10	Nombre max d’entrées listées par listReady à chaque tick.
backoffScheduleMs	DEFAULT_BACKOFF_SCHEDULE_MS	Barème de retry (le tableau ci-dessus).
baseBackoffMs / maxBackoffMs	1000 / 24 h	Base et plafond du fallback exponentiel et du plafond général.
workerId	poller-<uuid>	Identité inscrite dans locked_by lors du claim.
logger	écriture stdout/stderr	Logger structuré injectable.
onDeadLetter	—	Callback optionnel au dead-letter.

CleanupScheduler — l’expiration des sessions

Implémentation (api/src/infrastructure/workers/cleanup-scheduler.ts)

CleanupScheduler est volontairement plus simple que le poller : pas de claim, pas de backoff. Sa boucle (start() / loop() / stop()) est calquée sur celle du poller, mais tickOnce() se contente d’appeler le use case et de loguer s’il y a eu du travail :

async tickOnce(): Promise<void> {
  const result = await this.cleanup.execute({ batchSize: this.batchSize })
  if (result.expired > 0) this.log.info({ msg: 'cleanup tick', ...result })
}

Toute la logique réelle est dans le use case CleanupExpiredSessions (api/src/application/workers/cleanup-expired-sessions.ts), décrit côté Couche Application. En résumé, pour chaque session active dont le TTL est dépassé (listExpiredActive(now, batchSize)), il : annule les legs gift_card au statut CAPTURED (enqueue CANCEL_GIFT), libère les legs psp au statut HELD (enqueue RELEASE_HOLD), passe la session à expire(now), et enfile le webhook marchand correspondant. Il retourne un CleanupResult (scanned, expired, enqueuedCancelGifts, enqueuedReleases).

Le cleanup ne fait pas les appels externes, il enfile

Le scheduler ne contacte ni Titreo ni les PSP. Il enfile des entrées outbox (CANCEL_GIFT, RELEASE_HOLD) que l’OutboxPoller exécutera ensuite. Les deux workers se complètent : cleanup détecte l’expiration et planifie la compensation, le poller la réalise. C’est pourquoi en production les deux doivent tourner.

Options (CleanupSchedulerOptions)

Option	Défaut	Rôle
intervalMs	60_000 (1 min)	Pause entre deux scans d’expiration.
batchSize	50	Nombre max de sessions expirées traitées par tick.
logger	écriture stdout/stderr	Logger structuré injectable.

Le point d’entrée api/src/workers/cleanup.ts lit CLEANUP_INTERVAL_MS (défaut 60_000) et CLEANUP_BATCH_SIZE (défaut 50) dans l’environnement, et arrête proprement le scheduler sur SIGINT / SIGTERM.

Comment lancer les workers

En local (sans Docker)

# depuis app/api/
pnpm worker:outbox    # = tsx src/workers/outbox.ts
pnpm worker:cleanup   # = tsx src/workers/cleanup.ts

Les deux exigent un .env valide (Firebird joignable, PSP_CONFIG_KEK défini). Chaque commande est un process à part : pour drainer la file et expirer les sessions, lancez les deux dans deux terminaux.

Avec Docker (dev)

Le fichier docker-compose.dev.yml (à la racine app/) déclare un service worker-outbox :

worker-outbox:
  build: { context: ./api, dockerfile: Dockerfile, target: dev }
  env_file: [./api/.env]
  environment:
    OUTBOX_INTERVAL_MS: '1500'
    # ... FB_HOST, REDIS_URL, etc.
  command: ['pnpm', 'worker:outbox']
  depends_on:
    firebird: { condition: service_started }
    redis: { condition: service_healthy }
    api: { condition: service_started }

Ce service ne démarre que l’OutboxPoller (intervalle ramené à 1,5 s pour un dev réactif). Il dépend de Firebird, Redis et de l’API.

Le worker cleanup n’a pas de service compose dédié

docker-compose.dev.yml ne définit que worker-outbox ; il n’existe pas de service worker-cleanup. En environnement Docker dev, l’expiration des sessions n’est donc pas planifiée automatiquement : il faut lancer pnpm worker:cleanup manuellement (par ex. dans un conteneur api-cli) ou ajouter un service dédié. La commande pnpm worker:cleanup existe bien dans api/package.json ; c’est seulement l’orchestration compose qui ne l’inclut pas encore.

Pour aller plus loin

• Ce que fait réellement chaque action outbox (débit gift, autorize/capture PSP, rollback compensatoire, webhooks) : L’orchestrateur.
• Le pattern outbox, la garantie de cohérence et le rationale du backoff : Outbox & orchestration.
• Les transitions de statut des sessions et des legs déclenchées par ces traitements : Machines à états.
• La table outbox (colonnes, index, locked_by/locked_at) : Base de données.
• Pourquoi ProcessOutboxEntry et CleanupExpiredSessions restent purs et injectés : Architecture hexagonale et Couche Infrastructure.