Nodo: try-catch

Un nodo di passaggio che apre uno scope di gestione errori sui nodi a valle. I record escono invariati dalla porta out; gli errori dei nodi dello scope arrivano come DeadLetter sulla porta catch, che è un ramo di nodi veri: si può registrare l'errore, filtrare, rimappare e decidere dove va a finire il record.

0/4

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

Metodo	.tryCatch(from, id, { mode?, includeRawInReport?, guardSnapshots?, snapshotLimit?, onSnapshotOverflow?, purpose?, writeReport? })
Ingressi	1
Uscite	≥1 sulla porta out (predefinita) + esattamente 1 sulla porta catch
Dead-letter	no: è lui a raccogliere gli errori
Codice	src/core/graph.ts, src/core/graph-runner.ts

Quando usarlo

Serve…	Meccanismo
scartare il record che fallisce in un nodo	deadLetter su quel nodo
gestire gli errori di un tratto di grafo, sink compresi	try-catch
instradare i record falliti (recupero, sink di riserva, notifica)	try-catch + ramo catch
fermarsi al primo errore perché il batch è atomico	niente: è il default

Lo scope

Lo scope sono i nodi raggiungibili dalla porta out, fino ai sink, fermandosi ai try-catch annidati, che aprono uno scope loro. Ne fanno parte anche i nodi con una deadLetter propria: i loro errori per record finiscono prima nella loro dead-letter, e al guard arrivano solo quelli che non erano isolabili.

Conta dove l'errore è nato. Un errore che arriva da monte — da una source, o da un nodo prima del guard — risale e ferma la run, anche se attraversa lo scope: il guard protegge un tratto di grafo, non tutto ciò che ci passa dentro. I rami paralleli fuori dallo scope proseguono.

Esempio

.tryCatch("arricchisci", "guardia") // mode "per-record" di default
.transform("guardia", "contratto-fatto", validate("contratto-fatto", fattoSchema)) // protetto
.sink("contratto-fatto", fatti)                                                   // protetto
// Ramo catch: riceve un DeadLetter per ogni errore dello scope.
.transform("guardia", "registra-errore", mapTransform<DeadLetter, DeadLetter>("registra-errore", (dl, ctx) => {
  ctx.logger.warn("fatture.errore", { nodeId: dl.error.context.nodeId ?? null, message: dl.error.message });
  return dl;
}), { fromPort: "catch" })
.sink("registra-errore", errori)

Recupero verso un sink di riserva, quando il sink protetto è isolabile (ha loadOne):

.transform("registra-errore", "da-recuperare", filterTransform<DeadLetter>("da-recuperare",
  (dl) => dl.error.context.nodeId === fatti.name && dl.raw !== undefined))
.transform("da-recuperare", "ripristina", mapTransform<DeadLetter, Fatto>("ripristina", (dl) => dl.raw as Fatto))
.sink("ripristina", riserva)

Le modalità

mode	Semantica	DeadLetter.raw
"per-record" (default)	Ogni record che fallisce va sul catch, lo stream continua. N errori → N sul catch	presente
"circuit-breaker"	Al primo errore il sotto-flusso si ferma; al più un record per nodo sul catch	assente

Sotto per-record un sink con loadOne (per esempio arraySink, smtpSink, sendgridSink, noevaFilesSink) scrive un record alla volta, così un fallimento di scrittura porta con sé il record. Un sink senza loadOne resta opaco: il suo errore arriva al catch, ma senza raw. Una transform stateful non si isola mai per record.

Niente output parziali sotto per-record. Il guard promette che un record passa o dal try o dal catch. Un nodo che dichiara emit: "streaming" — un'espansione in streaming — può far uscire dei record e poi fallire: il record passerebbe da entrambi, e un fallback sul catch rischierebbe di riscrivere ciò che il try ha già scritto. validateGraph rifiuta quindi un nodo in streaming i cui errori arriverebbero al guard. Le vie d'uscita: una deadLetter sul nodo, che ferma i suoi errori prima del guard, oppure il nodo fuori dallo scope. Anche emit: "atomic" basta.

Niente verdetti a fine stream, in nessuna modalità, salvo lo scambio. Un nodo che dichiara emit: "before-verdict" — un assert in streaming con almeno un check error — fa uscire i record e li giudica dopo l'ultimo. Un guard assorbirebbe quel verdetto: lo stream si chiuderebbe pulito e il sink caricherebbe i dati rifiutati. validateGraph lo rifiuta in per-record e in circuit-breaker, anche con una deadLetter sul nodo. Le vie d'uscita: assert in gate, il nodo fuori dallo scope, oppure il write-then-swap: se ogni sink a valle ha un post-task, e a valle non ci sono action, il grafo si costruisce, perché i dati rifiutati restano in uno staging che nessuno scambia (vedi la matrice in Controlli sul dataset).

Un errore catturato ferma i post-task. Quando il guard cattura l'errore di un nodo e ne chiude lo stream, l'input dei sink a valle è troncato: scrivono ciò che hanno ricevuto, ma per gli edge di controllo non sono riusciti, e un task che li aspetta è saltato. Vale anche se l'errore arriva dopo l'ultimo record. I record che un guard per-record manda sul catch uno per uno non troncano niente.

A run annullata il guard non cattura niente. Un nodo che sta aspettando (un backoff, un polling) quando la run viene annullata riceve la ragione dell'annullamento, e spesso la avvolge in un errore suo (LookupError, LoadError). Non è il fallimento di quel record: il guard non lo manda sul catch, la dead-letter di nodo non lo riceve, e la run si chiude failed. Vale anche per l'arresto del worker.

Un nodo che fallisce senza aver emesso niente salta i sink a valle. Il circuit-breaker chiude lo stream pulito, ma il sink a cui non è arrivato nessun record non chiama load e nel report risulta skipped. È ciò che rende sicuro un assert in gate dentro un guard: vedi Controlli sul dataset.

Il DeadLetter

interface DeadLetter<I = unknown> {
  raw: I; // il record su cui l'errore è avvenuto, quando isolabile
  error: EtlError; // error.context.nodeId dice quale nodo
  lastConsumed?: unknown; // l'ultimo record letto, quando il nodo non era isolabile
  guardSnapshots?: { id: string; value: unknown }[]; // i record d'ingresso da cui deriva
  snapshotEvicted?: boolean;
}

lastConsumed e guardSnapshots contengono dati reali: non vanno mai nel report.

Opzioni

Opzione	Default	Effetto
mode	"per-record"	Vedi sopra
includeRawInReport	false	Mette il corpo dei record falliti nel report. ⚠️ Persiste dati reali
guardSnapshots	false	Trattiene i record che attraversano il guard, per risalire agli antenati di un errore a valle di un'aggregazione. Richiede graph(name, { recordIds: true })
snapshotLimit	1000	Tetto sugli snapshot tenuti insieme
onSnapshotOverflow	"fail"	A buffer pieno: "fail" ferma la run, "drop-oldest" scarta i più vecchi e marca snapshotEvicted

Nel report

Il guard conta caught e caughtErrors (lista limitata da caughtErrorsLimit, default 100). Una run con errori catturati termina partial, non failed: è arrivata in fondo con anomalie gestite. Un task non sta mai nello scope di un guard: lo scope segue gli edge di dati.

Nei contatori dal vivo un errore catturato è in errore sul nodo che ha fallito, quello del suo sourceNodeId, non sul guard; il guard mostra a parte quanti ne ha catturati. Così il DAG indica dove nascono gli errori, non dove vengono fermati. In circuit-breaker se ne conta uno per errore catturato, non uno per record dello stream interrotto (convenzione del report).

Il guard ha anche un in lavorazione: gli errori catturati in coda che il ramo catch non ha ancora preso. Sale quando il sink degli errori è più lento dello scope, e a run conclusa vale zero.

Anti-pattern

• Un try/catch JavaScript dentro una transform: nasconde l'errore invece di instradarlo.
• Un ramo catch che finisce in un sink senza che nessuno lo guardi: il guard serve a decidere dove va il record, non solo a zittire l'errore.