GraphRunReport — il risultato di una esecuzione

Il servizio è stateless e agnostico: esegue una pipeline e restituisce un GraphRunReport con tutti i dettagli del caso. Non possiede DB, non scrive da nessuna parte, non fa push. Cosa farne (archiviare, alert, ignorare) è deciso dal chiamante.

Contratto di I/O (CLI / container)

• stdout = un solo oggetto JSON: il report. Parsabile con jq.
• stderr = log umani / progress (riga pipeline.completed, ecc.).
• exit code: 0 se status è success/partial; ≠0 se failed.
• Anche in caso di errore viene emesso un report (status: "failed").

etl run sessions-daily > report.json 2> run.log   # separa report e log
etl run sessions-daily | jq '.status, .written'   # consuma al volo

Come libreria, graph(...).build().run(ctx) restituisce direttamente il report e lancia sull'errore fatale — contratto invariato. L'errore lanciato porta con sé un report completo (topologia reale e conteggi parziali), recuperabile con graphRunReportOf(err):

try {
  report = await pipeline.run(ctx);
} catch (err) {
  // L'errore mantiene il suo tipo concreto (ValidationError, ExtractError, …).
  report =
    graphRunReportOf(err) ??
    buildFailedGraphReport(ctx, name, err, startedAtMs);
}

buildFailedGraphReport resta il fallback per i fallimenti sollevati prima che il grafo esista (es. wiring dei connettori in run.ts), dove una topologia da riportare non c'è davvero.

Schema (reportSchemaVersion: 8)

Campo	Tipo	Note
reportSchemaVersion	number	versione dello schema del report
runId	string	id della run. Per un processo ritentato dall'API si ripete: a distinguere le esecuzioni è l'executionId del contesto, che sta nei log e non nel report
pipeline	string	nome pipeline
status	success | partial | failed	partial = girata con scarti
ok	boolean	true se non abortita
env	string	ambiente logico (ETL_ENV)
release	string | null	release del deploy che ha eseguito: ETL_RELEASE, altrimenti git:<sha> del repo dell'etl.config.ts. Fino allo schema v6 si chiamava version — vedi § Versione di pipeline e release
core	string | null	versione del pacchetto del core installato, letta a runtime. null nei report prima del v8 e per un grafo lanciato fuori dalla CLI — vedi § Versione di pipeline e release
client	string | null	cliente del deploy, il client di defineEtlConfig. null nei report prima del v8
pipelineVersion	number | null	versione di pipeline che ha eseguito, dal changelog del catalogo. null per un grafo che non viene dal catalogo — vedi § Versione di pipeline e release
startedAt / finishedAt	string (ISO 8601)
durationMs	number
nodes	NodeReport[]	topologia reale del grafo eseguito, task compresi — vedi § NodeReport (mai credenziali nei nomi)
edges	{ from: string; to: string }[]	archi di dati del grafo eseguito. Gli edge di controllo non ci sono: stanno nel piano (getSpec()), con kind: "control"
params	object	parametri della run (es. watermark). No segreti
extracted	number	record letti dal mondo esterno: somma dei soli nodi source (niente doppio conteggio su un fan-out)
written	number	record scritti verso l'esterno: somma dei nodi sink. Su un fan-out lo stesso record scritto su due sink conta due volte
rejected	number	record isolati a dead-letter, su tutti i nodi
rejectedReasons	object	categorie di scarto → conteggio, aggregate su tutti i nodi
metrics	object	contatori (standard + eventuali custom degli stage)
nodeReports	Record<string, unknown>	mappa nodeId → esito di writeReport del nodo, o di quello dichiarato dalla sua transform, o il default { name, success }
state	Record<string, SourceStateReport>	stato di pipeline, una voce per source incrementale (per id di nodo): da dove ha letto, cosa propone, se ha committato. {} se la pipeline non ne ha — vedi § Stato di pipeline
instances	RunInstance[]	le istanze di sub-pipeline del grafo eseguito, nella forma del piano, col riepilogo d'istanza di ognuna. [] senza sub-pipeline; assente in un report precedente a v7 — vedi § Istanze
error	object | null	{ type, message, stage, nodeId } se failed (no PII). nodeId è il nodo che ha fallito, non l'ultimo attraversato dall'errore mentre risaliva lo stream

NodeReport — cosa c'è per ogni nodo

Campo	Tipo	Note
id / kind / name	string	identità del nodo nel grafo eseguito
extracted / written	number	record entrati / usciti dal nodo. La differenza è quanto il nodo ha scartato senza errore — per un dedup con duplicates: "drop" è il numero di duplicati eliminati, e non esiste un campo dedicato. Un'espansione (flatMapTransform, explode) può invece farla crescere: written > extracted
rejected / rejectedReasons	number / object	record isolati a dead-letter e loro categorie
caught / caughtErrors	number / CaughtErrorSummary[]	solo sui nodi try-catch
ports	Record<string, number>	solo sui nodi route: record usciti per porta, default e zeri compresi
retries	number	solo sui nodi con una policy di retry: tentativi ripetuti nella run, zero compreso
versions	{ opened, closed, corrected, unchanged }	solo sui sink in modalità SCD2: versioni aperte, chiuse, corrette in giornata e invariate, sommate fra i blocchi come written. Su quei sink written = opened + corrected: una run con tutto invariato ha written: 0, e unchanged lo spiega. Assente in un rehash
skipped	{ because: string }	solo su un nodo saltato: non è mai partito, e because è l'id del nodo che l'ha fermato — vedi § «Nodi saltati»
startedAt	string | null	primo record attraversato (ISO 8601); per un task, l'ingresso in run(). null = il nodo non ha mai lavorato
finishedAt	string | null	completamento del nodo (ISO 8601). null = il nodo non si è mai concluso (run abortita)
durationMs	number	tempo di attraversamento — risponde a «qual è il collo di bottiglia»

I tre campi di tempo sono l'aggiunta dello schema v4 (additivi: nessun campo rimosso o rinominato, i consumer v3 continuano a funzionare).

versions è additivo e opzionale, e non ha cambiato la versione dello schema: compare solo sui sink che lo restituiscono.

skipped è l'aggiunta dello schema v5, anch'essa additiva. Distingue un nodo saltato da uno con input vuoto.

Nodi saltati

Un nodo saltato non è mai partito: ha skipped: { because }, success: false in nodeReports e startedAt: null. Un sink saltato non chiama load, così un sink che riscrive un file intero non lo riscrive vuoto. Succede in tre casi:

Chi è saltato	because
Un task o una source il cui predecessore di controllo non è riuscito	quel predecessore
I discendenti di dati di una source saltata, e i nodi a monte che alimentano solo nodi saltati	lo stesso della source
Un sink senza record in ingresso, con a monte un nodo troncato da un guard prima di emetterne	il nodo troncato

Un sink non è riuscito, e quindi i suoi successori di controllo sono saltati, se è fallito (anche con l'errore catturato da un guard), se è stato saltato, o se il suo input è stato troncato: un guard ha catturato l'errore di un nodo a monte. In quest'ultimo caso il sink ha scritto e nel report ha success: true: a dire che non è riuscito è lo skipped del suo successore.

La run è partial quando il salto viene da un errore catturato, failed quando viene da un errore fatale. In una run failed hanno skipped solo i nodi fermi a un gate e ciò che ne discende: un nodo interrotto dall'annullamento era partito, e resta con finishedAt: null o con success: false, senza skipped.

Durante la run. skipped non aspetta il report finale: viaggia anche sull'evento di nodo concluso, così chi segue la run dal vivo non vede un nodo saltato in errore. NodeProgressUpdate.skipped porta lo stesso valore di NodeReport.skipped, e arriva per gli stessi nodi:

Canale	Dove
ProgressReporter.onNodeComplete	update.skipped?: { because }
Stream SSE GET /runs/:id/events	node-completed con skipped?: { because }
Polling su processes (noeva-server-api)	output_data.skippedNodes: { [nodeId]: { because } }

Il campo c'è solo sui nodi saltati: un nodo concluso normalmente, o interrotto dall'annullamento, non ha la chiave. success: false con startedAt: null non basta a dedurlo, perché vale anche per un nodo che l'annullamento ha fermato prima di partire senza che un gate lo trattenesse. Un nodo fermo a un gate quando la run viene annullata è invece saltato, col predecessore ancora in corso come because, e lo è sia sull'evento sia nel report. È opzionale e additivo: reportSchemaVersion non cambia.

La versione è rimasta 5 con i task e gli edge di controllo: non c'è un campo nuovo né un campo che cambia significato. kind ammette "task", come ha ammesso union, ai e action senza cambiare versione, e edges resta di soli edge di dati, così un consumer della v5 non scambia un edge di controllo per un flusso di record.

Stato di pipeline (state)

state è l'aggiunta dello schema v6: una voce per ogni source incrementale, cioè una source che dichiara watermark. È un oggetto vuoto per una pipeline che non ne ha, e per una run fallita prima di leggere lo stato.

Campo	Tipo	Note
read	string	il watermark da cui la source ha letto, prima del lookback
origin	stored | override | initial	dallo stato salvato, da params.watermark.<nodeId>, o dal valore initial della source
proposed	string | null	il massimo visto sullo stream estratto. null se la source non ha visto righe
committed	boolean	se il watermark è stato committato
reason	string	solo con committed: false: perché no

I valori sono testo, nella precisione nativa del database: un timestamp Postgres conserva i microsecondi. Sono valori di watermark, non payload: il report resta sicuro da archiviare.

Il commit avviene a run conclusa, dopo i sink e i post-task. committed: true dice che lo store ha accettato la voce, non che il valore sia avanzato: il commit è monotono, e un backfill sotto il valore salvato lo lascia dov'era.

reason	Quando
skipped	la source non è mai partita: il suo gate non si è aperto. Prevale sulle altre
failed	la run è failed
partial	la run è partial e il grafo non dichiara commitOn: "partial"
records_lost	la run è partial con commitOn: "partial", ma non ogni scarto è una dead-letter per record: un guard ha chiuso uno stream, o un nodo è stato saltato
no_records	la source non ha visto righe: niente da proporre
generation_mismatch	un reset dello stato durante la run
type_mismatch	lo store ha un valore di un altro tipo
store_error	il commit non è riuscito: un errore permanente, o uno transitorio dopo i retry. La run resta success: i dati sono scritti

Una run success con committed: false è quindi possibile (no_records, generation_mismatch, type_mismatch, store_error): lo stato di pipeline non cambia l'esito della run. Cosa fare per ogni reason: runbook.

"state": {
  "ordini": {
    "read": "2026-09-28 10:00:00.123456",
    "origin": "stored",
    "proposed": "2026-09-29 06:00:00.000001",
    "committed": true
  },
  "fatture": {
    "read": "0",
    "origin": "initial",
    "proposed": "10",
    "committed": false,
    "reason": "partial"
  }
}

Istanze di sub-pipeline (instances)

instances è l'aggiunta dello schema v7. Il report resta piatto: nodes, nodeReports e metrics sono indicizzati per id qualificato, e i nodi di confine compaiono come normali transform. Accanto, instances porta una copia dell'albero delle istanze nella forma del piano (id, parent per un'istanza annidata, localId, subPipeline, params risolti, deps come { slot, kind, descriptor? }, outputs), più il riepilogo d'istanza in summary. Lo purpose dell'istanza no: lo scopo spiega il grafo, non una sua esecuzione, e sta solo nel piano. La copia rende leggibile per istanza anche una run vecchia, quando il piano corrente non descrive più un'istanza rinominata.

C'è sempre, [] per una pipeline senza sub-pipeline e nel report di buildFailedGraphReport. Un report archiviato prima di v7 non ha il campo e si legge come «nessuna istanza»: nessuna migrazione.

Campo di summary	Tipo	Note
extracted	number	record entrati: extracted del nodo d'ingresso, quello con l'id dell'istanza
written	Record<string, number>	record usciti per uscita: written del nodo d'uscita, una voce per uscita, zeri compresi
rejected	number	rifiutati da tutti i nodi dell'istanza, di confine e delle istanze annidate
rejectedReasons	object	categorie di scarto → conteggio, sugli stessi nodi di rejected
retries	number	tentativi ripetuti dai nodi dell'istanza e delle istanze annidate, 0 senza retry
startedAt	string | null	il primo startedAt dei nodi; null se nessuno ha visto un record
finishedAt	string | null	l'ultimo finishedAt dei nodi; null se nessuno si è concluso. In una run abortita, la fine dell'ultimo nodo concluso
durationMs	number	intervallo d'orologio fra i due, mai somma di durate (in streaming i nodi lavorano insieme); 0 se manca un estremo

Il riepilogo non ha uno stato: si ricava dai nodi dell'istanza, con la stessa regola in live e a fine run. Non sta in metrics: il nodo d'ingresso ha lo stesso id dell'istanza, e una chiave <istanza>:rejected coprirebbe quella del nodo. Lo calcola una funzione pura, summarizeInstances(nodes, instances) in src/core/instance-summary.ts, che GraphPipeline.run() applica al report finale e a quello allegato all'errore: il runner non sa delle istanze. Progress, eventi SSE e worker non cambiano; il riepilogo arriva solo col report finale, e durante la run l'albero si legge dal piano.

"instances": [
  {
    "id": "riepilogo-ordini",
    "localId": "riepilogo-ordini",
    "subPipeline": "riepilogo-ordini",
    "params": { "tierMancante": "unknown" },
    "deps": [],
    "outputs": [
      { "name": "per-tier", "nodeId": "riepilogo-ordini/per-tier" },
      { "name": "fatti", "nodeId": "riepilogo-ordini/fatti" }
    ],
    "summary": {
      "extracted": 120,
      "written": { "per-tier": 4, "fatti": 119 },
      "rejected": 1,
      "rejectedReasons": { "validation": 1 },
      "retries": 0,
      "startedAt": "2026-10-05T07:00:00.120Z",
      "finishedAt": "2026-10-05T07:00:01.480Z",
      "durationMs": 1360
    }
  }
]

etl runs show mostra la sezione «Istanze» dopo la tabella dei nodi: una riga per istanza, indentata per profondità, con entrati, usciti per uscita, rifiutati e durata («mai avviata» o «interrotta» quando l'istanza è partita ma nessun suo nodo si è concluso). Senza istanze la sezione non compare.

Task

Un task è un nodo del report con kind: "task". Non vede record: extracted, written e rejected restano a 0, e non entra nei totali della run. startedAt è l'ingresso in run(), finishedAt l'uscita. I dettagli, per esempio le righe toccate da un task SQL, stanno in nodeReports. retries compare con una policy di retry.

startedAt: null non è un dato mancante: significa che il nodo non ha mai attraversato un record — input vuoto, filtro che scarta tutto a monte, o run abortita prima che il nodo partisse. È il corollario documentato su ProgressReporter.onNodeStart. In quel caso durationMs è 0: una durata non misurata si dichiara zero, non si deriva da un solo estremo.

Run fallita: topologia e conteggi parziali

Al primo errore fatale il runner annulla tutti i rami e aspetta che ogni sink e ogni task chiuda, poi assembla il report: una run failed non scrive più niente dopo essere stata riportata tale. error è il primo errore, quello del nodo che è fallito; i nodi degli altri rami risultano interrotti.

Un report failed prodotto da una run che è partita contiene la topologia reale e i conteggi accumulati fino all'abort — è proprio la run che più serve ispezionare, e lasciarla come guscio vuoto sarebbe un difetto.

Un report failed prodotto da buildFailedGraphReport (fallimento prima del grafo) ha invece nodes: [] / edges: [], perché non c'è alcuna topologia eseguita da riportare, e instances: [].

Il report allegato all'errore porta anche instances, col riepilogo di quanto contato fino all'interruzione.

nodeReports e writeReport

Ogni nodo del grafo (source/transform/lookup/ai/action/route/join/union/sink/try-catch/task) può ricevere un'opzione writeReport: (ctx, stats) => Record<string, unknown> in fase di costruzione: GraphBuilder.source/transform/lookup/ai/action/route/join/union/sink/tryCatch/task(..., { writeReport }). stats espone i conteggi del nodo (extracted/written/ rejected/rejectedReasons) così un nodo può riassumere i volumi senza scrivere il payload grezzo (es. { recordsFetched: 1_000_000 } invece del milione di record). Se né il nodo né la sua transform (o il suo task) dichiarano un writeReport, o se quello in uso lancia, il runner scrive il default { name: node.name, success }, dove success è true se il nodo non ha abortito la run (uno scarto isolato a dead-letter non conta come fallimento).

Una transform può dichiarare il proprio writeReport di default (Transform.writeReport): è il modo in cui profile porta il profilo delle colonne, assert l'esito dei suoi check, e i nodi lookup e ai i propri contatori (sotto lookup), nel report senza che la pipeline cabli nulla. Un task fa lo stesso con Task.writeReport: il task SQL ci porta le righe toccate da ogni statement. L'ordine è: writeReport del nodo, poi quello della transform, poi il default. Quello del nodo sostituisce quello della transform, non lo unisce. Il report dichiarato dalla transform si calcola anche quando il nodo fallisce, e un suo errore ricade sul default con un warning, come per quello del nodo. La forma di nodeReports non cambia (resta una mappa di oggetti liberi), quindi non ha inciso su reportSchemaVersion.

Sui nodi lookup, ai e action, l'opzione recordReport: { row, key?, maxRows? } aggiunge al report del nodo una riga a campi semplici per ogni record riuscito, sotto key (default records), accanto al report della transform (o a { name }, se la transform non ne ha uno). Un writeReport sullo stesso nodo è un errore di composizione. Al più maxRows righe (default 500, il resto in rowsOmitted; le righe che row non è riuscita a scrivere in rowsFailed), stringhe tagliate a 500 caratteri. I campi che row restituisce sono persistiti: la scheda della pipeline li elenca. Anche questo non tocca reportSchemaVersion.

processes.output_data riceve gli stessi report incrementalmente, durante l'esecuzione (non solo a fine run). Accanto a nodeReports viaggia runningNodes: string[] — i nodi che stanno elaborando adesso (partiti e non ancora completati), da ProgressReporter.onNodeStart. È ciò che permette alla UI di distinguere un nodo che lavora da uno in attesa, e quindi di mostrare due rami concorrenti. Assente sulle run concluse.

Con loro viaggia skippedNodes: { [nodeId]: { because } }, i nodi saltati fin qui. È sempre presente, {} se nessun nodo è saltato: una chiave assente vuol dire un motore precedente. Sta in una mappa a sé e non dentro nodeReports[nodeId], che è l'output di writeReport e ha le chiavi scelte dalla pipeline. Come runningNodes è assente sulle run concluse, dove nodes[].skipped del report dice la stessa cosa.

La quarta chiave è nodeCounts: { [nodeId]: { entered, exited, errored, caught?, held? } }, i contatori dal vivo. Anche lei sempre presente, {} finché nessun nodo ha contato nulla, e assente sulle run concluse, dove valgono i conteggi di nodes[]. noeva-server-api sostituisce output_data a ogni scrittura invece di fonderlo, quindi il worker rimanda ogni volta tutte e quattro le chiavi.

nodes ed edges sono la topologia reale del grafo eseguito: un nodo per ogni source/transform/lookup/ai/action/route/join/union/sink/try-catch/task dichiarato nel builder. Non esiste un nodo sintetico che collassa la catena.

Contatori dal vivo

Mentre la run è in corso ogni nodo ha tre contatori, più due che hanno solo alcuni nodi. I primi quattro sono gli stessi numeri del report, letti prima che la run finisca:

Contatore	Campo di NodeCounts	A fine run, nel report
Entrati	entered	extracted
Usciti	exited	written
In errore	errored	rejected + i caughtErrors di tutti i guard con sourceNodeId = il nodo
Catturati	caught, solo guard	caught del try-catch
In lavorazione	held, solo chi lo dichiara	— (non c'è: a fine run vale zero)

«In errore» conta un errore catturato da un try-catch sul nodo in cui è nato, non sul guard: il guard mostra a parte i suoi catturati. Un retry poi riuscito non è un errore. In circuit-breaker, e su un sink opaco, la cattura è una per evento e non per record: il nodo segna un errore anche se lo stream interrotto ne portava di più, come caught sul guard.

In lavorazione (held) sono i record che il nodo sta trattenendo ora. Non si ricava per differenza: c'è solo sui nodi che lo sanno, e assente non vuol dire zero.

Nodo	Cosa conta
join	i record del probe letti in anticipo e non ancora confrontati
try-catch	gli errori in coda che il ramo catch non ha ancora preso
un nodo con più edge in uscita (fan-out), un route	i record nel buffer del tee che il ramo più lento non ha ancora preso
una transform che dichiara Transform.held	i record ricevuti e non ancora fatti uscire
lookup, ai	i record della finestra: accumulati, in volo col round-trip, non ancora emessi
action	il record in volo: uno mentre l'azione non ha risposto
un sink con retry o rateLimit	il blocco del runner, in riempimento o in scrittura; un loadOne in volo
un sink senza policy che dichiara Sink.held	le righe nel batch del sink non ancora scritte

Le transform di libreria che lo dichiarano oggi sono sort, groupBy, pivot, chunk, serialize e assert in gate. Un nodo che ha due di queste voci (un sort con un fan-out) le somma. Due nodi trattengono record senza dichiararlo, e quindi non hanno il campo: dedup con keep: "last" e i buffer di prefetch di un union in concat.

Su lookup, ai e action conta record, non chiamate: un round-trip copre una finestra intera, e un resolver che fa più richieste in parallelo (httpLookup per chiave) non cambia il numero. Dove il runner esegue il nodo record per record (guard per-record, dead-letter di nodo, recordIds) la finestra è da uno. Sui sink Sink.held lo dichiarano Postgres, SQL Server e MySQL, upsert e SCD2 dove c'è. Con una policy il blocco del runner comprende già il batch del sink, quindi conta solo lui: sommarli conterebbe due volte le stesse righe. Il batch del sink conta finché lo statement non risponde; le righe già scritte ma non ancora confermate dal commit non sono né in lavorazione né uscite. A run conclusa, riuscita o fallita, l'ultima emissione porta held: 0 su ogni nodo che lo dichiara, anche se un fallimento ha lasciato un buffer pieno.

Gli usciti di un sink salgono quando il sink dice di aver scritto: a ogni record sotto un guard per-record, a ogni blocco di batchSize con una policy di retry. Senza nessuno dei due il sink scrive in un solo load, e i suoi usciti arrivano tutti a fine scrittura, mentre gli entrati salgono record per record.

Il runner li emette con ProgressReporter.onProgress: un evento per run, al più ogni secondo (PROGRESS_INTERVAL_MS), con i soli nodi cambiati. Un nodo mai partito o saltato non compare mai. A run conclusa, riuscita o fallita, arriva un'ultima emissione prima che runGraph restituisca: l'ultimo valore di ogni nodo è quello del report. Senza un onProgress nel reporter non parte nessun timer.

I nodi sono quelli del grafo eseguito: su una root con sub-pipeline, che l'appiattimento riduce a nodi piatti al build(), i contatori arrivano per id qualificato (inst, inst/per-due, inst/dentro/per-due), come nel report. Il nodo di confine d'ingresso di un'istanza ha l'id dell'istanza, ed è lì che un record fuori schema conta «in errore».

Il worker non inoltra ogni emissione: accumula i contatori nodo per nodo e li scrive su processes al più ogni cinque secondi, la prima volta subito e poi a fine finestra con l'ultimo valore, se nel frattempo un avvio o un completamento di nodo non li ha già portati. Una scrittura fallita si riprova alla finestra dopo. A run conclusa il worker chiude la finestra e aspetta la scrittura in volo prima del report finale: arrivata dopo, lo sostituirebbe. Chi segue una run sul worker col solo polling vede quindi i contatori fino a ~5 s indietro, più i 2 s del polling della UI.

Canale	Dove
ProgressReporter.onProgress	update.nodes: NodeCounts[]
Stream SSE GET /runs/:id/events	node-progress con nodes; a chi si connette tardi il bus ritrasmette solo l'ultimo valore per nodo
Polling su processes (noeva-server-api)	output_data.nodeCounts: { [nodeId]: { entered, exited, errored, caught?, held? } }, scritto dal worker al più ogni ~5 s (WORKER_PROGRESS_INTERVAL_MS)

Il report non cambia: i contatori dal vivo non aggiungono campi, e lo schema resta quello di § «Versione di pipeline e release» (reportSchemaVersion 8). Chi riapre una run conclusa ricava «in errore» da rejected e da caughtErrors[].sourceNodeId. La lista è limitata a caughtErrorsLimit voci per guard (default 100): oltre il tetto, una parte dei catturati non è più attribuibile a un nodo, e il numero ricavato dal report è un minimo. La UI lo segna con «≥».

Scarti: report vs payload

Il report contiene solo conteggi e categorie di scarto (niente payload → niente PII). I record scartati completi vanno nel dead-letter sink, che è scelto dal chiamante (file, tabella, coda). Così il servizio resta agnostico e il report è sicuro da archiviare.

Il DeadLetter porta quindi campi che il report non ha, perché contengono payload reale:

Campo	Quando	Nel report?
raw	catture per-record (transform isolate, sink con loadOne)	Solo con includeRawInReport sul guard
lastConsumed	nodi non isolabili, durante il consumo	Mai
guardSnapshots	record d'ingresso da cui deriva il fallimento, risalendo derivedFrom	Mai

Sintetici e quindi sicuri da esporre: error.context.recordId (id del record entrato nel nodo che ha fallito) e error.context.derivedFrom (id dei contribuenti, quando il fallimento avviene in emissione di un nodo stateful). Sono id, non dati.

Nota: guardSnapshots è additivo e opzionale, quindi non ha inciso su reportSchemaVersion.

Per seguire quali record sono passati da un nodo, e com'erano, c'è la traccia di una run tracciata (etl run --trace): un documento separato, con i valori in chiaro, che il report non contiene. Vedi La traccia di una run.

Versione di pipeline e release

Una run dice con quale codice è girata in quattro campi. Lo schema v7 ha separato la versione di pipeline dalla release; lo schema v8 aggiunge la versione del core e il cliente:

• pipelineVersion è la versione di pipeline: con quale codice della pipeline è girata la run. È l'ultima voce del changelog della pipeline nel catalogo, e la mettono nel RunContextetl run, il worker e le run tracciate lanciate da etl serve, non l'ambiente. Una pipeline pronta porta il suo changelog con sé, dal pacchetto del core: la sua versione è quella che il core le ha dato. Un grafo lanciato fuori dal catalogo, come nei test, ha null.
• release è la release del deploy: con quale artefatto è girata, cioè il commit del repo cliente. Due run con la stessa pipelineVersion e release diverse sono girate sullo stesso codice di pipeline ma su un artefatto diverso. Viene da ETL_RELEASE se è impostata (l'immagine la riceve come build arg); altrimenti, quando il deploy gira da un checkout (sotto PM2), è il commit del repo dell'etl.config.ts, git:<sha> con -dirty se ha modifiche non committate a file tracciati (src/release.ts). Si chiede a git nella cartella del config e non in quella del core, che in un repo cliente sta in node_modules; se quella cartella non è la radice del checkout, la release è null. Si calcola una volta all'avvio del processo.
• core è la versione del core: la version del package.json del pacchetto installato, letta a runtime e mai da un build arg. La release la fissa già, attraverso il lockfile del repo cliente, ma qui si legge senza aprirlo. Se il pacchetto non si trova, l'avvio si ferma con un ConfigError che nomina dove l'ha cercato.
• client è il cliente del deploy, il client di defineEtlConfig.

etl run, il worker e le run tracciate calcolano i tre campi del deploy dal catalogo caricato (src/deploy-identity.ts), uguali per tutte le run del processo. Un grafo lanciato fuori dalla CLI li ha a null.

Fino allo schema v6 la release stava in version, un nome ambiguo. È l'unico rename dello schema: un consumer che legge version non lo trova più, e deve leggere release a partire da reportSchemaVersion 7. Il report del worker arriva com'è in processes.output_data, quindi anche lo storico Noeva porta tutti i campi.

Lo store locale delle run (etl runs, GET /runs/:id) rilegge i report salvati con uno schema precedente nella forma corrente: prima del v7 version diventa release e pipelineVersion vale null, prima del v8 core e client valgono null, perché quelle run non li hanno mai registrati. reportSchemaVersion resta quello con cui il report è nato.

Esempio (successo)

{
  "reportSchemaVersion": 8,
  "runId": "ebe7d7cc-…",
  "pipeline": "sessions-daily",
  "status": "success",
  "ok": true,
  "env": "dev",
  "release": "git:6936740b1d56",
  "core": "1.4.0",
  "client": "acme",
  "pipelineVersion": 1,
  "startedAt": "2026-06-03T19:02:10.878Z",
  "finishedAt": "2026-06-03T19:02:10.927Z",
  "durationMs": 49,
  "nodes": [
    {
      "id": "sessions",
      "kind": "source",
      "name": "trace_sessions",
      "extracted": 74,
      "written": 0,
      "rejected": 0,
      "rejectedReasons": {},
      "caught": 0,
      "caughtErrors": [],
      "startedAt": "2026-06-03T19:02:10.880Z",
      "finishedAt": "2026-06-03T19:02:10.902Z",
      "durationMs": 22
    },
    {
      "id": "raw-contract",
      "kind": "transform",
      "name": "raw-contract",
      "extracted": 74,
      "written": 74,
      "rejected": 0,
      "rejectedReasons": {},
      "caught": 0,
      "caughtErrors": [],
      "startedAt": "2026-06-03T19:02:10.881Z",
      "finishedAt": "2026-06-03T19:02:10.903Z",
      "durationMs": 22
    },
    {
      "id": "aggregate-daily",
      "kind": "transform",
      "name": "aggregate-daily",
      "extracted": 74,
      "written": 14,
      "rejected": 0,
      "rejectedReasons": {},
      "caught": 0,
      "caughtErrors": [],
      "startedAt": "2026-06-03T19:02:10.882Z",
      "finishedAt": "2026-06-03T19:02:10.918Z",
      "durationMs": 36
    },
    {
      "id": "sqlserver:dbo.fact_session_daily",
      "kind": "sink",
      "name": "sqlserver:dbo.fact_session_daily",
      "extracted": 14,
      "written": 14,
      "rejected": 0,
      "rejectedReasons": {},
      "caught": 0,
      "caughtErrors": [],
      "startedAt": "2026-06-03T19:02:10.883Z",
      "finishedAt": "2026-06-03T19:02:10.926Z",
      "durationMs": 43
    }
  ],
  "edges": [
    { "from": "sessions", "to": "raw-contract" },
    { "from": "raw-contract", "to": "aggregate-daily" },
    { "from": "aggregate-daily", "to": "sqlserver:dbo.fact_session_daily" }
  ],
  "params": {},
  "extracted": 74,
  "written": 14,
  "rejected": 0,
  "rejectedReasons": {},
  "metrics": { "extracted": 74, "written": 14 },
  "nodeReports": {
    "sessions": { "name": "trace_sessions", "success": true },
    "raw-contract": { "name": "raw-contract", "success": true },
    "aggregate-daily": { "name": "aggregate-daily", "success": true },
    "sqlserver:dbo.fact_session_daily": {
      "recordsWritten": 14,
      "success": true
    }
  },
  "state": {},
  "instances": [],
  "error": null
}

Esempio (fallimento durante la run)

La topologia c'è, e i conteggi dicono fin dove la run era arrivata: la source aveva già letto 40 record, il sink non ne aveva ancora scritto nessuno, e il nodo a valle non si è mai concluso (finishedAt: null).

{
  "reportSchemaVersion": 8,
  "status": "failed",
  "ok": false,
  "pipelineVersion": 1,
  "extracted": 40,
  "written": 0,
  "nodes": [
    {
      "id": "sessions",
      "kind": "source",
      "name": "trace_sessions",
      "extracted": 40,
      "written": 40,
      "rejected": 0,
      "rejectedReasons": {},
      "caught": 0,
      "caughtErrors": [],
      "startedAt": "2026-06-03T19:02:10.880Z",
      "finishedAt": "2026-06-03T19:02:12.104Z",
      "durationMs": 1224
    },
    {
      "id": "sqlserver:dbo.fact_session_daily",
      "kind": "sink",
      "name": "sqlserver:dbo.fact_session_daily",
      "extracted": 40,
      "written": 0,
      "rejected": 0,
      "rejectedReasons": {},
      "caught": 0,
      "caughtErrors": [],
      "startedAt": "2026-06-03T19:02:10.884Z",
      "finishedAt": null,
      "durationMs": 0
    }
  ],
  "edges": [{ "from": "sessions", "to": "sqlserver:dbo.fact_session_daily" }],
  "state": {},
  "instances": [],
  "error": {
    "type": "LoadError",
    "message": "Failed to connect …",
    "stage": "fact_session_daily",
    "nodeId": "sqlserver:dbo.fact_session_daily"
  }
}

Esempio (fallimento prima della run)

Wiring dei connettori fallito: nessun grafo è mai partito, quindi non c'è topologia da riportare. È il caso coperto da buildFailedGraphReport.

{
  "reportSchemaVersion": 8,
  "status": "failed",
  "ok": false,
  "pipelineVersion": 1,
  "extracted": 0,
  "written": 0,
  "nodes": [],
  "edges": [],
  "state": {},
  "instances": [],
  "error": {
    "type": "ConfigError",
    "message": "PGHOST non configurato",
    "stage": null,
    "nodeId": null
  }
}