0035. Identità di record e lineage derivedFrom, con snapshot al guard

• Stato: Accettata
• Data: 2026-07-28
• Issue: #78
• Correlate: ADR-0029 (nodo try-catch), ADR-0031 (guard per-record), ADR-0032 (sink isolabili), ADR-0033 (lastConsumed), ADR-0034 (includeRawInReport)

Contesto

Dopo ADR-0031/0032/0033 il ramo catch di un guard porta il record colpevole nei casi per-record (DeadLetter.raw) e l'ultimo record consumato nei nodi non isolabili (lastConsumed). Restavano due casi scoperti, entrambi con la stessa causa: il record che fallisce non è quello entrato nel guard.

1. Errore in fase di emissione di un nodo stateful. In un groupBy, un errore in finalize è causato da un gruppo. lastConsumed resta deliberatamente undefined in emissione (ADR-0033): l'ultimo record dello stream è scorrelato dal colpevole.
2. Nodo a valle di un groupBy. raw è valorizzato ma è l'aggregato: i record originali che hanno formato quel gruppo non sono ricostruibili.

Senza una relazione tra record, un nodo del ramo catch non può implementare un path di fallback per questi casi: ha un errore, non i dati.

Il vincolo che ha determinato il design

L'analisi ha stabilito che un envelope a livello runner, da solo, non basta per il caso 1.

groupBy (src/transforms/group-by.ts) accumula l'intero stream in una Map e poi emette un record per gruppo. La funzione di raggruppamento è codice utente dentro apply: il runner vede apply(input) → output e non ha modo di sapere che il terzo output deriva dagli input 2, 7 e 9. Qualunque lineage inferito da fuori può solo essere «tutti i record consumati dal nodo» — che su un groupBy è l'intero input, non il gruppo colpevole.

Le alternative valutate:

Opzione	Caso 1	Costo
Buffer al guard senza envelope	«i record entrati nel guard»	Basso
Envelope + derivedFrom inferito	idem — il gruppo resta invisibile	Alto
Envelope + contratto cooperativo	il gruppo colpevole	Alto + nuova API

Le prime due consegnano lo stesso risultato sui casi che motivano l'issue; la terza è l'unica che mantiene la promessa. Si è scelta la terza.

Decisione

1. L'identità viaggia in un envelope side-channel, non nel payload

interface TaggedRecord {
  readonly id: string;
  readonly derivedFrom: readonly string[];
  readonly branch: string | null;
  readonly value: unknown;
}

Un campo iniettato nel payload (__etl_id) è escluso per una ragione dirimente: Zod rimuove silenziosamente le chiavi sconosciute dagli oggetti non-strict, quindi ogni nodo validate() a valle lo cancellerebbe senza errore — e nella demo orders-enriched-daily c'è esattamente un validate() subito dopo il guard. Verrebbe inoltre scritto sui sink reali e perso dalle merge dei join e dagli output di groupBy, che costruiscono oggetti nuovi campo per campo. Il preteso vantaggio («sopravvive gratis») è falso: il runner dovrebbe re-iniettarlo dopo ogni nodo, cioè fare comunque il lavoro dell'envelope, sporcando anche i dati.

L'envelope è spacchettato prima di ogni chiamata a codice utente (transform.apply, join.buildKey/probeKey/merge, sink.load/loadOne) e ri-avvolto dopo. Payload, schemi Zod, connettori e firme pubbliche restano invariati: uno schema .strict() a valle passa senza modifiche.

Formato dell'id: ${runId}:${sourceNodeId}:${seq} — la stessa forma del recordId posizionale di ADR-0033, volutamente, così la sostituzione è trasparente per i consumatori del report. Non persistito: è un artefatto di processo, vive quanto la run.

2. Le transform stateful dichiarano il proprio lineage

interface Transform<I, O> {
  applyWithLineage?(
    records: AsyncIterable<{ value: I; id: string }>,
    ctx: RunContext,
  ): AsyncIterable<{ value: O; derivedFrom: readonly string[] }>;
}

Opzionale e additiva. Il contratto tiene la coniatura degli id nel runner: la transform riceve gli id, restituisce solo valore e antenati, non vede branch e non conia nulla — quindi non può corrompere l'envelope. È il minimo potere necessario.

Per il caso 1 serve anche attribuire un errore sollevato in emissione, dove non esiste un record in mano: lo fa EtlErrorContext.derivedFrom, popolato via withDerivedFrom. groupBy avvolge finalize e, su EtlError, rietichetta l'errore con i contribuenti del gruppo che ha sollevato. Contiene solo id sintetici, quindi è sicuro da esporre come recordId.

Chi non implementa applyWithLineage cade su TransformGraphNode.statefulLineage: "last" (default, costo costante), "all" (tutti i consumati — correlazione garantita ma imprecisa, O(record) in id), "none".

3. Regole di tagging

Nodo	derivedFrom dell'output
source	[]
transform stateless	[inputId]
join inner / left con match	[buildId, probeId]
join left senza match	[probeId]
transform stateful cooperante	dichiarato dalla transform
transform stateful opaca	secondo statefulLineage
fan-out (tee)	stesso id, branch = nodeId di destinazione
ramo catch	id nuovo, derivedFrom = [id del record fallito]

Nel path taggato una transform stateless gira sempre isolata per-record: è l'unico modo per correlare esattamente output e input, e per una stateless non cambia la semantica (processa già record per record). Conseguenza da conoscere: una transform che dichiara stateful: false ma trattiene stato tra record si comporterebbe diversamente — ma è esattamente ciò che il flag stateful serve a dichiarare.

Sul ramo catch il DeadLetter riceve un id nuovo invece di riusare quello del record fallito: sono due record distinti (il record e il verbale del suo fallimento) e non possono condividere un'identità, o la mappa degli snapshot avrebbe due significati per la stessa chiave. La relazione è conservata dove le relazioni vivono, in derivedFrom.

4. Snapshot al guard, con risalita della catena

Un guard con guardSnapshots: true trattiene id -> payload per ogni record che attraversa il suo pass-through, e id -> antenati per i record derivati dentro lo scope (solo id, nessun payload).

Su una cattura la lookup non è map.get(failedId): il record fallito ha un id diverso da quello entrato nel guard, e con un'aggregazione in mezzo gli antenati sono N. È una risalita di derivedFrom fino agli id presenti nella mappa (collectAncestors).

Il risultato arriva sul DeadLetter come guardSnapshots, ed è payload reale: resta fuori dal report, come lastConsumed. Se lo si vuole nel report deve passare da includeRawInReport (ADR-0034) o da un meccanismo altrettanto esplicito.

5. Overflow: "fail" di default

snapshotLimit (default 1000) sul totale delle due mappe, e onSnapshotOverflow:

• "fail" (default): la run aborta con un messaggio che dice cosa alzare.
• "drop-oldest": FIFO, run che continua, eviction contata e snapshotEvicted: true.

Il default è "fail" e non "drop-oldest" perché un path di fallback senza snapshot non può funzionare, e con l'eviction lo scoprirebbe solo a valle, su un DeadLetter senza antenati indistinguibile da «nessun antenato esisteva». Fallire dove il problema è, con un messaggio azionabile, è meglio di degradare in silenzio.

L'errore di overflow è un Error non-EtlError, deliberatamente: un EtlError sarebbe catturato da un guard esterno (try annidato) e la run proseguirebbe, che è l'opposto di ciò che "fail" promette. Non è un fallimento di dominio ma un limite operativo del runner, come gli altri invarianti interni.

Il refcount valutato nell'issue non aiuta. Con un nodo stateful nello scope il buffer è inerentemente O(input): ogni record consumato resta un possibile antenato finché l'aggregazione non ha finito di emettere, quindi il refcount resterebbe > 0 per tutti. Il cap con fallimento esplicito è la scelta onesta, non un compromesso temporaneo.

6. Attivazione: opt-in su due livelli

graph(name, { recordIds: true }) per l'identità; guardSnapshots: true sul singolo guard per gli snapshot. Due livelli perché hanno costi diversi: il primo è un envelope per record, il secondo tiene payload reale in memoria e va deciso consapevolmente.

A feature spenta si eseguono i path preesistenti, invariati — funzioni separate, non le stesse con un if per record — così comportamento e costo restano quelli di prima.

Conseguenze

Positive

• I due casi dell'issue sono chiusi, con il gruppo colpevole e non l'intero input.
• recordId diventa un id che si propaga: correlabile a valle, stessa forma stringa di prima, quindi sostituzione trasparente per i consumatori.
• Il join dichiara ora il recordId sul rigetto per-record, chiudendo un'asimmetria residua di ADR-0033 (le transform lo facevano, i join no).
• Nessun impatto sulle pipeline che non attivano recordIds.
• GRAPH_REPORT_SCHEMA_VERSION resta 3: tutte le aggiunte sono opzionali e additive.

Negative / limiti dichiarati

• Due code path nel runner per i nodi taggati e non. È il prezzo del vincolo «costo identico a feature spenta»; il rischio di divergenza è mitigato da test paralleli acceso/spento sugli stessi scenari.
• Il lineage esatto richiede cooperazione. Una transform stateful di terze parti che non implementa applyWithLineage degrada alla policy del nodo. groupBy è l'unica stateful spedita e coopera.
• Il lato build di un join alimentato da fuori lo scope non ha snapshot: quei record non attraversano il guard. Verificato da un test che documenta il limite.
• Buffer O(input) con un nodo stateful nello scope, per il motivo sopra.

Rapporto con ADR-0033

lastConsumed resta e non è deprecato: copre i casi in cui il lineage non è attivo (è opt-in) e ha costo zero. Il lineage copre ciò che lastConsumed per costruzione non può:

Caso	lastConsumed	guardSnapshots
Stateful che fallisce in consumo	✅ il record colpevole	✅ (superset)
Stateful che fallisce in emissione	❌ undefined per scelta	✅ i record del gruppo
Nodo a valle di un groupBy	❌ non applicabile	✅ i record originali
Circuit-breaker su sink opaco	✅ ultimo consumato	✅ gli antenati nel guard

Il recordId posizionale introdotto in ADR-0033 è sostituito da quello con lineage quando recordIds è attivo; a feature spenta resta quello posizionale.