Le cas présenté ici vient d’un retour d’expérience d’une mission chez un client.

Il est transposé dans un domaine produits/prix.

Un fournisseur publie des mises à jour de tarif. Chaque évènement met en file un job PRICE_REFRESH pour le catalogue concerné, avec la liste des produits à recalculer. Si le moteur de recalcul refuse la demande parce qu’un autre recalcul est déjà en cours, le job est remis en file.

Le monitoring a remonté une erreur sur une remise en file :

duplicate key value violates unique constraint (SQLSTATE 23505)

Elle concernait un job d’environ 4 000 produits, passé en FAILED.
Un second job PRICE_REFRESH du même catalogue, d’environ 2 400 produits sans produit commun avec le premier, avait été inséré environ une milliseconde avant la réservation du premier. Le second job s’est terminé normalement. Les produits du premier n’ont jamais été recalculés.

Une régénération complète du catalogue a eu lieu peu après : l’impact réel était probablement nul sur ce catalogue.

Le même chemin perd des données dès qu’aucune régénération ne suit.

Pourquoi un job est remis en file

Le job ne recalcule pas les prix lui-même. Il demande au moteur de recalcul un recalcul partiel des produits concernés, puis passe en COMPLETED. Le recalcul s’exécute ensuite en arrière-plan.

Le moteur refuse la demande dans trois situations :

  • un des produits visés, ou une variante du même article, est déjà en cours de recalcul.

Ce recalcul peut venir d’une actualisation lancée à la main, d’un autre traitement automatique ou d’un job précédent dont le recalcul tourne encore ;

  • une régénération complète du catalogue est en cours ;
  • le moteur ne peut pas accepter de travail : file pleine ou arrêt en cours.

Dans les trois cas, le worker ne peut rien faire de plus avec le job qu’il vient de réserver. Il le repasse en PENDING pour qu’une réservation ultérieure le reprenne, dans la limite d’un nombre maximal de tentatives.

Dans l’exemple, ces refus correspondent aux erreurs ErrProductRefreshInFlight, ErrFullRefreshCatalogInFlight, ErrQueueFull et ErrRunnerStopped, que Worker.Process traite par une remise en file.

L’incident relève des deux premières situations : l’analyse du monitoring indique que le job a rencontré un recalcul en cours, sans préciser lequel.

Un job réservé sort de l’index partiel

La table jobs suit le modèle décrit dans Gérer des jobs asynchrones avec Postgres, avec une colonne lock_token.
Le worker la renseigne à la réservation, et ses écritures ultérieures la vérifient.
L’invariant « un seul job en attente par catalogue » est porté par un index unique partiel :

CREATE UNIQUE INDEX jobs_price_refresh_pending_catalog_uq
    ON jobs ((config->>'catalogId'))
    WHERE status = 'PENDING' AND type = 'PRICE_REFRESH';

La colonne config contient catalogId et productIds. L’unicité ne s’applique qu’aux lignes qui satisfont le prédicat[1].

Un job réservé passe en RUNNING et sort donc de l’index. Remettre le job en PENDING le fait rentrer dans l’index. Si un PENDING existe déjà pour ce catalogue, l’UPDATE viole la contrainte.

Pourquoi un second PENDING existe

Cela peut se produire dans l’enchaînement suivant :

  1. Le worker réserve le job A.
    Dans le code d’origine, la réservation est UPDATE … SET status = 'RUNNING' … WHERE id IN (SELECT … FOR UPDATE SKIP LOCKED).
    La transaction n’est pas encore validée.

  2. Un évènement fournisseur arrive.
    L’entrée d’index de A, en version PENDING, est valide depuis sa mise en file.
    La transaction de réservation, qui remplace cette version, n’est pas validée.
    L’INSERT … ON CONFLICT du job B attend sa fin, puis refait le contrôle d’unicité[2].

  3. La réservation est validée.
    A ne satisfait plus status = 'PENDING', et l’insertion de B réussit sans conflit.
    Tout évènement arrivé pendant que A est RUNNING crée de même un nouveau PENDING.

  4. Le worker lance le recalcul de A.
    Un recalcul est déjà en cours sur une partie de ses produits ou de leurs variantes, ou sur tout le catalogue : le moteur refuse la demande et A doit être remis en file.
    L’UPDATE vers PENDING viole l’index et A passe en FAILED.

Le job B est inséré une fois la réservation de A validée ; la remise en file de A viole ensuite l'index Job A Job B PENDING transaction non validée validation réservation : RUNNING INSERT attend B inséré : PENDING un PENDING existe déjà retour PENDING : 23505 A B A passe en FAILED
La ligne du bas indique quel job occupe l'index unique partiel. Le job B est inséré dès que la réservation de A est validée. Quand A doit revenir en PENDING, l'index contient déjà B.

L’insertion fusionnait déjà, pas la remise en file

L’insertion utilise INSERT … ON CONFLICT avec le prédicat de l’index et fusionne les ids dans le job existant :

INSERT INTO jobs (type, status, config)
VALUES ('PRICE_REFRESH', 'PENDING',
        jsonb_build_object('catalogId', $1::text, 'productIds', $2::jsonb))
ON CONFLICT ((config->>'catalogId'))
    WHERE status = 'PENDING' AND type = 'PRICE_REFRESH'
DO UPDATE SET config = jsonb_set(jobs.config, '{productIds}', (
    SELECT jsonb_agg(elem) FROM (
        SELECT jsonb_array_elements_text(jobs.config->'productIds') AS elem
        UNION
        SELECT jsonb_array_elements_text(excluded.config->'productIds')
    ) merged
))

La remise en file est un UPDATE, et ON CONFLICT appartient à INSERT[3].

Attraper le 23505 abandonne la transaction

Une erreur place la transaction en état « aborted ». Seul un ROLLBACK TO SAVEPOINT ou un rollback complet permet de continuer[4].

Alternatives examinées :

  • Savepoint : ajoute deux allers-retours, et la fusion reste à écrire.
  • MERGE[5] : face à l’insertion concurrente d’un doublon, il lève une violation d’unicité[6].
  • Advisory lock par catalogue : sérialise les écritures si tous les chemins qui écrivent un PENDING le prennent (non testé).

La solution en une requête, CTE par CTE

WITH running AS (
    SELECT id, config FROM jobs
    WHERE id = $1 AND status = 'RUNNING' AND lock_token = $2
    FOR UPDATE
), merged AS (
    UPDATE jobs pending SET config = jsonb_set(pending.config, '{productIds}', (
        SELECT jsonb_agg(elem) FROM (
            SELECT jsonb_array_elements_text(pending.config->'productIds') AS elem
            UNION
            SELECT jsonb_array_elements_text(running.config->'productIds')
        ) merged_ids
    ))
    FROM running
    WHERE pending.status = 'PENDING' AND pending.type = 'PRICE_REFRESH'
      AND pending.config->>'catalogId' = running.config->>'catalogId'
    RETURNING pending.id
)
UPDATE jobs SET status = 'COMPLETED',
                updated_at = now(),
                finished_at = now(),
                duration_ms = (extract(epoch FROM now() - started_at) * 1000)::bigint,
                result = $3::jsonb || jsonb_build_object('mergedIntoJobId', merged.id)
FROM merged
WHERE jobs.id = (SELECT id FROM running)
RETURNING merged.id

running vérifie que le job est toujours RUNNING avec le bon lock_token, et le verrouille. En READ COMMITTED, FOR UPDATE attend la fin d’un concurrent et réévalue le WHERE sur la nouvelle version de la ligne[7].
Si le job n’est plus RUNNING ou a changé de lock_token, running est vide.

merged fusionne les ids du job RUNNING dans le PENDING du même catalogue. running contient au plus une ligne, filtrée par clé primaire, donc UPDATE … FROM n’a pas de jointure multiple à arbitrer[8]. Le prédicat de merged reprend celui de l’index partiel, donc merged renvoie au plus une ligne, ce qui protège l’UPDATE … FROM merged final.

L’UPDATE final clôt le job RUNNING en COMPLETED, et non en FAILED : ses produits sont repris par le job en attente. Son result reçoit la note passée en paramètre ($3), qui documente la fusion, et l’id du job absorbeur dans mergedIntoJobId.
Les sous-instructions partagent un snapshot[9], sauf la ligne verrouillée, relue dans sa dernière version validée.

Côté Go : sql.ErrNoRows signifie « rien à fusionner »

func (r *Repository) RequeuePriceRefresh(
	ctx context.Context, jobID int64, lockToken string, result json.RawMessage,
) (int64, error) {
	mergedIntoJobID, err := r.mergeIntoPending(ctx, jobID, lockToken, result)
	switch {
	case err == nil:
		return mergedIntoJobID, nil
	case !errors.Is(err, sql.ErrNoRows):
		return 0, fmt.Errorf("merge job %d into the pending job of its catalog: %w", jobID, err)
	}
	if err := r.returnToPending(ctx, jobID, lockToken); err != nil {
		return 0, fmt.Errorf("requeue job %d: %w", jobID, err)
	}
	return 0, nil
}

mergeIntoPending exécute la requête de fusion, returnToPending l’UPDATE de repli, EnqueuePriceRefresh l’upsert d’insertion :

UPDATE jobs SET status = 'PENDING', lock_token = NULL, updated_at = now()
WHERE id = $1 AND status = 'RUNNING' AND lock_token = $2

Sans PENDING concurrent, l’UPDATE final touche zéro ligne. Le code bascule alors sur la remise en file classique, gardée elle aussi par status et lock_token.

Tests

Les tests du dépôt d’origine couvrent la course (un RUNNING et un PENDING plus récent sur le même catalogue, aucun id perdu, job remplacé COMPLETED avec mergedIntoJobId), l’absence de job concurrent, un PENDING sur un autre catalogue et une réservation perdue. L’exemple reprend ces cas sur postgres:18 et ajoute :

  • la naissance du second PENDING : l’insertion de B attend la validation de la réservation de A, puis s’insère, et la remise en file naïve de A lève le 23505 ;
  • les tableaux vides à l’insertion et à la remise en file (23502, puis variante COALESCE) ;
  • la course résiduelle ;
  • le worker, avec un moteur bouchonné : remise en file pour chacun des quatre refus, fusion dans le PENDING plus récent, échec à la limite de tentatives.

Limites

Course résiduelle

En READ COMMITTED, chaque instruction prend son propre snapshot.
Si B est inséré après le début de la fusion, merged ne le voit pas et renvoie zéro ligne. Le repli est une autre instruction : si B est validé entre-temps, il viole l’index. Dans le test, le repli lève le 23505, puis 25P02, et Commit renvoie un rollback. A reste RUNNING.

La fusion ne voit pas le job B, qui est validé avant le repli : le repli viole l'index et la transaction est annulée Transaction 1 Transaction 2 B n'existe pas encore fusion : 0 ligne INSERT B, COMMIT B est dans l'index repli : 23505 transaction annulée (25P02), A reste RUNNING
La transaction 1 remet le job A en file, la transaction 2 insère le job B. La fusion et le repli sont deux instructions : B est validé entre les deux, le repli viole l'index unique partiel et la transaction 1 est annulée.

Si B n’est pas encore validé, le repli attend sa fin, puis échoue si B valide[2]. Non testé.

Tableaux vides

jsonb_agg sur zéro ligne renvoie NULL, et jsonb_set est strict : une nouvelle valeur NULL donne un résultat NULL[10]. Si les deux tableaux sont vides, config devient NULL, et la colonne NOT NULL lève l’erreur 23502 dans le test de l’exemple. La requête publiée ne gère pas ce cas, et l’upsert d’insertion a la même limite, vérifiée par un second test. La variante COALESCE(jsonb_agg(elem), '[]'::jsonb), appliquée aux deux requêtes, passe les deux tests.

Avec deux tableaux vides, jsonb_agg renvoie NULL et config devient NULL ; COALESCE conserve un tableau vide Requête publiée [] UNION [] → 0 ligne → jsonb_agg : NULL → jsonb_set : NULL → 23502 Variante COALESCE [] UNION [] → 0 ligne → COALESCE(jsonb_agg, '[]') : [] → productIds = []
Avec deux tableaux vides, jsonb_agg renvoie NULL et l'écriture de config échoue. COALESCE remplace ce NULL par un tableau vide.

Ordre et type des ids

UNION ne garantit aucun ordre[11]. jsonb_array_elements_text renvoie du text : des ids numériques deviendraient des chaînes JSON.

UNION ne garantit pas l'ordre des ids, et jsonb_array_elements_text transforme des nombres en chaînes Ordre ["p2","p1"] UNION ["p3","p1"] → ["p1","p3","p2"] un ordre possible parmi d'autres Type [1, 2] → jsonb_array_elements_text → ["1", "2"] nombres devenus chaînes
L'ordre des ids fusionnés n'est pas garanti. Des ids numériques ressortiraient sous forme de chaînes.

Ce que la solution corrige et ce qui reste ouvert

La requête de fusion réduit le problème d’origine sans le supprimer.

  • Course. Avant le correctif, tout évènement arrivé pendant l’exécution du job A créait un second PENDING, et la remise en file de A échouait. La fenêtre durait toute l’exécution du job. Elle se limite maintenant à l’intervalle entre la fusion et le repli. Dans cet intervalle, l’erreur d’origine reste possible : la solution ne rejoue pas la séquence et ne sérialise pas les écritures. L’index unique partiel garantit toujours qu’il n’existe pas deux PENDING pour un même catalogue. Il ne garantit pas l’absence d’erreur.
  • Tableaux vides. La requête de fusion et l’upsert publiés échouent quand les deux tableaux sont vides. La variante COALESCE couvre ce cas.
  • Ordre et type des ids. La fusion ne conserve ni l’ordre ni le type des ids. Les ids de l’exemple sont des chaînes : seul l’ordre change.

Conclusion

Un index unique partiel porte l’invariant « un seul job en attente par catalogue ».
Une remise en file par simple UPDATE le viole dès qu’un second job en attente existe.

Une requête avec CTE modifiante lit le job en cours, fusionne ses ids dans le job en attente et clôt le job en cours, dans une seule instruction.
Le repli vers PENDING reste une seconde instruction, et une course subsiste entre les deux.

Reproduire les exemples

Comment rejouer les exemples

Code : pg-cte-and-advanced-upsert, dossier go/.

go vet ./...
go test ./...

Ressources

  1. PostgreSQL 18 : Partial Indexes
  2. PostgreSQL 18 : Index Uniqueness Checks
  3. PostgreSQL 18 : INSERT, clause ON CONFLICT
  4. PostgreSQL 18 : Transactions et SAVEPOINT
  5. PostgreSQL 18 : MERGE
  6. PostgreSQL 18 : Transaction Isolation, Read Committed
  7. PostgreSQL 18 : SELECT, The Locking Clause
  8. PostgreSQL 18 : UPDATE
  9. PostgreSQL 18 : WITH Queries
  10. PostgreSQL 18 : JSON Functions et Aggregate Functions
  11. PostgreSQL 18 : Combining Queries