PD-3 — Plan d'implémentation¶

1. Découpage en composants¶

2. Flux techniques¶

FN-01 : Traitement nominal d'une tâche¶

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Producer   │────▶│    Redis    │────▶│   Worker    │────▶│  Completed  │
│ (API/Cron)  │     │   (Queue)   │     │  (BullMQ)   │     │   (State)   │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │                   │
       │              LPUSH job           BRPOPLPUSH          SET state
       │              SET state:waiting   SET state:active    completed
       └───────────────────┴───────────────────┴───────────────────┘
                         Redis Persistence (AOF)

1. Le producer enregistre un job dans la queue via queue.add()
2. Redis persiste le job (AOF appendonly)
3. Un worker prend le job via BRPOPLPUSH (atomique)
4. L'état passe de waiting → active
5. Le worker exécute la logique métier
6. L'état passe à completed ou failed
7. L'état final est persisté et consultable

FN-02 : Reprise après interruption¶

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Worker    │     │    Redis    │     │   Systemd   │
│  (crash)    │     │ (persisted) │     │  (restart)  │
└─────────────┘     └─────────────┘     └─────────────┘
       │                   │                   │
       X crash             │                   │
       │              Job in active list       │
       │                   │                   │
       │                   │            restart worker
       │                   │◀──────────────────┤
       │              stalled check            │
       │              move to waiting          │
       │                   │                   │
       └───────────────────┴───────────────────┘

1. Un worker crash pendant le traitement
2. Le job reste dans la liste active de Redis
3. Systemd redémarre le worker automatiquement
4. BullMQ détecte le job "stalled" (lockDuration expirée)
5. Le job est automatiquement remis en waiting ou passé en failed

2bis. Diagrammes Mermaid¶

Graphe de dépendances des composants¶

graph TD
    subgraph infra["Infra (ProbatioVault-infra)"]
        REDIS["Redis Server<br/>AOF persistence, pub/sub"]
        SYSTEMD["Systemd<br/>Restart=always"]
        ALERTINFRA["Alert Infra<br/>Slack/email"]
    end

    subgraph backend["Backend (ProbatioVault-backend)"]
        BULLMQ["BullMQ Module<br/>jobs.module.ts"]
        REGISTRY["Queue Registry<br/>queue.registry.ts"]
        MONITOR["Job Monitor<br/>job-monitor.service.ts"]
        EVENTS["Job Events Listener<br/>job-events.listener.ts"]
        CONTROLLER["Jobs Controller<br/>jobs.controller.ts"]
        SERVICE["Jobs Service<br/>jobs.service.ts"]
        ALERT["Alert Service"]
    end

    BULLMQ -->|connection| REDIS
    BULLMQ -->|enregistre queues| REGISTRY
    SERVICE -->|orchestre| BULLMQ
    SERVICE -->|checkDataIntegrity| REDIS
    CONTROLLER -->|expose API| SERVICE
    MONITOR -->|observe états| BULLMQ
    EVENTS -->|on failed/stalled| BULLMQ
    EVENTS -->|signale anomalie| ALERT
    ALERT -->|notifie| ALERTINFRA
    SYSTEMD -->|restart workers| BULLMQ

Séquence FN-01 : Traitement nominal d'une tâche¶

sequenceDiagram
    participant P as Producer (API/Cron)
    participant S as Jobs Service
    participant R as Redis (Queue)
    participant W as Worker (BullMQ)
    participant E as Job Events Listener
    participant M as Job Monitor

    P->>S: queue.add(jobName, payload)
    S->>R: LPUSH job + SET state:waiting
    R-->>S: jobId

    W->>R: BRPOPLPUSH (atomique)
    R-->>W: job data
    R->>R: SET state:active
    E->>E: on('active') → log INFO

    W->>W: Exécution logique métier

    alt Succès
        W->>R: SET state:completed
        E->>E: on('completed') → log INFO
    else Échec
        W->>R: SET state:failed
        E->>E: on('failed') → log ERROR
        E->>E: Emit alerte via Alert Service
    end

    M->>R: GET job state (observabilité)
    M-->>M: Expose via API /jobs/:queue/:jobId/status

Séquence FN-02 : Reprise après interruption (stalled job recovery)¶

sequenceDiagram
    participant W as Worker (BullMQ)
    participant R as Redis (Queue)
    participant SD as Systemd
    participant B as BullMQ Stalled Check
    participant E as Job Events Listener

    W->>R: BRPOPLPUSH → state:active
    W->>W: Traitement en cours...
    W-xW: CRASH (kill -9, OOM)

    Note over R: Job reste en liste active<br/>lockDuration court

    SD->>W: Restart=always → relance worker

    B->>R: stalledInterval check
    B->>R: Détecte lock expirée
    R->>R: Move job: active → waiting
    E->>E: on('stalled') → log WARN

    W->>R: BRPOPLPUSH (re-traitement)
    W->>R: SET state:completed
    E->>E: on('completed') → log INFO

3. Mapping invariants → mécanismes¶

4. Mapping critères d'acceptation → mécanismes¶

5. Mapping tests (TC-*) → mécanismes + observables¶

6. Gestion des erreurs¶

Mécanisme ERR-03 : Force majeure (obligatoire)¶

En cas de perte de données exceptionnelle (crash Redis sans AOF, corruption), le système doit :

1. Détecter l'incohérence : Au redémarrage, comparer le dernier checkpoint connu avec l'état Redis
2. Marquer les jobs affectés : État incomplete avec metadata { reason: 'FORCE_MAJEURE', detectedAt, lastKnownState }
3. Émettre un signal explicite : Log CRITICAL [JobsService] FORCE_MAJEURE: X jobs in incomplete state
4. Alerter les opérateurs : Alerte dédiée avec liste des jobs impactés

// jobs.service.ts - Mécanisme ERR-03
async checkDataIntegrity(): Promise<ForceMajeureReport | null> {
  const lastCheckpoint = await this.getLastCheckpoint();
  const currentState = await this.getAllJobStates();

  const incompleteJobs = this.detectIncompleteJobs(lastCheckpoint, currentState);

  if (incompleteJobs.length > 0) {
    // Marquer les jobs comme incomplete
    for (const job of incompleteJobs) {
      await this.markAsIncomplete(job.id, 'FORCE_MAJEURE');
    }

    // Log explicite (ERR-03)
    this.logger.error(`FORCE_MAJEURE: ${incompleteJobs.length} jobs in incomplete state`, {
      jobIds: incompleteJobs.map(j => j.id),
      detectedAt: new Date().toISOString(),
    });

    // Alerte ops
    await this.alertService.sendCriticalAlert('FORCE_MAJEURE', incompleteJobs);

    return { incompleteJobs, detectedAt: new Date() };
  }
  return null;
}

Observable TC-ERR-03 : Endpoint GET /jobs/health/integrity retourne { status: 'degraded', incompleteJobs: [...] } si force majeure détectée.

Codes d'erreur applicatifs¶

export enum JobErrorCode {
  JOB_NOT_FOUND = 'JOB_NOT_FOUND',
  JOB_ALREADY_COMPLETED = 'JOB_ALREADY_COMPLETED',
  JOB_PROCESSING = 'JOB_PROCESSING',
  QUEUE_NOT_FOUND = 'QUEUE_NOT_FOUND',
  REDIS_CONNECTION_ERROR = 'REDIS_CONNECTION_ERROR',
  FORCE_MAJEURE = 'FORCE_MAJEURE', // ERR-03
}

export enum JobState {
  WAITING = 'waiting',
  ACTIVE = 'active',
  COMPLETED = 'completed',
  FAILED = 'failed',
  DELAYED = 'delayed',
  INCOMPLETE = 'incomplete', // ERR-03: Force majeure
}

7. Impacts sécurité¶

Journalisation¶

• Création de job : INFO [JobsService] Job created: {jobId, queue, timestamp}
• Completion : INFO [Worker] Job completed: {jobId, duration}
• Failure : ERROR [Worker] Job failed: {jobId, error, attempt}
• Stalled : WARN [BullMQ] Job stalled: {jobId, queue}

8. Hypothèses techniques¶

9. Points de vigilance (risques, dette, pièges)¶

Dette technique identifiée¶

• Pas de métriques Prometheus → à ajouter post-MVP

Note : L'API /jobs/:id/status est obligatoire dès le MVP pour satisfaire CA-02 (observabilité). Bull Board est optionnel mais l'endpoint API est requis.

10. Hors périmètre¶

• Logique métier des jobs : Implémentée dans chaque consumer (PD-21, etc.)
• Choix hébergeur Redis : OVH retenu mais hors scope spec
• Redis Cluster : Non requis pour volumétrie initiale
• Scheduler avancé : Cron jobs simples, pas de scheduling complexe
• Multi-tenant queues : Une instance Redis pour toute l'app

11. Implémentation technique¶

Phase 1 : Infrastructure Redis¶

Fichiers Terraform :

# infra/terraform/modules/redis/main.tf
resource "ovh_cloud_project_database" "redis" {
  service_name = var.ovh_project_id
  engine       = "redis"
  version      = "7.0"
  plan         = "essential"
  nodes {
    region = "GRA"
  }
}

Configuration Redis :

# Persistence
appendonly yes
appendfsync everysec

# Memory
maxmemory-policy noeviction

# Security
requirepass ${REDIS_PASSWORD}
bind 127.0.0.1

Phase 2 : Module BullMQ (Backend)¶

Structure :

src/modules/jobs/
├── jobs.module.ts           # Module principal
├── jobs.controller.ts       # API observabilité (CA-02)
├── jobs.service.ts          # Service d'orchestration
├── dto/
│   └── job-status.dto.ts    # DTO état job
├── queues/
│   ├── queue.registry.ts    # Registre des queues
│   └── queue.config.ts      # Configuration par défaut
├── processors/
│   └── base.processor.ts    # Classe de base workers
└── monitoring/
    ├── job-monitor.service.ts
    └── job-events.listener.ts

Configuration BullMQ :

// jobs.module.ts
@Module({
  imports: [
    BullModule.forRootAsync({
      imports: [ConfigModule],
      useFactory: (config: ConfigService) => ({
        connection: {
          host: config.get('REDIS_HOST'),
          port: config.get('REDIS_PORT'),
          password: config.get('REDIS_PASSWORD'),
        },
        defaultJobOptions: {
          removeOnComplete: { age: 86400, count: 1000 }, // 24h ou 1000 jobs
          removeOnFail: { age: 604800 }, // 7 jours
          attempts: 3,
          backoff: {
            type: 'exponential',
            delay: 1000,
          },
        },
      }),
      inject: [ConfigService],
    }),
  ],
})
export class JobsModule {}

Phase 3 : Monitoring et Alertes¶

Event Listener :

@Injectable()
export class JobEventsListener {
  private readonly logger = new Logger(JobEventsListener.name);

  @OnQueueEvent('failed')
  onFailed(job: Job, error: Error) {
    this.logger.error(`Job ${job.id} failed: ${error.message}`, {
      jobId: job.id,
      queue: job.queueName,
      error: error.stack,
      attempt: job.attemptsMade,
    });
    // Emit alert via AlertService
  }

  @OnQueueEvent('stalled')
  onStalled(jobId: string) {
    this.logger.warn(`Job ${jobId} stalled`);
  }
}

API Endpoint d'observabilité (obligatoire pour CA-02) :

// jobs.controller.ts
@Controller('jobs')
export class JobsController {
  constructor(private readonly jobsService: JobsService) {}

  /**
   * Retourne l'état d'un job (CA-02: observabilité)
   * @returns { id, queue, state, progress, failedReason?, timestamps }
   */
  @Get(':queue/:jobId/status')
  async getJobStatus(
    @Param('queue') queue: string,
    @Param('jobId') jobId: string,
  ): Promise<JobStatusDto> {
    return this.jobsService.getJobStatus(queue, jobId);
  }

  /**
   * Liste les jobs d'une queue avec leur état
   */
  @Get(':queue')
  async listJobs(
    @Param('queue') queue: string,
    @Query('state') state?: JobState,
  ): Promise<JobStatusDto[]> {
    return this.jobsService.listJobs(queue, state);
  }
}

Phase 4 : Documentation Ops (CA-05 obligatoire)¶

Runbooks requis avec critères d'exécutabilité :

Structure obligatoire de chaque runbook :

# [Titre]
## Prérequis
- Liste des accès/outils nécessaires
## Diagnostic
- Commandes pour identifier le problème
## Résolution
- Étapes numérotées avec commandes copiables
## Vérification
- Commande(s) de validation du succès
## Rollback
- Procédure de retour arrière si échec

Protocole de validation TC-DOC-01 :

1. Sélection testeur : Personne non-experte Redis/BullMQ (ex: dev frontend, PM)
2. Environnement : Staging avec accès SSH configuré
3. Exécution : Testeur suit le runbook sans aide extérieure
4. Mesures :
5. Temps d'exécution (chronomètre)
6. Nombre de questions posées (doit être 0)
7. Résultat final (succès/échec)
8. Validation : Runbook validé si exécuté en ≤ temps max sans question
9. Traçabilité : Checklist signée avec date, nom testeur, temps mesuré

Observable TC-DOC-01 : Fichier docs/validation/runbooks-validation.md contenant les résultats de validation signés

12. Checklist de validation¶

• Redis déployé avec AOF enabled
• BullMQ module configuré avec retry/backoff
• Event listeners sur failed/stalled
• API /jobs/:queue/:jobId/status opérationnelle (CA-02 obligatoire)
• Endpoint /jobs/health/integrity opérationnel (ERR-03 obligatoire)
• Mécanisme checkDataIntegrity() implémenté avec état incomplete
• Systemd Restart=always pour tous services
• Tests TC-NOM-01 et TC-NOM-02 passants
• Tests TC-INV-02 (observabilité API) passants
• Tests TC-ERR-03 (force majeure) passants
• Alertes configurées sur failures
• Runbooks rédigés avec structure obligatoire (CA-05)
• Runbooks validés par testeur non-expert (TC-DOC-01)
• Fichier runbooks-validation.md signé avec temps mesurés