Connettore: processo esterno

Lancia un programma sulla macchina del worker, una volta per record, dentro un nodo action: il record porta dei file, il programma li elabora e restituisce altri file più un riepilogo JSON su stdout. Serve a riusare un programma che fa già bene il suo lavoro, per esempio uno script di elaborazione già validato dal cliente, senza riscriverne la logica. Codice in src/connectors/process/, subpath @noeva/etl-core/connectors/process.

Funzione	Tipo	Cosa fa
processoEsternoAction	Action<I, O>	Lancia il programma sui file del record e prosegue con riepilogo e file

Non è un tipo di nodo: nel DAG è un nodo action con descriptor processo-esterno, e ne eredita isolamento per record, dead-letter, ramo catch e lineage.

Uso

import { z } from "zod";
import { processoEsternoAction } from "../../connectors/process/index.js";

const Riepilogo = z.object({ mese: z.number(), avvisi: z.array(z.string()) });

const elabora = processoEsternoAction({
  name: "elabora-export",
  eseguibile: "/opt/elabora/bin/elabora", // meglio un percorso assoluto
  argomenti: ["--json"], // fissi: compaiono nel piano
  versione: "v1.4.0", // compare nel piano
  timeoutMs: 120_000, // obbligatorio
  ambiente: ["LANG"], // le sole variabili del worker che il processo vede
  ingresso: (r: Export) => ({
    file: [{ nome: "ingresso/export.xlsx", bytes: r.bytes }],
    argomenti: ["ingresso/export.xlsx"], // ognuno è UN elemento di argv
  }),
  riepilogo: Riepilogo,
  merge: (r, esito) => ({
    ...r,
    riepilogo: esito.riepilogo,
    prodotti: esito.file,
  }),
});

graph("report-mensili")
  .source("export", exportSource)
  .action("export", "elabora", elabora, { deadLetter: scarti })
  .sink("elabora", caricamento)
  .build();

La cartella del record

Ogni record ha una cartella temporanea nuova, che è la working directory del programma:

noeva-processo-XXXXXX/      ← cwd del programma
├── ingresso/export.xlsx    ← i file del record, nei percorsi scelti dalla pipeline
└── uscita/                 ← già creata: quello che il programma ci scrive torna indietro
    └── 2026/07/report.txt

• I file d'ingresso hanno un nome relativo alla cartella del record, con / come separatore. Un nome che ne uscirebbe (../, un percorso assoluto, \, segmenti vuoti) è un errore del record: il nome arriva dai dati, e non si normalizza.
• I file prodotti sono quelli in uscita/, sottocartelle comprese, con il nome relativo a uscita/ e in ordine di nome. Solo file regolari: i link simbolici non si seguono.
• Un file d'ingresso messo sotto uscita/ torna indietro anche se il programma non lo tocca: è il modo di far aggiornare un file sul posto, come un workbook a cui il programma aggiunge un foglio.
• La cartella si cancella sempre, anche quando il record fallisce. Se la cancellazione non riesce, un warn processo-esterno.cartella-non-cancellata lo dice nel log.

cartellaBase sceglie dove nascono le cartelle; il default è la cartella temporanea del sistema.

Esecuzione

• Senza shell. Il programma parte con spawn e gli argomenti come array: un argomento con ; rm, $(…) o spazi arriva al programma tale e quale. I dati del record entrano solo come contenuto di un file o come singolo elemento di argv.
• Ambiente da allowlist. Il processo riceve solo le variabili elencate in ambiente, con il valore che hanno nel worker; una variabile elencata ma non impostata non si passa. Senza ambiente non riceve nulla, nemmeno PATH: un eseguibile dato per nome si trova solo con PATH in elenco, quindi meglio un percorso assoluto.
• Timeout obbligatorio. Allo scadere di timeoutMs, o se la run viene annullata (ctx.signal), il processo riceve SIGTERM, e SIGKILL se dopo terminazioneMs (default 5000) non è ancora uscito.
• Un gruppo di processi per lancio. Il programma parte in un gruppo suo, e i segnali vanno al gruppo intero: un sotto-processo, come l'interprete lanciato da uno script di avvio, terrebbe aperti stdout e stderr e la fine del lancio non arriverebbe mai. Su Windows, dove i gruppi non ci sono, si termina il solo figlio. Il gruppo non sopravvive a un timeout o a un annullamento, ma sopravvive a un worker che muore di colpo: ripulirlo è compito di chi riavvia il worker.
• stdout è il riepilogo: un unico documento JSON, validato con lo schema riepilogo. Al massimo MAX_STDOUT_CARATTERI (un milione): i dati escono come file, non su stdout.

Errori

Questi casi sono errori del record (ActionError), che il nodo action instrada in dead-letter, sul ramo catch o in fail-fast:

Caso	Messaggio
Exit code diverso da zero	… è terminato con exit code N
Terminato da un segnale non nostro	… è stato terminato dal segnale SIGSEGV
Timeout	… ha superato il timeout di N ms ed è stato terminato
Run annullata	… è stato terminato: run annullata
stdout che non è JSON	… non ha scritto su stdout un documento JSON
Riepilogo che non passa lo schema	… ha scritto un riepilogo non conforme allo schema
stdout oltre il tetto	… ha scritto su stdout più di … caratteri
Eseguibile che non parte	Impossibile avviare il processo …
Nome di file d'ingresso fuori cartella	nome di file d'ingresso non ammesso: …

I details dell'errore portano exitCode, segnale e stderr: la coda di stderr, gli ultimi CODA_STDERR_CARATTERI (2000) caratteri, perché stderr può contenere dati del record. Lo stdout non entra mai nell'errore; di un riepilogo non conforme si dicono i campi che non tornano, non i loro valori.

Una configurazione sbagliata (eseguibile vuoto, timeout non intero positivo, nome di variabile non valido) è un ConfigError alla costruzione dell'action, non al primo record.

Nel piano

Il descriptor (kind: "processo-esterno") mostra eseguibile, argomenti fissi, versione se c'è, timeout e i nomi delle variabili in allowlist, mai i loro valori. Gli argomenti fissi stanno nel piano e nella lista dei processi del sistema: mai segreti lì. Un segreto che il programma deve conoscere passa per una variabile in allowlist.

Cambiare versione, eseguibile o argomenti cambia il piano, e quindi chiede una versione nuova della pipeline.

Che cosa il connettore non fa

• Nessuna sandbox. Il programma gira con i permessi del worker: si lanciano solo programmi nostri e versionati, mai qualcosa che arriva dai dati.
• L'idempotenza resta del chiamante. Il contratto chiede che il programma lavori solo dentro la cartella che riceve, senza effetti esterni; il connettore lo dichiara ma non lo può imporre.
• Una scatola nera nel DAG. Il report del nodo conta un record per lancio: quello che succede dentro si legge dal riepilogo.
• L'eseguibile deve esserci dove gira il worker. È un problema del deploy, non del connettore.

Test

src/connectors/process/action.test.ts lancia un programma Node vero ma finto, scritto su disco dal test: andata e ritorno dei file, riepilogo validato, JSON non valido, exit code con la coda di stderr, timeout, SIGKILL dopo un SIGTERM ignorato, sotto-processi terminati con il gruppo, annullamento, cartella cancellata in ogni esito, argomenti senza shell, ambiente limitato all'allowlist, e un record che fallisce dentro un grafo e finisce nel dead-letter del nodo. Nessun test richiede un interprete diverso da Node. descriptor.test.ts verifica i campi del piano e che i valori delle variabili in allowlist non ci finiscano mai.