0032. Sink proteggibili da un guard e isolabili per-record (path di fallback)
- Stato: Accettata
- Data: 2026-07-27
- Decisori: Team BI/ETL
Contesto
Con ADR-0029 e ADR-0031 il guard cattura per-record gli errori dei nodi transform/join protetti. I sink erano esclusi: computeTryScopes li trattava come confine dello scope, e il loro errore veniva rilanciato dalla fase di esecuzione dei sink, abortendo la run.
Questo rendeva inesprimibile un caso concreto e ricorrente: un path che termina con una chiamata HTTP; se la chiamata fallisce si finisce nel catch e si prosegue con un secondo path di fallback che salva il record in un altro modo. La chiamata HTTP è un sink — quindi la run abortiva e basta.
Il punto di fondo: un guard non serve solo a loggare, serve a instradare. Finché i sink restavano fuori dallo scope, il guard era cieco esattamente sul punto in cui i fallimenti sono più probabili, cioè l'I/O verso sistemi esterni.
Due ostacoli distinti, non uno:
- Topologico —
computeTryScopes(src/core/graph-runner.ts) escludeva i sink dascopeByNode, quindi non c'era untryIda cui deviare l'errore. - Contrattuale —
Sink<T>.load(records: AsyncIterable<T>, ctx)consuma l'intero stream e restituisce un totale: dall'esterno è impossibile sapere quale record abbia fatto fallire la scrittura. È la stessa ragione per cui i sink non hanno mai avuto la dead-letter per-nodo (ADR-0027).
Decisione
1. loadOne opzionale sul contratto Sink
loadOne?(record: T, ctx: RunContext): Promise<void>;Un sink che lo implementa è isolabile per-record. load resta invariato ed è l'unico path usato fuori da uno scope guard.
Perché opzionale e non sostitutivo di load:
- Retrocompatibilità: i sink esistenti non cambiano e mantengono il fail-fast di oggi. Nessuna firma esistente è stata toccata.
- Throughput: i sink batch (COPY, INSERT multipla, upload di un file intero) perderebbero moltissimo se forzati a una scrittura per record. L'isolamento si paga solo dove serve, cioè dentro uno scope guard.
2. I sink entrano nello scope del guard
computeTryScopes non tratta più il sink come confine. Restano confine i soli try-catch annidati, che aprono il proprio scope. I sink non hanno edge in uscita (garantito da validateGraph), quindi entrano nello scope senza far proseguire la visita.
Effetto collaterale voluto: un guard il cui out sfocia direttamente in un sink aveva scopeSize 0 e la coda catch veniva chiusa subito; ora è uno scope reale.
3. Dispatch a tre vie nella fase sink
| Condizione | Comportamento |
|---|---|
in scope guard per-record e loadOne presente | scrittura per record; EtlError → ramo catch con il record (DeadLetter.raw); lo stream prosegue |
in scope guard (sink opaco, oppure guard circuit-breaker) | load in try/catch; errore → ramo catch con raw: undefined; la run non aborta, il nodo è marcato success: false |
| fuori scope | invariato: fail-fast, la run aborta |
Gli errori non-EtlError continuano a propagare in tutti e tre i casi: sono bug di programmazione, non fallimenti di dominio.
4. recordId posizionale
EtlErrorContext.recordId era dichiarato e mai popolato. Ora il runner lo valorizza nel path per-record dei sink, nella forma ${runId}:${nodeId}:${seq}, ed è esposto in CaughtErrorSummary come campo opzionale.
È sintetico e posizionale — «il seq-esimo record consumato da quel nodo» — quindi non deriva dai dati, non può contenere informazioni personali ed è sicuro da esporre nella UI, coerentemente con la sanitizzazione di CaughtErrorSummary. Non è un id di lineage: non viene propagato da un nodo al successivo.
Alternative considerate
Batch con esiti per record — load restituisce anche gli scarti (LoadResult & { rejected: {record, error}[] }). Più efficiente per i sink batch, ma cambia una firma esistente e sposta l'onere dell'isolamento su ogni connettore. Scartata: loadOne ottiene lo stesso risultato senza rompere nulla.
Solo topologia — includere i sink nello scope e catturare l'errore del load come evento singolo, senza raw. Poco costoso, ma non abilita il fallback perché il path alternativo non riceve il payload. Adottata come comportamento di ripiego per i soli sink opachi, non come soluzione.
Prima l'identità di record con lineage (issue #78) — envelope TaggedRecord con id/derivedFrom e snapshot all'ingresso del guard, come prerequisito. Scartata come prerequisito: se il sink è isolato per-record, il record che ha fallito è già in mano nel punto di fallimento, quindi lo scenario di punta non richiede né envelope né lineage né buffer di snapshot. Per un fallback che deve persistere il record, il payload che il sink stava scrivendo è anzi più utile dello snapshot all'ingresso del guard, che è uno stato precedente.
Quella infrastruttura resta necessaria per i casi in cui il record che fallisce non è quello entrato nel guard: fallimento dentro un nodo stateful, o a valle di un groupBy. Issue #78 resta aperta con quello scope residuo.
Conseguenze
Positive
- Il path di fallback è esprimibile end-to-end e dimostrato nella demo
orders-enriched-daily. - Il guard copre finalmente il punto di fallimento più probabile (I/O esterno).
recordIdnon è più un gancio morto: la UI può correlare l'errore al record.- Nessuna modifica alle firme pubbliche: retrocompatibile a compilazione.
Negative / da tenere d'occhio
- Cambio di comportamento per le pipeline che hanno già un sink a valle di un guard: quel sink entra ora nello scope, quindi un suo errore non aborta più la run ma produce una run
partial. È il comportamento voluto, ma va comunicato. - I sink isolabili pagano una scrittura per record dentro gli scope guard. Il requisito di idempotenza già dichiarato su
Sinkvale identico perloadOne: un retry dello stesso record non deve produrre duplicati. - Restano due classi di sink (isolabili e opachi) con garanzie diverse sul
DeadLetter. Il report distingue i casi, ma è una asimmetria da documentare per chi scrive un nodo sul ramocatch.
Non toccato
GRAPH_REPORT_SCHEMA_VERSIONresta a3:recordIdsuCaughtErrorSummaryè additivo e opzionale, quindi retrocompatibile con la UI attuale.- La precedenza dead-letter propria vs guard resta invariata.