Skip to content

Épique CVN-N014-ED — Transport de données inter-tâches (XCom / object-storage)

Code : CVN-N014-ED (série plateforme/runtime N014 — voisine de N014-EA FTF engine, N014-EB runtime invariants, N014-EC skills) OpenProject : wp#241 · GitHub : #1104 · Statut : Draft · Date : 2026-06-05 Une phrase : établir un standard projet unique pour le passage de données volumineuses entre tâches Airflow — object-storage XCom backend, pass-by-référence, piloté par seuil — et migrer le parc de façon non-disruptive (strangler-fig), pour que la classe de bug « transport bricolé par-tâche » disparaisse partout.

§0. Origine — l'incident déclencheur

L'épique naît d'un raté concret sur CVN-N001-EI-S05 / s43 : le transport cross-pod des prédictions était implémenté via np.savez vers un Path filesystem, résolu en /tmp pod-local (non partagé). Les 5 pods d'acquisition écrivaient chacun dans leur propre /tmp ; le pod gate lisait son /tmp vide → cohorte incomplète → INCONCLUSIVE_TOOLING, quoi qu'il arrive. Le transport, tel que mergé, n'était pas fonctionnel in-pod.

Pourquoi ni les tests ni le comité ne l'ont vu : les tests utilisaient tmp_path (un vrai FS local — round-trip OK en test, mais pas cross-pod) ; le diff écrivait le npz correctement ; le fait « /tmp n'est pas partagé » est un fait d'infra invisible dans le code. Classe S03 : le mock passe pendant que le chemin porteur diverge in-pod (§0bis).

La racine n'est pas « /tmp au lieu de S3 ». C'est qu'il n'existe aucun standard de transport : chaque DAG bricole son propre stockage intermédiaire, avec une cible configurable qui peut être fausse et que les tests unitaires ne voient pas. Cette épique supprime la cause, pas le symptôme.

§1. Problème & principe établi

Le problème. XCom est persisté dans la metadata-DB Airflow : conçu pour de petits messages (valeurs de retour, flags), borné en taille (~48 KB côté Postgres → truncate silencieux au-delà) et en types sérialisables. Y faire transiter des arrays/DataFrames est un anti-pattern documenté.

Le principe (pratique Airflow, non négociable). Petits messages → XCom ; données volumineuses → store externe, la tâche pousse la référence (clé/URI) en XCom, l'aval relit depuis le store. C'est le pass-by-référence. s43 respectait déjà ce principe (XCom = clés) — le défaut était la cible du store et le fait que le transport vivait en code de tâche plutôt qu'en couche-policy.

L'anti-pattern à éradiquer. Le « stockage intermédiaire bricolé par-tâche » : chaque tâche choisit où/comment sérialiser. C'est ce qui a permis la cible fausse de s43, et c'est invisible au test et au review.

§2. Le standard retenu

Object-storage XCom backend (Airflow common.io), pass-by-référence, piloté par seuil.

  • Config instance : XCOM_OBJECTSTORAGE_PATH (ex. s3://<conn>@<bucket>/xcom), XCOM_OBJECTSTORAGE_THRESHOLD (octets), XCOM_OBJECTSTORAGE_COMPRESSION (optionnel).
  • Sérialisation — correction (vérifié in-pod, common-io 1.4.2, CVN-N014-ED-S01 v4) : le backend ne transporte pas nativement numpy/pandas. serialize_value fait json.dumps(value, cls=XComEncoder) avant le test de seuil ; XComEncoder couvre le JSON-sérialisable (dict/list/str/num/bool, datetime, etc.) mais lève TypeError sur un numpy.ndarray (testé sur ndarray nu, unicode, et la forme dict-d'arrays de s43) → l'échec précède l'offload, à toute taille. Donc : JSON-sérialisable → offload > seuil vers S3 / inline ≤ seuil en metadata-DB ; numpy/pandas exigent une stratégie dédiée — serializer enregistré (airflow.serialization serde) ou pass-by-référence explicite (XCom = pointeur ; défaut retenu pour s43/ML, cf. S02 route Y). « object-storage backend » ≠ « gros objet arbitraire dans S3 » : la valeur doit d'abord être sérialisable par XComEncoder.
  • Le transport devient une policy centralisée : la tâche fait return value / pull, le backend gère l'offload + ne garde que l'URI. Plus de code de stockage par-tâche → la classe de bug « cible fausse » devient structurellement impossible (pour les payloads JSON-sérialisables ; numpy/pandas via S02, cf. ci-dessus).

Pourquoi object-storage et non un FS partagé RWX. S3 est déjà le substrat du projet (MLFLOW_ARTIFACT_ROOT, S3_BUCKET_NAME) ; le choix est cohérent avec une stack à pods stateless. Le FS partagé RWX (Scaleway File Storage CSI le permet, mais hors-défaut : tag cluster scw-filestorage-csi + PVC sfs-standard, le Block Storage sbs par défaut étant RWO-seul) donne la sémantique POSIX « drop-in » au prix d'un FS mutable partagé (concurrence, SPOF, GC manuel, ops plus lourde) — à contre-courant du reste de l'archi. Retenu uniquement si un vrai besoin POSIX émerge ; ce n'est pas le cas ici.

Le seuil = le mécanisme de non-disruption (cf. §4) : réglé pour que le contrôle-flow actuel (< seuil) reste en DB octet pour octet comme avant, et que seuls les payloads-data (> seuil, déjà cassés/truncés aujourd'hui) offload.

§3. Non-goals (hors-scope explicite)

  • Pas de changement d'orchestrateur, ni de framework de data-passing (Ray/Dask), ni de feature store.
  • Pas de FS partagé RWX (cf. §2), sauf besoin POSIX avéré ouvert en story séparée.
  • La correctness par-DAG (keying, complétude de cohorte, assertions métier — ex. K_eff de s43) reste la responsabilité de chaque DAG. L'épique mutualise la plomberie de transport, pas la logique de correction.
  • Pas de migration forcée du parc : aucun DAG existant n'est « basculé » de force (cf. §4).

§4. Modèle de migration (strangler-fig)

Point mécanique fondateur : le backend XCom est un réglage global de l'instance Airflow, pas un opt-in par-DAG. Flipper xcom_backend le flippe pour tous les DAGs d'un coup. Donc on ne « bascule » pas les DAGs un par un — c'est le seuil qui assure l'incrémentalité :

  • Anciens DAGs à petits XCom (la majorité) : < seuil → restent en metadata-DB, zéro changement, zéro action.
  • Nouveaux DAGs : bénéficient automatiquement dès le flip — aucune action requise, ils font juste return/pull.
  • Anciens DAGs qui bricolent déjà un contournement large-data (chemins S3 à la main, écriture FS, etc.) : simplifiés en return/pull quand on les touche/active (strangler-fig). Ce n'est pas « activer le backend par DAG » (le flip l'a fait pour tous), c'est supprimer le workaround.

Caveat dur, à ne pas oublier : le seuil protège l'emplacement de stockage, pas la sérialisation — le chemin serialize/deserialize est swappé pour tout le monde au flip. Un objet custom qui passait via le BaseXCom (pickling) peut casser. → l'audit de sérialisation sur tous les DAGs est un prérequis dur du flip (Story 1).

§5. Stories

Story 1 — Fondation : flip global du backend object-storage (load-bearing)

La story qui rend tout le reste possible. Ce n'est pas s43.

Tâches : 1. Audit de sérialisation (tous les DAGs) — recenser les types qui transitent en XCom (xcom_push, return de tâches @task, xcom_pull) ; isoler tout ce qui n'est pas JSON-sérialisable par XComEncoder (numpy/pandas inclus — cf. §2 : ils ne passent pas nativement ; objets custom, dataclasses non-supportées, types pickle-only). C'est l'audit qui mesure le blast-radius et qui décide le routage s43 (§7). (Recon S01 v4 : le parc actuel est JSON-only + sous seuil → flip near-inert ; le seul payload tableau, s43, écrit déjà hors-XCom vers un store.) 2. Choix de seuil — calibrer XCOM_OBJECTSTORAGE_THRESHOLD pour que le contrôle-flow courant reste en DB et que seuls les payloads offload (mesurer la distribution de tailles XCom actuelles). 3. TTL / lifecycle bucket — politique de rétention sur le préfixe XCom S3 (lifecycle-rule + purge/clear du backend) pour éviter la croissance non-bornée. 4. Rollout staged — valider sur Airflow de staging (ou un sous-ensemble), puis flip prod ; blast-radius = tous les DAGs, donc test propre, pas de flip à l'aveugle. 5. ADR (cf. §8) codifiant le standard.

Done-criteria : - audit publié (liste des types XCom par DAG ; aucun type non-sérialisable non-traité) ; - seuil + TTL + path configurés et documentés ; - flip prod effectué, aucun DAG existant régressé (vérif post-flip sur un échantillon représentatif) ; - ADR mergé.

Story 2 — s43 : premier bénéficiaire / validateur (routage X ou Y)

s43 est l'incident déclencheur et le premier consommateur de gros payloads cross-pod. Son travail concret reste tracké en CVN-N001-EI-S05 (cf. §7) ; cette story porte la décision de routage, conditionnée à l'audit de la Story 1 :

  • Route X — s43 = validateur du flip (si audit petit / flip livrable vite) : pas de fix interim ; s43 passe en return/pull direct sur le backend, et son smoke devient le test d'acceptation du backend. Zéro code jetable.
  • Route Y — s43 débloqué avant le flip (si audit gros / flip incertain) : fix A(ii) local en S05 (persist/load → cvntrade_s3_manager via BytesIO, + lifecycle-rule sur le préfixe s43) ; s43 livre indépendamment ; l'épique range s43 (suppression du persist/load bespoke) dans la Story 3 quand le flip atterrit.

Done-criteria : routage tranché sur evidence de l'audit ; s43 runnable in-pod (smoke vert) ; si route Y, dette A(ii) inscrite à la Story 3.

Story 3 — Track de cleanup opportuniste (rolling)

Simplifier les DAGs à workaround large-data en return/pull, au fil des activations/retouches — jamais en big-bang.

  • Inventaire : rempli par l'audit de la Story 1 (DAGs qui écrivent en S3/FS à la main, ou qui frôlent la limite XCom).
  • Inclut le cleanup de s43 si route Y.
  • Story volontairement non-terminante au sens calendaire : elle se ferme quand l'inventaire est vide, pas à une date.

Done-criteria : inventaire publié ; chaque DAG touché post-flip est soit déjà conforme, soit simplifié dans le même PR ; aucun nouveau workaround introduit après le flip (gate de review).

Story 4 — Standard & docs

  • ADR (Story 1) référencé partout.
  • Guideline auteur-de-DAG : « comment passer des données entre tâches » (petit → XCom natif ; volumineux → return/pull, le backend gère ; jamais de stockage bricolé par-tâche).
  • Item de checklist de review (CodeRabbit / comité) : tout nouveau DAG qui écrit un store intermédiaire à la main est rejeté → return/pull.

Done-criteria : guideline live ; item de review en place.

§6. Risques & mitigations

  • Blast-radius du flip (config globale, tous DAGs d'un coup) → rollout staged (Story 1.4) + vérif post-flip.
  • Audit incomplet (un type custom raté → casse silencieuse au flip) → l'audit est le prérequis dur ; en cas de doute, garder le DAG concerné en sérialiseur custom avant flip.
  • Seuil mal réglé : trop bas → latence/coût S3 sur du contrôle-flow ; trop haut → s43 reste cassé → calibrage sur la distribution mesurée (Story 1.2).
  • Accumulation S3 → TTL/lifecycle (Story 1.3).
  • Couplage plateforme ↔ story : ne pas faire rentrer le flip global sur le dos du déblocage s43 → c'est précisément pourquoi Story 1 ≠ s43, et pourquoi la route Y existe.

§7. Relation à CVN-N001-EI-S05 / s43

s43 est le déclencheur et le premier validateur, mais son fix concret n'appartient pas à cette épique :

  • le transport-fix de s43 (A(ii)) ou son passage return/pull (X) est tracké en S05 ;
  • cette épique fournit le standard dont s43 bénéficie ;
  • l'audit de la Story 1 décide le routage (X vs Y) de la Story 2 ;
  • si route Y, le cleanup de s43 est inscrit à la Story 3.

Séparation délibérée : la science de s43 (keying run_id, complétude K_eff, cohorte) reste en S05 ; seule la plomberie de transport remonte ici.

§8. Livrables

  • ADR (numéro à attribuer) — « Données inter-tâches volumineuses → object-storage XCom backend ; policy de seuil ; aucun stockage bespoke par-tâche » (invariant + rationale + alternative FS-RWX rejetée et pourquoi).
  • Backend configuré en prod (path + seuil + compression + TTL).
  • Document d'audit de sérialisation (types XCom par DAG).
  • Inventaire de cleanup (Story 3).
  • Guideline auteur-de-DAG + item de review.

Références

  • Pratique Airflow : XCom pour petits messages, store distant (S3/HDFS) pour les données volumineuses, passer la référence (Airflow Best Practices ; Astronomer Pass data between tasks / Custom XCom backend strategies).
  • Object Storage XCom backend (apache-airflow-providers-common-io) : XCOM_OBJECTSTORAGE_PATH / _THRESHOLD / _COMPRESSION.
  • Scaleway File Storage CSI (RWX, hors-défaut) vs Block Storage sbs (RWO) — alternative FS-partagé, rejetée en §2.
  • ADR-0098 (mapper le régime de déploiement), ADR-0099 (existence≠sélection) — discipline §0bis dont relève l'incident §0.
  • Incident déclencheur : CVN-N001-EI-S05 / s43, transport /tmp pod-local non-fonctionnel in-pod.

| CVN-N014-ED-S01 | Fondation — flip global du backend object-storage XCom (load-bearing) | #1105 · wp#? | New |

| CVN-N014-ED-S02 | s43 — premier bénéficiaire / validateur (routage X ou Y) | #1106 · wp#? | New |

| CVN-N014-ED-S03 | Track de cleanup opportuniste (rolling) | #1107 · wp#? | New |

| CVN-N014-ED-S04 | Standard & docs (guideline + review gate) | #1108 · wp#? | New |