Nodo: task

Un'operazione senza flusso di record: svuotare uno staging, scambiare due tabelle, un ANALYZE, il refresh di una vista materializzata. Non riceve e non emette record: quando parte lo decidono gli edge di controllo, non il flusso dei dati.

0/4

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

Metodo	.task(id, task, { after?, purpose?, retry?, rateLimit?, writeReport? })
Ingressi	0 di dati; edge di controllo da task e sink (after)
Uscite	0 di dati; edge di controllo verso task e source
Dead-letter	no: un errore del task ferma la run
Codice	Task in src/core/task.ts, gate ed esiti in src/core/control-flow.ts

Nella UI. L'app disegna il task, gli edge di controllo e i nodi saltati: una pipeline registrata può usarli. Il motore porta skipped già sull'evento di nodo concluso; finché la UI non lo legge da lì, durante una run seguita dal vivo un nodo saltato appare in errore fino al report finale.

Quando usarlo

Per ciò che un ETL fa intorno al caricamento e che non è un record che scorre:

• svuotare lo staging prima di leggere (pre-task);
• scambiare lo staging con la tabella in uso dopo aver caricato (post-task): è il write-then-swap;
• ANALYZE, refresh di una vista materializzata, una stored procedure di chiusura.

Se l'operazione lavora su ogni record è un action o un sink. Se serve solo un ordine fra due rami di dati, per esempio le dimensioni prima dei fatti, non serve un task: basta un edge di controllo da un sink a una source.

Il contratto

interface Task {
  readonly name: string;
  run(ctx: RunContext): Promise<void>;
  readonly descriptor?: ConnectorDescriptor; // su che cosa opera, per il DAG
  readonly retrySafe?: boolean; // rieseguirlo non cambia il risultato
  readonly writeReport?: WriteReportFn; // i dettagli nel report, es. le righe toccate
}

ctx.signal è il segnale della run e va rispettato: quando la run viene annullata il runner aspetta che run termini, senza timeout, perché un'operazione abbandonata a metà lascia lo stato del sistema esterno ignoto.

Task pronti

Funzione	Sistema	Cosa fa	Scheda
postgresSqlTask	PostgreSQL	Statement statici in una transazione	postgres
mssqlSqlTask	SQL Server	Statement statici in una transazione	sqlserver
mysqlSqlTask	MySQL, MariaDB	Statement statici in una transazione (il DDL fa commit implicito)	mysql
excelTask	Excel	Operazioni sul workbook: fogli e colonne (riordina, rinomina, cancella, inserisci)	excel

Un task SQL con destructive: true cancella o sostituisce dati: prima di usarlo in produzione leggi il runbook. Lo stesso vale per excelTask con deleteSheet o deleteColumns: runbook delle operazioni distruttive di Excel.

Edge di controllo: after

Un edge di controllo dice «B parte solo dopo che A è riuscito». Non porta record. Si scrive con l'opzione after dei due soli nodi che possono aspettare, .task() e .source():

graph("ordini-swap")
  .task("svuota-staging", svuota)
  .source("ordini", ordini, { after: ["svuota-staging"] }) // pre-task
  .transform("ordini", "contratto", validate("contratto", ordineSchema))
  .sink("contratto", staging) // l'id del nodo è staging.name: "carica-staging"
  .task("scambia", scambia, { after: ["carica-staging"] }) // post-task
  .build();

Da → a	Si usa per
task → source	pre-task: finisce prima che la source estragga il primo record
sink → task	post-task: parte dopo che il sink ha finito di scrivere
sink → source	un ramo prima di un altro: le dimensioni prima dei fatti
task → task	una catena di operazioni

• Parte da un nodo che finisce (task, sink) e arriva a un nodo che inizia (task, source). Mai da o verso un nodo in mezzo al flusso: build() lo rifiuta e indica l'alternativa.
• Più predecessori valgono in AND: after: ["a", "b"] aspetta entrambi.
• Solo sul successo: non esiste un edge «se fallisce». Per gestire gli errori c'è il try-catch.
• Un task senza after parte all'avvio della run. Restano obbligatori almeno una source e un sink: un grafo di soli task non si costruisce.
• Un lookup a valle di una source con after parte anch'esso dopo: sink(dimensioni) → source(fatti) copre anche «il lookup dei fatti legge le dimensioni appena caricate».

Quando il predecessore è «riuscito»

A non fa partire i suoi successori se:

• è fallito, anche se l'errore è stato catturato da un try-catch (un sink nello scope di un guard);
• oppure il suo input è stato troncato: un guard ha catturato l'errore di un nodo a monte e ne ha chiuso lo stream. Conta l'evento, non i record mancanti: vale anche se l'errore arriva dopo l'ultimo record, come il verdetto di un assert in streaming. Il sink ha scritto, e il suo load si è chiuso senza errori, ma lo scambio non deve partire su dati che un controllo ha rifiutato;
• oppure è stato saltato.

Non bloccano i rifiuti per record: la deadLetter di un nodo e i record che un guard per-record manda sul ramo catch. Sono un percorso previsto, il resto dei record è arrivato, e la run è partial.

«Nessuno scarto, altrimenti niente scambio». Un assert prima del sink non basta: vede solo i record che gli arrivano, non quelli già finiti in dead-letter a monte. Per la regola rigida togli deadLetter e guard per-record a monte del sink dello staging: il primo record non valido fa fallire la run, e lo scambio non parte.

Nodi saltati

Un nodo che non parte perché un predecessore di controllo non è riuscito è saltato. Nel report ha skipped: { because }, success: false e startedAt: null.

• Una source saltata porta con sé i suoi discendenti di dati: transform, join, sink. Un sink saltato non chiama mai load, così un sink di file non riscrive il file vuoto. Sono saltati anche i nodi a monte che alimentano soltanto nodi saltati, perché nessuno li tirerebbe.
• Un sink alimentato da più source scrive solo quando i gate di tutte sono aperti: con union(S1, S2 con after), se S2 è saltata il sink non scrive nemmeno la parte di S1.
• because è l'id del predecessore di controllo che non è riuscito; i discendenti di una source saltata portano lo stesso because della source.
• Il ramo catch di un guard che sta nel sotto-grafo saltato si chiude con zero errori: il suo sink non è saltato.

Errori e annullamento

Caso	Esito
Il task fallisce	run failed; i successori di controllo e i loro discendenti di dati sono saltati
Un successore è saltato per un errore catturato a monte	run partial
La run viene annullata a metà task	il runner aspetta run; il task ha startedAt valorizzato e success: false, non è saltato
Un altro nodo fa fallire la run	i gate si chiudono: nessun task parte dopo che la run è fallita

Un task non sta mai nello scope di un try-catch: lo scope segue gli edge di dati, e gli edge di controllo non lo estendono. Non esiste un task «opzionale»: un ANALYZE che fallisce rende failed una run con i dati già caricati.

Fail-fast fra rami. Al primo errore fatale il runner annulla la run e aspetta che tutti i sink e i task chiudano, poi riporta la run failed. Vale per ogni grafo, anche senza task: un ramo indipendente non continua a scrivere dopo che la run è fallita.

Retry e rate limit

Come sul sink: retry e rateLimit sono ammessi solo se il task dichiara retrySafe: true, altrimenti la costruzione del grafo fallisce. Si ritenta solo un errore transitorio, e l'attesa si interrompe con l'annullamento della run. Un TRUNCATE o un ANALYZE sono ripetibili; uno scambio per rename no.

.task("analizza", analizza, { after: ["scambia"], retry: { attempts: 3 } }) // solo se retrySafe

Le regole che build() fa rispettare

• un edge di controllo parte da un task o da un sink e arriva a un task o a una source;
• niente cicli sull'unione di edge di dati e di controllo: sink → source sulla source che alimenta quel sink è un ciclo;
• niente deadlock senza ciclo: A ⇢ B è rifiutato se un join o un union a valle di B ha un antenato di dati in comune con A. Il fan-in aspetterebbe B, il suo ramo dall'antenato comune si riempirebbe, e A non finirebbe mai. Il messaggio nomina l'antenato: si risolve leggendolo con una source per ramo;
• un task non ha edge di dati, in nessuna direzione;
• retry e rateLimit solo su un task retrySafe.

Nel DAG e nel report

Nel piano (getSpec()) un task ha stepType: "task", columns: [] e il suo descriptor. Gli edge di controllo stanno in edges con kind: "control", id <from>-><to>:control, senza porte.

Nel report il task è un nodo con kind: "task": extracted, written e rejected restano a 0, e i tempi sono quelli di run. I dettagli (le righe toccate da un task SQL) vanno in nodeReports, dal writeReport del nodo o da quello dichiarato dal task. retries compare con una policy di retry. Gli edges del report restano quelli di dati. onNodeStart arriva all'ingresso in run, onNodeComplete all'uscita; un nodo saltato notifica solo il completamento, così progress arriva a 100, e il completamento porta già skipped: { because }.

Anti-pattern

• Un TRUNCATE dentro il load di un sink, o in run.ts prima della pipeline: il DAG non lo mostra e il report non lo misura. È un pre-task.
• TRUNCATE sulla tabella in uso seguito dal caricamento: fra i due la tabella è vuota. Si carica uno staging e lo si scambia.
• Uno scambio con retry: ripeterlo riporta le tabelle com'erano. Non è retrySafe.
• Statement SQL composti con valori della run: i valori entrano solo come parametri bind.
• Un post-task come unica garanzia di qualità con deadLetter a monte: gli scarti non lo fermano.