Pipeline: orders-enriched-daily

• Scopo: pipeline di riferimento per il DSL a grafo (graph()): arricchisce gli ordini con l'anagrafica clienti via join a chiave e distribuisce il risultato (fan-out) su due destinazioni indipendenti. Non movimenta dati reali — dataset dimostrativo in-memory — pensata come esempio eseguibile e banco di prova per le funzionalità DAG (join, fan-out, dead-letter per nodo, nodo try/catch, path di fallback su sink isolabile, corpo dei record falliti visibile nel report).

Nota: il guard di questa pipeline attiva includeRawInReport, quindi il corpo degli ordini respinti finisce nel report ed è visibile nel pannello del guard in UI. È accettabile qui perché la demo gira su fixture sintetiche. Su una pipeline con dati reali significherebbe persisterli nel report: ci si pensa due volte prima di copiare quell'opzione altrove.

• Owner: — (esempio di piattaforma)
• Cadenza: on-demand (demo)
• Runbook: non applicabile (nessun sistema esterno).

Flusso dati

orders-enriched-daily

Istanze di sub-pipeline, chiuse: entra col doppio click o da qui.scheda di riepilogo-ordini

0/7

Scegli una scena e premi ▶ per guardarla scorrere, oppure → per andare avanti un passo alla volta.

Lo scopo dei nodi

Nodo	Tipo	Scopo
orders	source	Legge gli ordini di prova della dimostrazione, ognuno con il cliente che l'ha fatto, l'importo e la data.
customers	source	Legge l'anagrafica di prova dei clienti, con il nome e la fascia di ciascuno, che servono a completare gli ordini.
validate-orders	transform	Controlla che ogni ordine abbia cliente, importo e data validi. Un ordine fuori regola viene messo da parte fra gli scarti, e gli altri proseguono.
validate-customers	transform	Controlla che ogni cliente abbia un nome e una fascia ammessa. Un cliente fuori regola ferma la run: gli ordini completati con un'anagrafica sbagliata sarebbero sbagliati anche loro.
enrich	join	Completa ogni ordine con il nome e la fascia del suo cliente. Un ordine il cui cliente non è in anagrafica prosegue lo stesso, senza nome né fascia.
guard	try-catch	Protegge, ordine per ordine, il riepilogo e il salvataggio: un ordine rifiutato più avanti, come quello con un importo non positivo, finisce fra gli errori e la run prosegue con gli altri.
riepilogo-ordini	pipeline	Prepara i due risultati della giornata a partire dagli ordini completati: gli ordini controllati, pronti da salvare, e il riepilogo per fascia cliente.
enriched_orders	sink	Salva gli ordini della giornata completati con i dati del cliente, uno per ordine: è il dettaglio su cui si appoggia ogni analisi.
tier_summary	sink	Salva il riepilogo della giornata per fascia cliente, con quanti ordini e quale importo totale per ciascuna.
log-errors	transform	Registra nel log ogni ordine finito fra gli errori, con il passo in cui l'errore è nato e il motivo, così si capisce che cosa è andato storto.
caught_errors	sink	Conserva gli errori registrati, ognuno con l'ordine che l'ha causato, per poterli esaminare dopo la run.
recover-orders	transform	Fra gli errori, tiene solo gli ordini che non si sono potuti salvare: sono gli unici da recuperare. Gli altri errori restano solo nel log.
restore-order	transform	Riprende l'ordine com'era prima del salvataggio non riuscito, per poterlo salvare altrove.
orders-fallback	sink	Salva in un archivio di riserva gli ordini che non si sono potuti salvare al loro posto, così nessun ordine va perso.
orders_dead_letter	dead-letter	Raccoglie i record scartati dai nodi di questa pipeline, ognuno con il motivo dello scarto.

Il grafo qui sopra è disegnato dalla topologia reale della pipeline (getSpec()): nodi, archi, dead-letter e ramo catch sono quelli eseguiti. Premi ▶ per far passare un record, o → per avanzare un passo alla volta.

Due sorgenti indipendenti confluiscono in un join a chiave (enrich), il cui output entra nell'istanza riepilogo-ordini della sub-pipeline omonima. Il suo confine d'ingresso valida l'ordine arricchito e fa fan-out sulle due uscite dell'istanza, due rami che non si influenzano a vicenda, salvo un errore fatale, che ferma la run: il fatto arricchito (riepilogo-ordini → riepilogo-ordini/fatti → enriched_orders) e un riepilogo aggregato per tier cliente (riepilogo-ordini → riepilogo-ordini/tier-summary → riepilogo-ordini/per-tier → tier_summary).

Il riepilogo è una sub-pipeline istanziata col nodo pipeline: alla costruzione si appiattisce nei suoi nodi, con gli id qualificati. Il grafo qui sopra mostra l'istanza come un nodo chiuso, in cui si entra col doppio click o da «Entra» (il link ?istanza=riepilogo-ordini apre la scheda già dentro); il report mostra i nodi interni con l'id qualificato. La root cabla le due uscite con fromPort: "fatti" e fromPort: "per-tier".

Codice: src/pipelines/orders-enriched-daily/ · DSL a grafo: src/core/graph-builder.ts.

Sorgenti

Sistema	Schema/tabella	Chiave naturale	Modalità
In-memory (demo)	ordini grezzi (orders)	order_id	full (dataset fisso in run.ts)
In-memory (demo)	anagrafica clienti (customers)	customer_id	full (dataset fisso in run.ts)

Sink

Sistema	Destinazione	Idempotenza
In-memory (demo)	enriched_orders	n/a (demo — in un caso reale: upsert su order_id)
In-memory (demo)	tier_summary	n/a (demo — in un caso reale: upsert su tier)
In-memory (demo)	caught_errors	n/a (demo) — ramo catch del guard (log-errors): gli errori catturati, col loro raw
In-memory (demo)	orders-fallback	n/a (demo) — ramo catch: gli ordini che enriched_orders ha rifiutato (recover-orders → restore-order)
In-memory (demo)	orders_dead_letter	n/a (demo) — dead-letter di validate-orders, opzionale

Trasformazioni e regole di business

Lo scopo di ogni nodo è nella tabella sotto il grafo, presa dal piano. Qui le regole che i nodi applicano.

• validate-orders (contratto raw-order-contract) — valida l'ordine grezzo (rawOrderSchema): id non vuoti, amount numerico, created_at coercibile a data. Isolato su dead-letter per nodo (ordersDeadLetter, opzionale).

• validate-customers (contratto raw-customer-contract) — valida il cliente grezzo (rawCustomerSchema): id/nome non vuoti, tier in gold|silver|bronze.

• enrich — hashJoin a chiave (build = clienti in memoria, probe = ordini in streaming, pompato in concorrenza mentre la tabella di build si materializza), how: "left": un ordine senza cliente noto sopravvive con customer_name/tier a null. Non solleva errori (il join si limita ad arricchire); entrambe le sorgenti validate vi confluiscono simmetricamente.

• guard — nodo try/catch subito dopo enrich: nel suo scope stanno l'istanza riepilogo-ordini e i sink a valle.

• riepilogo-ordini — istanza della sub-pipeline riepilogo-ordini, alimentata dal guard. Usa il parametro tierMancante al default, "unknown", quindi si istanzia senza parametri; il piano lo mostra risolto in instances[].params. Appiattita, diventa quattro nodi:

  • riepilogo-ordini, il confine d'ingresso, che valida l'ordine arricchito (enrichedOrderSchema, amount positivo) e fa da gate/fan-out verso le due uscite. Era il nodo fact-contract della root, che l'istanza ha assorbito: il contratto del fatto non si ripete più a monte dell'istanza. riepilogo-ordini:extracted conta gli ordini entrati nell'istanza. Un ordine con amount <= 0 passa il join ma viene qui respinto (ValidationError); essendo a valle del guard, l'errore è catturato dal guard e instradato sul ramo catch (log-errors → caught_errors) invece di abortire. Il guard è per-record: tutti gli ordini con amount <= 0 finiscono nel catch (col loro raw) e i validi continuano a essere caricati;
  • riepilogo-ordini/fatti, il confine dell'uscita fatti, alimentato direttamente dall'ingresso, verso enriched_orders;
  • riepilogo-ordini/tier-summary, groupBy stateful per tier cliente (order_count, total_amount, gli ordini senza cliente noto sotto il valore di tierMancante, unknown); come ogni aggregazione, è fail-fast (nessun dead-letter per-record: l'aggregazione bufferizza l'intero stream). Essendo stateful non viene mai isolato per-record, quindi un suo errore arriva al catch con raw: undefined; porta però lastConsumed, l'ultimo ordine consumato dal nodo — esatto se l'errore emerge durante il consumo dello stream, assente se emerge in fase di emissione. L'aggregazione di questa demo non ha punti in cui sollevare, quindi il caso è coperto dai test del core e non dalla demo;
  • riepilogo-ordini/per-tier, il confine dell'uscita per-tier, che rivalida il riepilogo (tierSummarySchema) prima del sink. Era il nodo tier-summary-contract.

  Il guard sta a monte dell'istanza, quindi la copre per intero: un errore nato in uno di questi quattro nodi arriva al suo ramo catch col proprio id qualificato in error.context.nodeId.

  Il caso dell'errore in emissione (e quello di un nodo a valle dell'aggregazione, dove raw è l'aggregato e non gli ordini) è coperto dalla lineage dei record: attivando recordIds: true sul grafo e guardSnapshots: true sul guard, il catch riceve guardSnapshots con gli ordini originali del gruppo colpevole. Questa pipeline non li attiva, per lo stesso motivo per cui il caso non è dimostrato qui: manca un punto di fallimento realistico nell'aggregazione, e gli snapshot terrebbero payload in memoria senza che serva a nulla.

• recover-orders / restore-order — path di fallback sul ramo catch. Il sink enriched_orders sta dentro lo scope del guard: se implementa loadOne è isolabile per-record, quindi un fallimento di scrittura non aborta la run ma finisce sul ramo catch portando con sé l'ordine (DeadLetter.raw). recover-orders filtra i soli errori provenienti da quel sink, restore-order ne estrae il payload e lo carica sul sink orders-fallback. Gli altri errori catturati (contratto, aggregazione, sink opachi) non hanno un ordine da recuperare e sono filtrati via.

Contratti (schema Zod)

Definiti in schema.ts:

• Input: rawOrderSchema, rawCustomerSchema.
• Intermedio/output: enrichedOrderSchema (fatto arricchito), tierSummarySchema (riepilogo per tier). Sono i contratti d'ingresso e d'uscita della sub-pipeline riepilogo-ordini: si definiscono in src/sub-pipelines/riepilogo-ordini/schema.ts e schema.ts li riesporta.

Colonne nel piano (getSpec): da questi contratti l'introspezione Zod ricava le columns dei nodi mostrate nella UI del DAG. I nodi validate() (validate-orders, validate-customers) e i confini dell'istanza (riepilogo-ordini, riepilogo-ordini/fatti, riepilogo-ordini/per-tier) espongono le colonne del proprio schema; i sink (enriched_orders, tier_summary) le ereditano dal contratto a monte per propagazione. Il join enrich (predecessori discordi) e le source in-memory restano senza colonne. Il groupBy riepilogo-ordini/tier-summary dichiara tierSummarySchema come outputSchema, quindi mostra l'aggregato e non gli ordini che riceve. Il ramo catch non eredita (porta DeadLetter): log-errors, caught_errors e recover-orders restano senza colonne, finché restore-order riporta il record a un ordine e lo dichiara con enrichedOrderSchema. Vedi columnsFromZodObject e toPlanSpec in src/core/.

Parametri ed env

• ctx.params: nessuno (dataset demo fisso in run.ts).
• Variabili d'ambiente: nessuna — pipeline interamente in-memory, non richiede .env.example.

  Nota storica. In passato questa pipeline aveva un'opzione stepDelayMs che inseriva un ritardo artificiale nel writeReport di ogni nodo, per rendere osservabile l'avanzamento in UI. Era una pezza a un buco nell'osservabilità: il runner notificava solo onNodeComplete, quindi la UI non aveva modo di sapere quali nodi stessero girando e senza ritardo la run finiva entro un singolo intervallo di polling. Con onNodeStart il segnale c'è davvero, e il ritardo — che stava nel percorso dati e bloccava anche il downstream — è stato rimosso.

Gestione errori

• Dead-letter per nodo (scelta di design del DSL a grafo): validate-orders ha il proprio sink di dead-letter (ordersDeadLetter, opzionale) che isola gli ordini grezzi non conformi record-by-record, continuando con i record buoni.
• Try/catch (per-record): il nodo guard (dopo enrich) protegge l'istanza riepilogo-ordini e i nodi a valle. Ogni record che solleva un EtlError in un nodo protetto a valle del guard è deviato sul ramo catch (errorLog, opzionale; default: sink interno in memoria) come envelope DeadLetter { raw, error }, dopo essere stato loggato da log-errors, e lo stream continua (N record → N nel catch). La run non abortisce; lo stato diventa partial. Il comportamento "fermati al primo errore" è opt-in via .tryCatch(..., { mode: "circuit-breaker" }). Le transform stateful (es. riepilogo-ordini/tier-summary, un group-by) restano streaming, mai isolate.
• Errori a monte del guard: il guard cattura solo gli errori nati a valle di sé. Un cliente non conforme in validate-customers (senza dead-letter) o un errore del joinenrich fanno abortire la run, anche se l'errore attraversa l'istanza riepilogo-ordini.
• riepilogo-ordini/tier-summary resta fail-fast: essendo stateful, un errore qui aborta l'intera run (come ogni aggregazione).
• Il RunReport (GraphRunReport) espone status: "partial" quando almeno un nodo ha scarti (rejected) o il try/catch ha catturato errori (caught), con nodes[].rejected/rejectedReasons e nodes[].caught per nodo — mai il payload del record (quello vive solo nel dead-letter/errorLog sink).
• Visibile nella DAG: getSpec() collega sia il dead-letter sia il ramo catch nella topologia statica, così la UI mostra i branch d'errore — da validate-orders verso orders_dead_letter (reject di schema, stepType: "dead-letter") e dal nodo guard (stepType: "try-catch") verso log-errors (kind: "catch") — con edge tratteggiato destructive.

Test

• Unit pipeline: pipeline.test.ts — join left, fan-out su entrambi i sink, dead-letter per-record su validate-orders, cattura via try/catch dell'importo non positivo respinto dal confine d'ingresso riepilogo-ordini, instradato sul ramo catch, forma della topologia reale (nodes/edges, incl. guard, ramo catch e nodi dell'istanza riepilogo-ordini) nel report, riepilogo il cui totale trabocca rifiutato da riepilogo-ordini/per-tier e catturato dal guard.
• Sub-pipeline: sub-pipeline.test.ts prova riepilogo-ordini da sola con runSubPipeline.
• Esecuzione manuale: pnpm run etl run orders-enriched-daily stampa il GraphRunReport completo su stdout (vedi docs/conventions/run-report.md).

Versioni

La storia della pipeline, dalla versione più recente: la voce di changelog di ogni versione (src/pipelines/orders-enriched-daily/changelog.ts) e, sotto, cosa è cambiato nel piano rispetto alla precedente. Vedi versioni di pipeline.

v32026-10-09 · ETL-193, ETL-195

Modifiche

• Nessun cambio di comportamento: il piano è identico alla v2. Gli import seguono l'infrastruttura (changelog, configurazione delle istanze, report di volume) uscita da `src/pipelines/` (ADR-0093 §1), e la pipeline si dichiara nel catalogo di `etl.config.ts` invece che in un registro (ADR-0093 §2)

Il piano non cambia rispetto alla v2.

v22026-10-05 · ETL-85, ETL-87, ETL-88

Modifiche

• Contratto del fatto e riepilogo per tier passano nella sub-pipeline `riepilogo-ordini`, istanziata da un nodo `pipeline` (ADR-0079): `fact-contract`, `tier-summary` e `tier-summary-contract` lasciano il posto ai nodi dell'istanza, e i due sink leggono dalle sue uscite `fatti` e `per-tier`
• Un ordine con importo non positivo è rifiutato al confine d'ingresso dell'istanza invece che da `fact-contract`; il `guard` lo cattura e lo instrada sul ramo `catch` come prima

Il piano rispetto alla v1:

Nodi

• +riepilogo-ordinitransform
• +riepilogo-ordini/tier-summarytransform
• +riepilogo-ordini/per-tiertransform
• +riepilogo-ordini/fattitransform
• −fact-contracttransform
• −tier-summarytransform
• −tier-summary-contracttransform

Archi

• +guard → riepilogo-ordiniporta out
• +riepilogo-ordini/fatti → enriched_orders
• +riepilogo-ordini/per-tier → tier_summary
• +riepilogo-ordini → riepilogo-ordini/tier-summary
• +riepilogo-ordini/tier-summary → riepilogo-ordini/per-tier
• +riepilogo-ordini → riepilogo-ordini/fatti
• −guard → fact-contractporta out
• −fact-contract → enriched_orders
• −fact-contract → tier-summary
• −tier-summary → tier-summary-contract
• −tier-summary-contract → tier_summary

v12026-10-05 · ETL-120

Aggiunte

• Versione iniziale sotto versionamento (ADR-0086)

Prima versione: non c'è una versione precedente da confrontare.