Beaucoup d’applications ont besoin d’exécuter du travail « en arrière-plan » : imports de fichiers, créations en masse, envois d’e-mails, génération de rapports…

Le premier réflexe est souvent un simple process asynchrone; le deuxième, le deuxième, une table de tâches sur laquelle on réinvente laborieusement le verrouillage et la synchronisation ; le troisième, un composant dédié comme Redis, RabbitMQ ou SQS.

Pourtant, entre ces options, Postgres sait faire nativement le plus dur et vous permettant de garder une architecture simple : une table, un worker, et une requête bien choisie.

Ce tutoriel construit pas à pas une file de jobs robuste avec Postgres.

On explique les bases au fur et à mesure, puis on ajoute l’observabilité et les rejeux en cas d’erreur.

pending → running → done → fail

1. Pourquoi Postgres

Utiliser la base que vous avez déjà comme file de jobs apporte plusieurs avantages concrets :

  • Une brique de moins à opérer. Pas de broker à déployer, monitorer, sauvegarder, sécuriser et faire monter en version. C’est autant de complexité opérationnelle en moins.
  • La transactionnalité gratuite. Vous pouvez insérer un job et modifier vos données métier dans la même transaction. Si la transaction échoue (rollback), le job n’existe jamais. Avec un broker externe, ce couplage est un problème récurrent (le fameux dual-write problem).
  • De la visibilité. Un job est une ligne. Vous pouvez le lire, le filtrer, le rejouer, l’auditer avec du SQL ordinaire. Debug d’une file Redis à 3 h du matin vs SELECT * FROM jobs WHERE status = 'FAILED' : le choix est vite fait.

Le prix à payer, c’est le débit (throughput) : Postgres n’est pas conçu pour des millions de messages par seconde.
On y reviendra à la section 7.
Pour l’écrasante majorité des applications, quelques dizaines à quelques milliers de jobs par minute, c’est amplement suffisant.

2. La table jobs

Voici le schéma :

CREATE TABLE IF NOT EXISTS jobs (
    id                  BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id           UUID NOT NULL REFERENCES tenants (id), -- multi-tenancy
    type                TEXT NOT NULL,
    operated_by         UUID,
    status              TEXT NOT NULL DEFAULT 'PENDING'
                        CHECK (status IN ('PENDING', 'RUNNING', 'COMPLETED',
                                          'COMPLETED_WITH_ERRORS', 'FAILED')),

    -- Entrées immuables du run
    config              JSONB NOT NULL DEFAULT '{}'::jsonb,
    -- Compteurs + payloads, fusionnés au fil de l'exécution
    result              JSONB NOT NULL DEFAULT '{}'::jsonb,

    error               TEXT,
    trace_id            TEXT,

    -- Timing
    started_at          TIMESTAMPTZ,
    finished_at         TIMESTAMPTZ,
    duration_ms         BIGINT,

    created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at          TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX jobs_type_tenant_created_idx
    ON jobs (type, tenant_id, created_at DESC);

CREATE INDEX jobs_pending_idx
    ON jobs (created_at)
    WHERE status IN ('PENDING', 'RUNNING');

COMMENT ON TABLE jobs IS 'Runs de jobs asynchrones unifiés (imports, créations en masse, …).';
COMMENT ON COLUMN jobs.result IS 'Compteurs + payloads rows/items.';

Les choix qui comptent

Gestion de la clé primaire via des entiers séquencés
👉 GENERATED ALWAYS AS IDENTITY plutôt que BIGSERIAL.
C’est la façon moderne (SQL standard, depuis Postgres 10) de déclarer une clé auto-incrémentée, et elle corrige un défaut réel de SERIAL/BIGSERIAL.
Avec BIGSERIAL, la colonne a simplement un DEFAULT nextval(...) 👉 rien n’empêche du code d’écrire explicitement une valeur d’id, ce qui ne fait pas avancer la séquence.
Résultat, un jour, nextval renvoie une valeur déjà utilisée et vous prenez une erreur de clé dupliquée — la séquence est « désynchronisée ».

GENERATED ALWAYS interdit purement et simplement l’insertion manuelle d’id (elle lève une erreur), donc la séquence reste toujours la seule source de vérité.
La conséquence pratique côté application : dans vos INSERT, vous ne fournissez jamais la colonne id, vous la laissez à la base.

Si vous avez besoin ponctuellement de forcer un id (migration, import), GENERATED BY DEFAULT AS IDENTITY est la variante permissive. Pour une table de jobs, ALWAYS est le bon défaut : personne n’a de raison légitime d’imposer un id.

status comme machine à états.
Un job traverse PENDING → RUNNING → COMPLETED (ou COMPLETED_WITH_ERRORS / FAILED).
Le CHECK garantit qu’on ne peut pas écrire un statut invalide — la base refuse la ligne, pas besoin de faire confiance au code applicatif.

config vs result (deux JSONB).
La séparation est volontaire : config contient les entrées immuables du run (on ne les touche plus après création) ; result est mutable et grossit au fil de l’exécution (compteurs, lignes traitées, erreurs par item).
Garder les deux séparés évite de mélanger « ce qui a été demandé » et « ce qui s’est passé ».

Note sur JSONB. C’est pratique mais ce n’est pas une excuse pour tout y mettre. Ce qui sert à filtrer ou trier (status, type, tenant_id) doit rester en colonnes typées et indexées. Le JSONB, c’est pour le payload variable propre à chaque type de job.

L’index partiel jobs_pending_idx.
C’est le plus important pour la performance.
Un index partiel n’indexe que les lignes qui satisfont la clause WHERE — ici, uniquement les jobs actifs (PENDING/RUNNING).
Vos millions de jobs terminés ne sont pas dans cet index, qui reste donc petit et rapide même quand la table grossit indéfiniment.
C’est exactement ce que le worker interroge en boucle.

3. Le cœur du réacteur

Le vrai défi d’une file de jobs, c’est la concurrence : plusieurs workers tournent en parallèle et ne doivent jamais prendre le même job. La solution tient dans une clause SQL : FOR UPDATE SKIP LOCKED.

Le problème

Imaginez deux workers qui exécutent en même temps :

SELECT id FROM jobs WHERE status = 'PENDING' ORDER BY created_at LIMIT 1;

Les deux lisent le même job. Les deux le traitent. Le job s’exécute en double. 💥

La solution

Décortiquons le mécanisme :

  • FOR UPDATE verrouille les lignes sélectionnées jusqu’à la fin de la transaction. Un deuxième worker qui tenterait de verrouiller les mêmes lignes devrait attendre que le premier ait fini — ce qui sérialiserait tout le monde. Lent.
  • SKIP LOCKED change la donne : au lieu d’attendre, le deuxième worker ignore les lignes déjà verrouillées et prend les suivantes disponibles. Chaque worker repart aussitôt avec son propre lot, sans contention.

C’est précisément le comportement qu’on veut d’une file : « donne-moi les N prochains jobs que personne d’autre n’est en train de traiter ».

La requête de dequeue

UPDATE jobs
   SET status     = 'RUNNING',
       started_at = now(),
       updated_at = now()
WHERE id IN (
   SELECT id
   FROM jobs
   WHERE status = 'PENDING'
   ORDER BY created_at
   FOR UPDATE SKIP LOCKED
   LIMIT 10
)
RETURNING id, type, config

Ligne par ligne :

  • La sous-requête sélectionne jusqu’à 10 jobs PENDING, les plus anciens d’abord (FIFO via ORDER BY created_at), en posant un verrou et en sautant ce qui est déjà pris.
  • L’UPDATE les fait passer en RUNNING de façon atomique — dès la validation de la transaction, ils sont invisibles pour les autres workers (leur statut n’est plus PENDING).
  • RETURNING renvoie au worker ce dont il a besoin pour travailler, en un seul aller-retour réseau. Pas de SELECT puis UPDATE séparés.

Deux pièges classiques à éviter :

  • La casse des statuts. Si votre CHECK impose 'PENDING' en majuscules, un WHERE status = 'pending' (minuscules) ne matchera jamais aucune ligne. Restez cohérent partout.
  • Le bon statut cible. L’UPDATE doit passer le job à 'RUNNING', pas le laisser à 'PENDING' — sinon vous ne « prenez » jamais réellement le job et vous le re-sélectionnez à l’infini.

4. Le worker

Un worker, c’est une boucle : prendre des jobs → les exécuter → enregistrer le résultat → recommencer.

Voici une implémentation en Go, avec pgx. On commence par les types et le dequeue :

package worker

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"

    "github.com/jackc/pgx/v5/pgxpool"
)

type Job struct {
    ID     int64
    Type   string
    Config json.RawMessage
}

// Handler exécute un type de job et renvoie son résultat
type Handler func(ctx context.Context, config json.RawMessage) (json.RawMessage, error)

const dequeueSQL = `
    UPDATE jobs
       SET status     = 'RUNNING',
           started_at = now(),
           updated_at = now()
    WHERE id IN (
       SELECT id
       FROM jobs
       WHERE status = 'PENDING'
       ORDER BY created_at
       FOR UPDATE SKIP LOCKED
       LIMIT $1
    )
    RETURNING id, type, config
`

func dequeue(ctx context.Context, db *pgxpool.Pool, batchSize int) ([]Job, error) {
    rows, err := db.Query(ctx, dequeueSQL, batchSize)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var jobs []Job
    for rows.Next() {
        var j Job
        if err := rows.Scan(&j.ID, &j.Type, &j.Config); err != nil {
            return nil, err
        }
        jobs = append(jobs, j)
    }
    return jobs, nil
}

On enregistre les handlers par type au démarrage :

type Worker struct {
    db       *pgxpool.Pool
    handlers map[string]Handler
}

func NewWorker(db *pgxpool.Pool) *Worker {
    return &Worker{
        db:       db,
        handlers: make(map[string]Handler),
    }
}

func (w *Worker) RegisterHandler(jobType string, handler Handler) {
    w.handlers[jobType] = handler
}

func (w *Worker) Run(ctx context.Context, batchSize, pollInterval int) error {
    for ctx.Err() == nil {
        jobs, err := dequeue(ctx, w.db, batchSize)
        if err != nil {
            return fmt.Errorf("dequeue: %w", err)
        }

        for _, job := range jobs {
            w.process(ctx, job)
        }

        // Attendre avant la prochaine boucle
        select {
        case <-ctx.Done():
            return nil
        case <-time.After(time.Duration(pollInterval) * time.Millisecond):
        }
    }
    return nil
}

func (w *Worker) process(ctx context.Context, job Job) {
    handler, ok := w.handlers[job.Type]
    if !ok {
        w.finalize(ctx, job.ID, "FAILED", nil, 
            fmt.Sprintf("unknown job type: %s", job.Type), 0)
        return
    }

    start := time.Now()
    result, err := handler(ctx, job.Config)
    if err != nil {
        w.finalize(ctx, job.ID, "FAILED", nil, err.Error(), elapsedMs(start))
        return
    }

    w.finalize(ctx, job.ID, "COMPLETED", result, "", elapsedMs(start))
}

const finalizeSQL = `
    UPDATE jobs
       SET status      = $2,
           result      = result || $3::jsonb,
           error       = $4,
           finished_at = now(),
           duration_ms = $5,
           updated_at  = now()
    WHERE id = $1
`

func (w *Worker) finalize(
    ctx context.Context,
    id int64,
    status string,
    result json.RawMessage,
    errMsg string,
    durationMs int64,
) {
    if result == nil {
        result = json.RawMessage(`{}`)
    }
    var errVal *string
    if errMsg != "" {
        errVal = &errMsg
    }

    _, err := w.db.Exec(ctx, finalizeSQL, id, status, result, errVal, durationMs)
    if err != nil {
        log.Printf("finalize job %d: %v", id, err)
    }
}

func elapsedMs(start time.Time) int64 {
    return time.Since(start).Milliseconds()
}

Dans cette version simple, un échec de handler passe directement en FAILED. Quand on ajoutera les retries (section 6), ce chemin appellera plutôt une fonction de retry, qui décide entre re-mettre en file et abandonner.

Le dispatch par job.Type via la map handlers est ce qui rend la table unifiée : un seul mécanisme de file, plein de types de jobs différents (csv_import, bulk_create, send_report…), chacun avec son handler enregistré au démarrage.

L’opérateur || sur du JSONB fait un merge au niveau supérieur : utile si le job a déjà écrit des compteurs partiels dans result pendant son exécution et qu’on veut les compléter plutôt que les remplacer.

Sortir proprement des jobs bloqués

Un worker peut mourir (crash, OOM, déploiement) en laissant un job coincé en RUNNING pour toujours. Un petit reaper périodique remet ces zombies en file :

UPDATE jobs
   SET status = 'PENDING',
       updated_at = now()
WHERE status = 'RUNNING'
  AND started_at < now() - INTERVAL '15 minutes'

Adaptez l’intervalle à la durée max attendue de vos jobs. C’est aussi ici que l’idempotence (section 6) devient indispensable : un job re-mis en file sera réexécuté.

5. Observabilité

Une file invisible est une file qu’on ne debuggera pas. Trois leviers.

trace_id : relier le job au reste du système.
Quand une requête HTTP crée un job, propagez l’identifiant de trace (OpenTelemetry, ou un simple UUID de corrélation) dans la colonne trace_id. Quand le worker exécute le job, il rattache ses logs à ce même trace_id. Vous pouvez alors suivre une opération de bout en bout — depuis le clic de l’utilisateur jusqu’à l’exécution asynchrone, même plusieurs minutes plus tard.

duration_ms : mesurer, puis alerter.
En stockant la durée à chaque run, vous pouvez sortir des statistiques directement en SQL :

SELECT type,
       count(*)                                            AS runs,
       percentile_cont(0.50) WITHIN GROUP (ORDER BY duration_ms) AS p50,
       percentile_cont(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95,
       count(*) FILTER (WHERE status = 'FAILED')           AS failures
FROM jobs
WHERE finished_at > now() - INTERVAL '24 hours'
GROUP BY type
ORDER BY p95 DESC;

On pensera à rajouter l’index suivant :

CREATE INDEX jobs_stats_idx ON jobs (finished_at)
    INCLUDE (type, duration_ms, status)
    WHERE finished_at IS NOT NULL;

Ce genre de requête, branché sur un Grafana ou un rapport quotidien, vous dit tout de suite quel type de job dérive.

La profondeur de file : votre signal d’alerte n°1.
La métrique qui compte le plus est le nombre de jobs en attente. Si elle grimpe, vos workers ne tiennent pas la charge :

SELECT type, count(*) AS pending
FROM jobs
WHERE status = 'PENDING'
GROUP BY type;

Grâce à l’index partiel de la section 2, cette requête reste instantanée même sur une table énorme. Alertez au-delà d’un seuil.

6. Retries, idempotence et dead-letter

En production, les jobs échouent : timeout réseau, API tierce indisponible, pic de charge. Il faut retenter — mais intelligemment.

Compter les tentatives

On ajoute deux colonnes :

ALTER TABLE jobs
    ADD COLUMN attempts     INT NOT NULL DEFAULT 0,
    ADD COLUMN max_attempts INT NOT NULL DEFAULT 3;

Le worker incrémente attempts à chaque prise. En cas d’échec, deux issues :

func (w *Worker) onFailure(ctx context.Context, job Job, cause error) {
    if job.Attempts >= job.MaxAttempts {
        // abandon → dead-letter (statut FAILED)
        w.finalize(ctx, job.ID, "FAILED", nil, cause.Error(), 0)
        return
    }
    // on retente plus tard, avec backoff
    w.requeueWithBackoff(ctx, job.ID, job.Attempts, cause.Error())
}

(Cela suppose d’ajouter les champs Attempts et MaxAttempts au struct Job et de les lire dans le RETURNING du dequeue.)

type Job struct {
    ID          int64
    Type        string
    Config      json.RawMessage
    Attempts    int
    MaxAttempts int
}

const dequeueSQL = `
UPDATE jobs
   SET status     = 'RUNNING',
       started_at = now(),
       updated_at = now(),
       attempts   = attempts + 1     -- on compte la prise ici
WHERE id IN (
   SELECT id
   FROM jobs
   WHERE status = 'PENDING'
     AND run_after <= now()          -- respecte le backoff (voir plus bas)
   ORDER BY created_at
   FOR UPDATE SKIP LOCKED
   LIMIT $1
)
RETURNING id, type, config, attempts, max_attempts

Deux changements clés : attempts est incrémenté au moment de la prise (pas à l’échec), ce qui garantit qu’un worker qui crashe après avoir pris le job aura quand même consommé une tentative — sinon un job qui fait planter le worker à chaque fois tournerait à l’infini.

Backoff exponentiel

Retenter immédiatement contre une API en rade ne fait qu’aggraver les choses. On espace les tentatives. Il suffit d’ajouter une colonne run_after et de la respecter dans le dequeue :

ALTER TABLE jobs ADD COLUMN run_after TIMESTAMPTZ NOT NULL DEFAULT now();

Le WHERE du dequeue devient WHERE status = 'PENDING' AND run_after <= now().
Au requeue, on repousse run_after : now() + INTERVAL '1 minute' * power(2, attempts) donne 2, 4, 8 minutes…

Côté Go, la fonction qui remet le job en file en calculant le délai directement dans le SQL :

const requeueSQL = `
UPDATE jobs
   SET status     = 'PENDING',
       error      = $2,
       run_after  = now() + (interval '1 minute' * power(2, $3)),
       updated_at = now()
WHERE id = $1
`

func (w *Worker) requeueWithBackoff(
    ctx context.Context, id int64, attempts int, cause string,
) {
    if _, err := w.db.Exec(context.Background(), requeueSQL, id, cause, attempts); err != nil {
        log.Printf("requeue job %d: %v", id, err)
    }
}

Idempotence

Règle d’or : un job doit pouvoir être exécuté plusieurs fois sans effet de bord indésirable.

Avec les retries, un même job peut s’exécuter plusieurs fois (crash après finalisation, timeout long, etc.). Il faut donc que le handler soit idempotent :

  • Soit il est naturellement idempotent (INSERT … ON CONFLICT DO NOTHING)
  • Soit il trace ce qu’il a déjà fait (table de traces, statut externe) et saute si déja traité

Sans idempotence, les retries deviennent un multiplicateur de problèmes.

7. Limites

Quand votre charge et votre architecture commencent à scaler, n’utiliser que Postgres peut ne plus suffire.
Voici les signaux qui invitent à surveiller :

  • Le débit devient très élevé. Quand le polling des workers de Postgres devient le goulot d’étranglement, un broker dédié (Kafka, SQS, NATS) devient plus efficace et libère de la charge sur la base.
  • Le VACUUM peine. Une file très active génère beaucoup de morts (dead tuples) — chaque UPDATE de statut crée une nouvelle version de ligne. Il faut un autovacuum bien réglé, sinon la table gonfle (bloat) et ralentit. Archiver ou purger régulièrement les jobs terminés est une bonne pratique.
  • Il vous faut du fan-out / pub-sub. Si un même événement doit être consommé par plusieurs services indépendants, c’est le modèle d’un bus de messages (Kafka), pas d’une file de tâches.
  • Vous avez besoin de latence sub-milliseconde. Le polling introduit une latence égale, en moyenne, à la moitié de l’intervalle de poll. Pour du temps réel strict, ce n’est pas le bon outil (LISTEN/NOTIFY peut aider, mais avec ses propres limites).

Pour tout le reste — c’est-à-dire la plupart des besoins réels — une table jobs, FOR UPDATE SKIP LOCKED et un worker en boucle vous emmènent étonnamment loin, sans ajouter la moindre pièce mobile à votre infrastructure.

En résumé

  • Une table avec une machine à états (status) et un index partiel sur les jobs actifs.
  • FOR UPDATE SKIP LOCKED pour distribuer le travail entre workers concurrents sans doublon ni contention.
  • trace_id + duration_ms pour l’observabilité, et l’index partiel pour surveiller la profondeur de file à moindre coût.
  • attempts + backoff + idempotence pour des retries sûrs, et le statut FAILED comme dead-letter queue requêtable.
  • Pas de broker externe tant que vous n’avez pas dépassé les limites de la section 7, ce qui sera souvent le cas.

Le meilleur composant d’infrastructure est souvent celui que vous n’avez pas à ajouter.

Code exemple complet disponible sur github fourni avec cet article.