Documentazione

Tema

0037. Parallelismo reale e osservabile: onNodeStart e join non serializzante ​

  • Stato: Accettata
  • Data: 2026-07-28
  • Decisori: Team BI/ETL

Contesto ​

orders-enriched-daily è la pipeline di riferimento del DSL a grafo: due sorgenti indipendenti (ordini, clienti) confluiscono in una join. Eseguendola, i nodi si accendevano uno alla volta, senza alcuna sovrapposizione — il contrario di ciò che un modello a grafo lascia intendere.

L'analisi (issue #85) ha trovato quattro cause distinte, non una:

  1. La hash join serializzava per costruzione. hashJoin faceva await buildJoinTable(build, …) prima di leggere un solo record del lato probe. In orders-enriched-daily il ramo clienti girava quindi fino a esaurimento prima che il ramo ordini emettesse il primo record: i due rami "indipendenti" non si sovrapponevano mai.
  2. Non esisteva un evento di avvio. ProgressReporter aveva il solo onNodeComplete (ADR-0018). La UI marcava running in base alla fase globale della run, non per nodo: non c'era proprio il dato con cui disegnare concorrenza.
  3. L'ordine di completamento è per forza una cascata source → sink. Il runner è pull-based e i nodi sono generator pigri: un nodo si chiude — e quindi settleNode scatta — solo quando il suo input è esaurito. onNodeComplete esce perciò in ordine quasi-topologico anche quando i nodi si interlacciano davvero.
  4. stepDelayMs peggiorava la percezione. withStepDelay inseriva uno sleep artificiale nel writeReport di ogni nodo per rendere osservabile la cascata. Ma settleNode attende writeReport dentro il percorso dati: il ritardo bloccava anche il downstream, rendendo la run più sequenziale di quanto già fosse.

Concorrenza reale ne esisteva (i sink sono guidati da un Promise.all, il fan-out usa tee con backpressure), ma in questa pipeline era quasi nulla e comunque invisibile.

Decisione ​

ProgressReporter.onNodeStart (opzionale) ​

Nuovo evento di ciclo di vita, emesso quando il nodo attraversa il primo record — emesso per source/transform/join/try-catch, consumato per i sink.

La semantica è la parte che conta. L'alternativa ovvia — notificare al primo next() — è sbagliata: il grafo è pull-based, la domanda si propaga dai sink alle source in un colpo solo, quindi "avvio = primo pull" accenderebbe tutti i nodi insieme e non direbbe nulla su chi stia davvero lavorando.

Corollario accettato: un nodo che non attraversa alcun record (input vuoto, filtro che scarta tutto) passa da pending direttamente a completato, senza mai risultare in esecuzione. È corretto — non ha mai elaborato niente.

Il metodo è opzionale sull'interfaccia: un reporter che non lo implementa continua a funzionare invariato. Come onNodeComplete, un reporter che lancia logga un warning e non abortisce la run (ADR-0018).

output_data.runningNodes e la UI ​

createProcessProgressReporter mantiene l'insieme dei nodi partiti e non ancora completati e lo scrive in output_data.runningNodes. deriveNodeRunStatus in noeva-etl-ui lo consuma con questa precedenza: nodeReports (finito) → runningNodes (sta lavorando) → fase globale (fallback per run prodotte da versioni precedenti del servizio).

Un avvio non fa avanzare progress: quella metrica conta i nodi completati su totali (ADR-0018) e deve restare monotona.

Join non serializzante ​

hashJoin (e isolatedJoinOutput nel runner) pompano il lato probe in una coda bounded mentre il lato build si materializza. Il join non può emettere prima che la tabella sia completa, ma questo non è un motivo per tenere fermo il probe nel frattempo: i due upstream ora lavorano insieme.

Il primitivo è prefetch(source, capacity) in src/core/async-queue.ts, costruito su un nuovo createBoundedQueue. Due dettagli non ovvi, entrambi costati un bug:

  • prefetch non è una async function*. Il corpo di un generatore non gira finché qualcuno non lo tira, quindi non anticiperebbe nulla: il pump deve partire alla chiamata.
  • Chiudere la coda non basta a fermare il pump. push dopo close è un no-op, quindi il ciclo continuerebbe a drenare la sorgente a vuoto — su una sorgente infinita, per sempre. Serve un flag che interrompa l'iterazione, così il for await chiama source.return() uscendo.

Rimozione di stepDelayMs ​

src/pipelines/shared/with-step-delay.ts e l'opzione OrdersEnrichedDailyOptions.stepDelayMs sono cancellati: erano una pezza alla causa 2, e con onNodeStart il segnale c'è davvero.

Alternative considerate ​

  • Pump eager generalizzato di tutte le radici del grafo (ogni source senza dipendenze produce in concorrenza dentro code bounded, con un tetto) — scartata per ora: copre anche fan-in futuri, ma cambia quando si aprono le connessioni e quando emergono gli errori, toccando insieme scope try-catch (ADR-0029) e lineage derivedFrom (ADR-0035). Il caso concreto era uno solo — la join — e intervenire lì ha un blast radius di due funzioni invece che dell'intero runner. Da riprendere se emerge un secondo caso reale.
  • Solo osservabilità (onNodeStart + rimozione di stepDelayMs), motore invariato — scartata: la UI mostrerebbe correttamente che un nodo alla volta sta lavorando. Il sintomo diventerebbe leggibile senza essere risolto.
  • Notificare l'avvio al primo next() invece che al primo record — scartata per il motivo in § Decisione: accenderebbe tutto insieme.
  • capacity come opzione di createAsyncQueue invece di una coda separata — scartata: il ramo catch di un try-catch spinge da un contesto che non può attendere, e un push che di colpo ritorna una promise da ignorare sarebbe un errore silenzioso a ogni chiamata. Due tipi distinti rendono la differenza visibile nei tipi.

Conseguenze ​

  • Positive:
    • I due rami di ingresso di una join si sovrappongono nel tempo — verificato da un test di interleaving (non di solo risultato) sia su hashJoin sia su runGraph.
    • La UI distingue un nodo che lavora da uno in attesa, senza ritardi artificiali.
    • prefetch è un primitivo riusabile: è il pezzo che servirà anche a un eventuale pump generalizzato delle radici.
  • Negative / costi:
    • Una scrittura reportProgress in più per nodo avviato, che raddoppia i round-trip verso noeva-server-api rispetto al solo onNodeComplete. ADR-0018 aveva già lasciato aperto il tema del throttling; con questo cambio diventa più concreto. Accettabile finché i grafi hanno decine di nodi, non centinaia.
    • Il lato probe viene consumato anche quando il build fallisce. Se il build solleva (es. chiave duplicata), il pump del probe ha già tirato fino a capacity record: le eventuali dead-letter dei nodi a monte del probe si materializzano dove prima non esistevano. L'errore resta lo stesso e la run aborta comunque, ma l'effetto collaterale a monte è nuovo — è il prezzo esplicito della concorrenza, non un incidente. hashJoin chiude il pump appena il build fallisce, per non lasciarlo a tirare la sorgente mentre l'errore risale.
    • Memoria: fino a DEFAULT_PREFETCH_CAPACITY (64) record del lato probe trattenuti in più durante la materializzazione del build.
  • Cosa diventa più difficile da cambiare: la semantica di onNodeStart («primo record», non «primo pull») è ora un contratto osservato dalla UI. Cambiarla cambierebbe cosa l'utente vede, non solo un dettaglio interno.

Riferimenti ​

Noeva è un marchio registrato di 4D S.R.L.

© 2026 4D S.R.L. · Gruppo4D