La crate pg-jobs peut-elle remplacer BullMQ/Redis pour réécrire tsundoku en Rust ?

tsundoku exige que la priorité du parent soit propagée à tous les enfants d'un flow (buildSerieInserterFlow). Le builder Flow expose désormais cette priorité, à deux niveaux.

Ce qui a été ajouté(1) Flow::priority(node, p) — setter par nœud, calqué sur dedup_key/node_metadata, stocké en Option<i32> sur FlowNodeDef (un priority(node, 0) explicite n'est donc pas confondu avec "non défini"). (2) submit_flow(flow, metadata, default_priority) — un défaut flow-wide propagé à tout nœud laissé sans priorité, résolu en node.priority.or(default_priority).unwrap_or(0). (3) rerun_flow corrigé : il SELECT/INSERT maintenant priority (auparavant remise à 0 au re-run). Couvert par 4 tests (par nœud, défaut flow-wide, 0 explicite préservé, préservation au rerun).

Note sémantique de portageLe "higher first" de pg-jobs (HIGH=10) est un choix délibéré et normal : c'est la convention naturelle d'un ORDER BY priority DESC (src/store.rs:699) et celle de la plupart des ordonnanceurs (un plus grand nombre = plus prioritaire). C'est BullMQ qui adopte la convention inverse "lower first" (HIGH=1). Aucun changement à faire dans pg-jobs : seule la couche de portage mappe les valeurs BullMQ (p. ex. 10 - p) à l'enqueue.

En pg-jobs, une « queue » n'est qu'une colonne texte sur les jobs — pas un objet à « oblitérer » comme le bucket Redis de BullMQ. « Purger une queue » = « supprimer en masse les jobs où queue=X » : le vrai primitif est job-level, une suppression par filtre dont « toute une queue » et purge-all/obliterateAllQueues ne sont que des cas. Ce primitif existe désormais.

Ce qui a été ajoutéJobStore::purge(filter, policy) (src/store.rs). (1) Filtre PurgeFilter { queue, kind, status, flow_id } — chaque champ Some est ANDé ; tout-à-None purge tout (le cas obliterateAllQueues). (2) Cœur = sémantique de remove() : pour chaque job, resolve_dependents_conn(tx, id, false) avant le DELETE, donc le FK cascade ne laisse jamais un pending_deps faussé — jamais de DELETE WHERE queue=X brut qui casserait les flows. (3) running explicite via RunningPolicy { Skip, Cancel } : Skip les laisse (comptés dans PurgeReport::skipped_running), Cancel les supprime aussi (le worker ne tient aucun verrou de ligne ; son complete/fail guardé no-ope sur la ligne disparue) — pas de force-kill silencieux. (4) retour PurgeReport { deleted, skipped_running }. Une seule transaction.

DurcissementsORDER BY id sur le SELECT … FOR UPDATE (ordre de verrouillage stable, anti-deadlock purge-vs-purge) ; en mode Skip seules les lignes non-running sont verrouillées (ne bloque pas le complete/fail des workers actifs) ; skipped_running recompté après la boucle, pour qu'un running annulé par la cascade de la purge elle-même ne soit pas mal compté. Couvert par 9 tests (par queue/kind/status/flow_id, Skip, Cancel + cascade hors filtre, Ignore qui débloque, filtre vide, et le chemin recompte-après-cascade).

La dedup existait déjà : JobSpec::dedup_key (builder) et Flow::dedup_key (par nœud) + l'index unique partiel (queue, dedup_key) WHERE status in (pending,running) → au plus un job en vol par (queue, dedup_key), l'enqueue retournant l'id existant en cas de conflit (src/store.rs:85-134). C'est le besoin réel de tsundoku (« one-active-per-key »). Le résidu n'était pas de la dedup mais un besoin de lecture par clé — désormais couvert.

Ce qui a été ajouté(1) JobStore::get_by_dedup_key(queue, dedup_key) — le sibling de get(id) mais keyé par identité de dedup : il ne renvoie Some que tant qu'un job de cette clé est en vol (status in ('pending','running')), l'index unique partiel garantissant au plus une ligne → Option<JobRow> propre. C'est exactement le getJob(jobId) de contrôle du merge tsundoku, à ceci près qu'on interroge par clé et non par id (src/store.rs). (2) JobFilter gagne un champ dedup_key : Tracking::list_jobs filtre donc par clé tous statuts confondus (historique inclus), là où le lookup en vol ignore les jobs terminés. 4 tests verrouillent : lookup en vol, scoping par queue, terminé → None côté lookup mais visible côté list_jobs, et filtre dedup_key sur list_jobs.

Idempotence post-terminaisonVolontairement laissée au domaine applicatif : un job completed/dead de même clé n'empêche pas un re-enqueue (l'index ne couvre que pending/running). Réimplémenter le deduplication{ttl} de BullMQ (une fenêtre après succès) reproduirait un artefact du TTL Redis ; le bon découpage est un check métier (« déjà fait ») côté app — et le filtre dedup_key historique ci-dessus en fournit précisément la brique de lecture.

pg-jobs a 5 statuts (pending/running/completed/dead/cancelled, src/types.rs), exposés par queue_stats/list_jobs (src/tracking.rs). cancelled est un vrai état terminal (un job/flow non-running peut être annulé), là où BullMQ en fait ~ de la suppression (encore le contexte Redis) — pg-jobs est plus riche ici. Les sous-états « fins » d'un dashboard se dérivent des colonnes, non matérialisés — c'est désormais fait dans l'API typée, sans nouvelle colonne ni statut stocké.

Sous-états dérivésdelayed (planifié) = pending + run_at futur + attempts=0 ; retry (backoff après échec) = pending + run_at futur + attempts>0 ; waiting-children = pending + pending_deps>0 ; ready (éligible) = pending + pending_deps=0 + run_at dû ; active = running ; failed = dead (retries épuisés ou panic fatal) ; plus completed/cancelled. Pas de « prioritized » : priority n'est qu'un ordre d'affichage entre jobs d'un même type, pas un état.

Ce qui a été ajouté(1) un enum SubState { Delayed, Retry, WaitingChildren, Ready, Active, Failed, Completed, Cancelled } (src/tracking.rs) qui encode le besoin (le sous-état) et non la représentation — les définitions ci-dessus vivent dans le SQL, pas dans des statuts stockés. Les sous-états sont mutuellement exclusifs et partitionnent tout job (leur somme = le total) ; pour un pending, waiting-children prime sur delayed/retry (un job bloqué par ses deps n'est pas planifiable, quel que soit son run_at). (2) JobFilter gagne un champ sub_state: Option<SubState> : Tracking::list_jobs peut donc lister un sous-état donné (p.ex. tous les jobs delayed). (3) Tracking::substate_counts(filter) -> SubStateCounts renvoie le décompte complet par sous-état sur le scope du filtre (queue/kind/flow_id/group_key/dedup_key). Le grain job ET flow tombe du même code : passer flow_id scope le décompte à un flow. Non cassant — JobFilter dérive Default. 4 tests verrouillent : partition exhaustive des 8 sous-états, précédence waiting-children>delayed, filtre list_jobs par sous-état, décompte scopé par flow.

flow_view(flow_id) était mono-flow (outcome + compteurs d'un seul flow, src/tracking.rs). Il n'existait aucune API pour lister les flows — or un dashboard flow-first doit énumérer/paginer les flows (les flow_id distincts) avec leur outcome, filtrables (par outcome, queue, date…). Le contournement était un N+1 (flow_view par flow) ou un agrégat custom. Désormais couvert par une API typée, sans nouvelle colonne ni migration (tout se dérive de jobs.job/jobs.flow).

Ce qui a été ajouté(1) Tracking::list_flows(filter) -> Vec<FlowSummary> (src/tracking.rs) — le jumeau de list_jobs au grain flow. Il énumère les flow_id distincts de jobs.job (donc couvre tout flow ayant des jobs, avec ou sans ligne jobs.flow), LEFT JOIN jobs.flow pour le nom, le tout en une requête (CTE d'agrégat), trié newest-first. (2) chaque FlowSummary porte flow_id, name: Option<String> (None pour un flow d'avant les display names, même sémantique LEFT JOIN que list_jobs), l'outcome dérivé (précédence identique à flow_view : un nœud actif → Running, sinon un cancelled → Failed, sinon un dead → CompletedWithFailures, sinon Completed — verrouillé par un test croisé list_flows↔flow_view), les timestamps created_at (début = plus vieux job) / finished_at (fin = dernier job terminé, None tant que le flow tourne), et le SubStateCounts complet par flow — les sous-états dérivés (card ci-dessus) agrégés par flow, tombant de la même requête. (3) un FlowFilter (dérive Default, donc non cassant) : queue en sémantique EXISTS (un flow touchant la queue est gardé, mais ses compteurs restent ceux du flow entier — un DAG cross-queue n'est pas rogné, via HAVING bool_or), plus outcome, name (préfixe littéral insensible à la casse ; les métacaractères LIKE sont échappés) et created_after/created_before (sur le début du flow). limit<=0 → 100, offset borné à 0. (4) FlowOutcome gagne as_str/from_db (miroir de JobState) et dérive Clone/Copy. 11 tests verrouillent : un summary par flow, les 4 outcomes + accord avec flow_view, EXISTS-queue à compteurs pleins, préfixe de nom (dont l'échappement du _), fenêtre de dates, finished_at terminal-only, partition des sous-états par flow, pagination newest-first, nom None pour flow legacy.

L'ancien flow_view ne renvoyait qu'un agrégat plat ({ name, outcome, total, completed, failed, cancelled }) — un sous-ensemble strictement plus pauvre qu'une ligne de list_flows, et incapable de dessiner le DAG (nœuds + arêtes). Un dashboard flow-first a besoin de la vue détail d'un flow : son résumé et son graphe de dépendances. C'est désormais ce que flow_view fournit, en une seule méthode — sans nouvelle table (le graphe se dérive de jobs.job + jobs.job_dependency).

Ce qui a été ajouté — changement cassant(1) signature flow_view(flow_id) -> Result<Option<FlowView>> : None quand le flow n'a aucun job (id inconnu ou tout GC'd) — corrige le défaut latent où un flow absent renvoyait faussement outcome=Completed. (2) FlowView refondu en { stats: FlowSummary, graph: FlowGraph }. stats est exactement le FlowSummary d'une ligne list_flows — calculé via le même agrégat SQL partagé (une const FLOW_AGG_COLUMNS), donc outcome/compteurs ne peuvent pas diverger entre les deux API (verrouillé par un test croisé). (3) graph = FlowGraph { nodes: Vec<FlowNode>, edges: Vec<FlowEdge> }. FlowNode est une projection légère par job (job_id, name, kind, queue, status, sub_state dérivé, attempts, finished_at, last_error) — sans payload/output pour tenir à ~1000 nœuds ; le détail profond d'un nœud se lit à la demande via get(id). FlowEdge { prerequisite, dependent, on_failure } vient de jobs.job_dependency. (4) SubState et FlowOutcome gagnent from_db (parsing des discriminants du CASE). Le sub_state par nœud suit la même précédence que substate_counts (waiting_children > delayed/retry > ready). 4 tests dédiés (stats+graphe A→B, None pour flow inconnu, stats == ligne list_flows, sous-états par nœud) + migration des tests existants.

Le blocage — par volumequeue_metrics agrégeait finished_at par minute/heure/jour en live sur jobs.job, avec deux défauts qui ne tenaient pas au volume de tsundoku (~10k jobs/jour, hyperneo pire) : (1) l'historique des métriques = la rétention des jobs (le GC supprime → les métriques disparaissent), alors qu'on veut purger vite ET garder le débit longtemps ; (2) aucun index sur finished_at → la requête scanne à chaque rafraîchissement. Désormais levé.

Mécanisme retenu — sweep, pas per-finishBullMQ précalcule via un compteur Redis par job fini (INCR O(1) atomique) — forme Redis. Le transposer en UPSERT count+1 par complétion = hot-row contention + écriture dans la tx de complétion. Choix PG-idiomatique retenu : une table queue_metric alimentée par un sweep batch (INSERT…SELECT…GROUP BY…ON CONFLICT depuis un watermark finished_at), réutilisant la machinerie GC (advisory-lock, single-runner). Métriques désormais indépendantes du GC.

Ce qui a été ajouté(1) migration 0011_metrics_rollup.sql : table jobs.queue_metric(queue, bucket_ts, granularity, status, count) — la colonne granularity ('minute'/'day') est dans la PK pour désambigüiser un bucket minute à minuit d'un bucket jour ; curseur singleton jobs.metrics_rollup_state(watermark) ; index partiel job_finished_idx sur finished_at (corrige le défaut (2), sert le sweep et le tail live). Additive, non cassante. (2) JobStore::rollup_sweep(settle_lag, max_window, downsample_after, batch) (src/store.rs), calquée sur gc : fenêtre half-open [watermark, cutoff) avec cutoff = min(now−settle_lag, watermark+max_window), avance du watermark dans la même tx → exactly-once (chaque finished_at dans exactement une fenêtre) ; advisory-lock distinct de celui du GC (les deux passes peuvent courir en //) ; le settle_lag évite qu'une tx committée tardivement soit ratée ; renvoie un RollupReport {swept, downsampled, watermark}. (3) downsampling intégré à la même passe : replie les lignes minute au-delà de downsample_after en lignes jour (insert agrégé + delete borné par batch, atomiques → un lecteur ne voit jamais les deux granularités). (4) Tracking::queue_metrics (signature inchangée) refondue en lecture unifiée auto-stitch : rollup pour la partie settled (< watermark) recollée à la requête live pour le tail (le bucket à cheval somme les deux moitiés, sans doublon ni trou), et zero-fill dense (chaque bucket × les 3 statuts terminaux). Tant que le sweep n'a pas tourné (watermark à l'epoch) tout est servi live → comportement identique out-of-the-box. (5) les deux passes de maintenance deviennent des jobs détachés built-in planifiables (voir carte suivante). 8 tests verrouillent : exactly-once + re-sweep no-op, settle_lag, accumulation += sur bucket à cheval, no-op sous lock tenu, downsampling minute→jour, auto-stitch rollup+live, zero-fill, et le job qui alimente le rollup.

En rendant le rollup natif, une question de cohérence a émergé : pourquoi le GC serait-il un arm de boucle worker dédié (Worker::with_gc/GcConfig + un select!) quand le rollup, lui, est un job ? Choix tranché : un seul mécanisme. Les deux passes de maintenance sont désormais des jobs détachés built-in — GcJob (kind pg_jobs.gc) et MetricsRollupJob (kind pg_jobs.metrics_rollup), module pg_jobs::maintenance — que l'appelant enregistre et planifie via le Scheduler, exactement comme ses propres jobs. La crate détient le SQL (les primitives JobStore::gc / rollup_sweep, chacune advisory-lockée donc single-runner fleet-wide et idempotente sous at-least-once), l'appelant détient la cadence.

Prérequis par exigence de portagetechniquement le pattern update_data marchait (tsundoku le faisait), donc ce n'était pas un blocage dur — mais porter ce smell était refusé, donc l'ajout natif était un prérequis. Désormais implémenté : le smell n'a plus lieu d'être.

D'abord, lever la confusion : Delayed ≠ retry ≠ rate-limit. Le rate-limit proactif (cap de débit par queue) est couvert par set_queue_rate (src/store.rs:600). Cette card parle du report réactif : JobError::Delayed repousse un job sans consommer d'attempt (reschedule_current rend l'attempt) — pour un 429, un cache pas prêt, etc. C'est un report, pas un échec.

Le manque, résoluil n'y avait aucun suivi uniforme du nombre de reports d'un job ni de son max : la crate ne bornait pas les Delayed. Le contournement — compter dans la payload via update_data — était un smell : il mutait l'input du job, ce qui (a) polluait le type d'Input avec une préoccupation infra (rate_limit_retry_attempt dans le schéma tsundoku), (b) obligeait à y penser à la main, (c) n'était pas lock-token guardé.

Ce qui a été ajouté — symétrie exacte avec les retries(1) deux colonnes (0010_deferrals.sql, migrations/) : deferrals (compteur, incrémenté par le moteur dans reschedule_current — le miroir du claim qui incrémente attempts) et max_deferrals (budget), toutes deux not null default 0 donc backfill instantané et non cassant. (2) Job::max_deferrals() / JobSync::max_deferrals() (const par kind, comme max_attempts ; défaut 0 = illimité, l'ancien comportement) + builder JobSpec::max_deferrals(n) pour l'override par enqueue. (3) le dépassement dead-letter le job (deferrals ≥ max_deferrals → dead), exactement la transition « exhausted » que fail() applique à max_attempts, décidée dans le moteur (src/engine.rs) et tracée en PG (correct multi-worker via le guard lock_token). (4) le compteur est exposé sur JobRow.deferrals et via Ctx::deferrals()/SyncCtx::deferrals() pour une borne dynamique côté handler (escalade du Retry-After, ou Fatal au-delà d'un seuil) par-dessus le budget statique. (5) JobFilter gagne un champ de scope min_deferrals: Option<i32> : Tracking::list_jobs liste les jobs deferrés ≥ n fois et substate_counts ventile la population deferrée par sous-état. (6) symétrie sur les re-drive : requeue/requeue_dead/requeue_flow remettent deferrals=0 avec attempts (un re-drive manuel = cycle de vie neuf, budget rendu), et rerun_flow clone max_deferrals avec max_attempts. 11 tests verrouillent (8 d'intégration + 3 unitaires) : bornage → dead-letter à N, illimité (0) ne meurt jamais, compteur exposé sur JobRow, ctx.deferrals() pilotant une borne dynamique, filtre min_deferrals, no-op quand l'ownership est perdu, reset du budget au re-drive, défaut/override du trait + builder.

Non requis par tsundoku, mais implémenté : Queue::requeue_flow(flow_id) reprend un flow échoué en place — re-drive les nœuds dead + cancelled vers pending, recalcule pending_deps pour restaurer l'ordre du DAG, et préserve les nœuds completed (et leurs outputs). C'est exactement le chaînon que requeue/requeue_dead ne traitaient pas (ils ne recalculent pas les dépendants Cascade) ; alternative en place à rerun_flow (qui clone tout sous un nouveau flow_id). Refuse (Error::FlowBusy(flow_id)) si un nœud tourne encore.

BullMQ donne à chaque job un nom libre (queue.add('recompute-dedup-${id}', …)) qui sert d'étiquette dans le dashboard. pg-jobs route par KIND typé (const par type), séparé du nom d'affichage : c'est le bon découpage — routing type-safe d'un côté, label cosmétique de l'autre, là où BullMQ confond les deux. Implémenté aux deux échelles, en restant strictement cosmétique — jamais le routing (KIND), le dispatch ni le dedup (dedup_key).

Par jobcolonne name (NOT NULL) dérivée du trait : Job::name(&input) / JobSync::name(&input) (miroir de group_key, label par instance depuis l'input typé), défaut = KIND donc tout job porte une étiquette ; surcharge ponctuelle par le builder JobSpec::name(..) et, en flow, par Flow::node_name(node, ..). Surfacée sur JobRow.name (String, toujours présent). Pas de coalesce : la garantie « toujours un nom » vit dans le défaut du trait.

Par flowFlow::new(name) n'est plus jeté : submit_flow persiste le flow comme entité de premier ordre dans une nouvelle table jobs.flow (flow_id reste une référence souple, sans FK), exposée via FlowView.name. Le GC reape les lignes jobs.flow orphelines une fois le dernier job collecté.

Par schedulel'API scheduler est passée type-safe (upsert_every::<J> / upsert_cron::<J> + variantes _sync) : kind et payload viennent de J + un input typé. Chaque job émis est labellisé par J::name(&input) (précalculé à l'upsert dans jobs.schedule.job_name), exactement comme un enqueue direct ; le name d'identité du schedule reste la clé d'upsert/remove.