Runbook: stato di pipeline e watermark

Backfill, reset e diagnosi dello stato delle source incrementali.

Sintesi

• Cosa copre: lo stato di pipeline delle source incrementali: leggerlo, rileggere un intervallo (backfill), riportarlo indietro (reset). Non copre la logica delle singole pipeline.

• Dove vive lo stato:

  Chi esegue	Store
  etl worker, etl serve	noeva-server-api, workspace del processo (run normali)
  etl run --workspace <id>	noeva-server-api, lo stesso stato del worker
  etl run senza --workspace	un file per pipeline in ETL_STATE_DIR (default .state/)
  run tracciata da etl serve	lo stesso file locale di etl run

  I due stati sono separati: una run da CLI senza --workspace non vede né tocca quello del worker. Se l'ambiente configura NOEVA_WORKSPACE_ID o NOEVA_WORKSPACE_IDS, il workspace del flag dev'essere uno di quelli: altrimenti stato e dati finirebbero su workspace diversi. Lo stato non sta mai nello storico .runs/, che ha un pruning.

• Regola da ricordare: il watermark avanza da solo, a run conclusa e riuscita. Arretra solo con un reset.

• Criticità: alta. Un reset in avanti senza aver messo al sicuro gli scarti perde dati in modo permanente; un reset all'indietro fa rileggere, e i sink idempotenti assorbono.

Leggere lo stato

etl state show <pipeline>                     # stato locale
etl state show <pipeline> --workspace <id>    # stato del worker, su noeva-server-api
etl state show <pipeline> --node <id> --limit 50 --json

Per ogni nodo source: tipo, valore, instanceId, generazione, ultimo aggiornamento; sotto, gli ultimi eventi (commit con la run, reset con autore e motivo), dieci se non si passa --limit. Exit code: 0 riuscito, 2 argomenti o configurazione sbagliati (un flag sconosciuto, NOEVA_API_KEY mancante), 1 errore dello store. Un flag scritto male o nella forma --workspace=<id> è rifiutato: non ricade in silenzio sullo stato locale. Con --workspace servono NOEVA_API_BASE_URL e NOEVA_API_KEY, e la chiave dev'essere di un membro del workspace.

Nel report della run c'è lo stesso dato dal punto di vista della run, in state.<nodeId>:

Campo	Significato
read	il watermark da cui la source ha letto (prima del lookback)
origin	stored (dallo stato), override (da params), initial (prima run o dopo un reset)
proposed	il massimo visto sullo stream estratto; null se non ha visto righe
committed	se il watermark è stato committato
reason	perché no, quando committed è false

Da CLI: etl runs show <runId> --json | jq .state. Per il worker il report è in processes.output_data:

select id, completed_at, status, output_data -> 'state' as state
from processes
where function = '<pipeline>'
order by completed_at desc
limit 10;

Diagnosi per reason

reason	Cosa significa	Cosa fare
failed	run fallita: lo stato è fermo	niente sullo stato: si corregge e si rilancia
partial	run con scarti, e la pipeline non dichiara commitOn: "partial"	§ Finestra bloccata
records_lost	run con scarti che non sono tutti dead-letter per record	§ Record persi
skipped	la source non è partita: un nodo da cui dipende (after) non è riuscito	guardare nodes[].skipped.because nel report, e quel nodo
no_records	la source non ha visto righe	normale solo alla prima run su una sorgente vuota: con >= una run successiva rilegge almeno il confine
generation_mismatch	un reset è avvenuto durante la run	niente: la run successiva parte dal valore del reset
type_mismatch	lo store ha un valore di un altro tipo	§ Tipo cambiato
store_error	i sink hanno scritto, il commit no. Un errore transitorio (API giù, 5xx) si ritenta fino a tre volte; uno permanente no	verificare l'API o il disco; la run successiva rilegge la stessa finestra

Con store_error la run resta success: i dati ci sono. Non serve rilanciare a mano, ma finché il commit non riesce ogni run rilegge la stessa finestra.

Allarme consigliato: tre run consecutive della stessa pipeline con committed: false e la stessa reason. Una è rumore, tre sono una finestra che non avanza.

Procedure

Finestra bloccata (partial: la pipeline resta sulla stessa finestra)

1. Dal report, leggere rejectedReasons e la dead-letter: errore transitorio o record strutturalmente invalido (poison pill)?
2. Transitorio (un'API giù che manda in dead-letter molti record): correggere la causa e rilanciare. Il watermark avanza alla prima run success.
3. Poison pill: correggere il dato a monte. Se non si può, far scrivere gli scarti in una dead-letter durevole e rielaborabile e dichiarare graph("…", { commitOn: "partial" }): è una modifica di codice, in PR, non un'operazione.
4. Mai sbloccare con un reset in avanti senza aver messo al sicuro i record scartati: il watermark li supererebbe per sempre.

Record persi (records_lost)

La pipeline dichiara commitOn: "partial", ma in questa run uno scarto non ha conservato i record: un guard ha chiuso uno stream (circuit-breaker, assert in gate fallito, sink interrotto), oppure un nodo è stato saltato.

1. Nel report cercare i nodi con skipped e i try-catch con caught > 0: caughtErrors dice quale nodo ha fallito e perché.
2. Correggere la causa e rilanciare. Non c'è altro da fare sullo stato: non è avanzato, quindi la run successiva rilegge la finestra intera.

Istanza connettore cambiata (la run fallisce prima di estrarre)

La run si ferma con un ConfigError che nomina questo runbook quando il valore salvato ha un altro instanceId rispetto alla source.

1. Confermare che la sorgente sia davvero cambiata: altro database, altro host.
2. Stessa sorgente, altro id dell'istanza (rinominata nel registry): reset con --to al valore corrente. Il reset azzera l'instanceId, e il commit successivo lo rilega.
3. Sorgente diversa: reset senza --to, poi una run. È un full reload: vedi sotto.

Tipo cambiato (la run fallisce prima di estrarre)

La colonna di watermark è passata da intero a timestamp, o viceversa. Un reset non lo risolve: il reset conserva il tipo della chiave, anche senza --to, e la run continuerebbe a fallire.

Una chiave è (pipeline, workspace, nodeId): per cambiare tipo si cambia l'id del nodo source nel grafo (in PR). La source nuova non ha stato, parte da initial e scrive una chiave sua; quella vecchia resta nello storico, inerte. È un full reload: vedi sotto.

Backfill di un intervallo

Un backfill è una run con un override del watermark:

etl run <pipeline> --params '{"watermark":{"<nodeId>":"2026-09-01 00:00:00"}}'
etl run <pipeline> --workspace <id> --params '{"watermark":{"<nodeId>":"2026-09-01 00:00:00"}}'

Dal worker, lo stesso oggetto in input_data del processo.

0. Un override in avanti salta righe. Se il valore è oltre quello salvato, la run legge da lì, committa il massimo visto, e le righe fra i due valori non vengono mai estratte. Un backfill parte sempre da un valore più vecchio di quello salvato: controllare prima con etl state show.
1. L'override vince in lettura: la source parte da lì. Il nome del nodo dev'essere quello di una source incrementale della pipeline, altrimenti la run fallisce subito (un refuso non passa per un backfill riuscito).
2. Lo stato non arretra: al commit vale il massimo fra il valore salvato e quello visto.
3. Intervalli grandi: più run per finestre, verificando l'idempotenza (sink con chiave naturale). L'override fissa solo l'inizio: la fine la decide la query.
4. Traccia: il report della run (origin: "override") e processes.input_data.

Reset

etl state reset <pipeline> --node <id> --to <valore> --reason "<testo>" [--workspace <id>]
etl state reset <pipeline> --node <id> --reason "<testo>" [--workspace <id>]   # cancella il valore

1. --reason è obbligatorio: finisce nello storico. L'autore è --actor, altrimenti ETL_STATE_ACTOR, altrimenti l'utente del sistema operativo. È dichiarato, non autenticato: l'autorizzazione è la chiave API.
2. Senza --to il valore si cancella: la run successiva riparte da initial.
3. --to dev'essere del tipo già salvato. Una chiave mai committata non ha un tipo: si può solo resettare senza valore.
4. Il reset alza la generazione: le run in corso su quella source non committano (generation_mismatch). Meglio farlo a worker fermo o fra una run e l'altra.
5. Verificare con etl state show che l'evento sia nello storico, con motivo e autore.

Un reset sullo stato del worker (--workspace) tocca uno stato condiviso: richiede la stessa attenzione di un'operazione distruttiva, perché un reset in avanti salta record.

Full reload

Reset senza --to, poi una run. Stimare prima il volume: è un'estrazione completa della sorgente, e un sink con retry la scrive a blocchi.

Il file di stato locale è corrotto

etl run fallisce con «Stato di pipeline corrotto» o «con una forma inattesa», e nomina il file. Non riparte da initial da sola: sarebbe un full reload non richiesto.

1. Ripristinare il file da una copia, se c'è.
2. Altrimenti rimuoverlo: è un full reload voluto. Lo storico locale degli eventi si perde con lui.

Escalation

• Primo livello: chi ha in carico la pipeline.
• Secondo livello: team BI/ETL. Un sospetto di record saltati (reset in avanti, commitOn: "partial" senza dead-letter durevole) va trattato come un incidente sui dati.