Runbook: worker etl worker (polling su processes)

Sintesi

• Cosa fa: processo daemon (non batch) che resta in polling sulla tabella processes, rivendicando le righe con status="pending", worker_type=<ETL_WORKER_TYPE> e workspace_id=<NOEVA_WORKSPACE_ID> tramite POST /api/processes/etl/claim su noeva-server-api (claim CAS + filtro fatti server-side — il worker non ha mai un client Supabase diretto, così non porta chiavi Supabase su infrastruttura potenzialmente del cliente dove è deployato), e per ciascuna esegue la pipeline il cui nome è nel campo pipeline, scrivendo l'esito (completed/failed + output_data/ error_message) via PATCH /api/processes/etl/:processId/progress.
• Owner: team ETL.
• Schedulazione: non è schedulato — è un processo long-running, avviato e tenuto vivo dall'orchestrazione infra (PM2/container/ECS), non da un cron. Una singola istanza può servire più workspace dello stesso cliente (un poller per workspace) impostando NOEVA_WORKSPACE_IDS — vedi § Multi-workspace.
• Criticità: media. Un'istanza ferma significa che i processi ETL di quel workspace restano pending a tempo indeterminato (nessun altro worker li rivendica), ma non causa perdita di dati né corruzione: al riavvio il polling riprende e li rivendica normalmente.
• Scheda pipeline: ogni pipeline dispatchata da qui ha la propria scheda in docs/pipelines/; questo runbook copre solo il meccanismo di claim/dispatch, non la logica delle singole pipeline.

Esecuzione normale

# in sviluppo (la CLI carica il .env della radice del repo)
pnpm run etl worker

# equivalente, CLI compilata
node dist/cli.js worker

Variabili d'ambiente (vedi .env.example):

Variabile	Obbligatoria	Default	Significato
NOEVA_WORKSPACE_ID	sì¹	—	workspace di cui questa istanza rivendica i processi (monoworkspace)
NOEVA_WORKSPACE_IDS	no	—	lista (separata da virgole) di workspace; se presente ha la precedenza su NOEVA_WORKSPACE_ID (§ Multi-workspace)
NOEVA_API_BASE_URL	sì	—	base URL di noeva-server-api per claim/progress dei processi ETL; senza credenziali nell'URL: se ci sono, nei log compaiono come ***@
NOEVA_API_KEY	sì	—	API key service-role del workspace, header Authorization: Bearer
ETL_PIPELINE_WORKSPACES	no	— (tutte)	pipeline ammesse per workspace (§ Pipeline per workspace)
ETL_WORKER_TYPE	no	etl	valore di processes.worker_type filtrato
ETL_POLL_INTERVAL_MS	no	60000	intervallo tra un ciclo di polling e il successivo
ETL_ENV / ETL_RELEASE / ETL_LOG_LEVEL	no	dev / commit / info	propagate come RunContext alle pipeline dispatchate; il commit è quello del repo dell'etl.config.ts

Nessuna credenziale Supabase: il worker non ha mai un client Supabase diretto, solo le credenziali NOEVA_ verso noeva-server-api.

¹ NOEVA_WORKSPACE_ID è obbligatoria solo nel caso monoworkspace: se è impostata NOEVA_WORKSPACE_IDS, i workspace arrivano da quella lista e NOEVA_WORKSPACE_ID non serve (l'NOEVA_API_KEY resta comunque necessaria e condivisa).

Config fail-fast: se non c'è né NOEVA_WORKSPACE_ID né NOEVA_WORKSPACE_IDS, se manca NOEVA_API_BASE_URL/NOEVA_API_KEY, o se ETL_POLL_INTERVAL_MS non è un intero positivo, il processo termina subito con un ConfigError (non parte in uno stato ambiguo).

Stato di pipeline

Le pipeline con source incrementali tengono i loro watermark su noeva-server-api (/api/processes/etl/state/*), con le stesse NOEVA_API_BASE_URL e NOEVA_API_KEY dei processi e uno stato per workspace: il container non ha storage durevole, e non serve montare ETL_STATE_DIR. Il proprietario della chiave dev'essere membro del workspace. Il report di ogni run, in processes.output_data, porta l'esito in state.<nodeId>.

Leggere lo stato, backfill e reset: runbook dello stato incrementale. Da una shell, lo stato del worker si tocca solo con --workspace <id>.

Multi-workspace

Una singola istanza può servire più workspace dello stesso cliente impostando NOEVA_WORKSPACE_IDS alla lista dei workspace, separati da virgole:

NOEVA_API_BASE_URL=https://api.noeva.ai
NOEVA_API_KEY=<service-role-key>              # UNA sola, per tutti i workspace
NOEVA_WORKSPACE_IDS=<uuid-a>,<uuid-b>

• L'NOEVA_API_KEY è una sola e vale per tutti i workspace: cambia solo l'id inviato nell'header x-noeva-workspace. Nessun secret per workspace.
• baseUrl, workerType, pollIntervalMs sono condivisi (da NOEVA_API_BASE_URL, ETL_WORKER_TYPE, ETL_POLL_INTERVAL_MS).
• L'istanza avvia un poller indipendente per workspace; ognuno rivendica con lo stesso key e il proprio header. Un job lento su un workspace non blocca gli altri.
• I connettori sorgente/destinazione (PG/MSSQL/…) restano su env globale condiviso.
• Log: ogni poller emette component: "etl-worker" con binding workspaceId distinto — filtrare per workspaceId per isolare l'attività di un workspace.

Senza NOEVA_WORKSPACE_IDS, il comportamento resta monoworkspace (da NOEVA_*), identico alle versioni precedenti.

Pipeline per workspace

Un deploy conosce un solo cliente: le pipeline che può vedere ed eseguire sono quelle del catalogo del suo etl.config.ts, e il perimetro fra clienti è il repo. Quando la stessa istanza serve più workspace del cliente (NOEVA_WORKSPACE_IDS), ETL_PIPELINE_WORKSPACES restringe quali pipeline del catalogo vede ed esegue ognuno: per esempio una pipeline che gira solo nel workspace di produzione e non in quello di prova.

# una pipeline, un workspace
ETL_PIPELINE_WORKSPACES=orders-enriched-daily:<uuid-produzione>
# più voci separate da virgole, più workspace per voce separati da |
ETL_PIPELINE_WORKSPACES=crm-leads-sync:<uuid-a>|<uuid-b>,orders-enriched-daily:<uuid-a>

• Non impostata: tutte le pipeline del catalogo, in ogni workspace servito. È il caso normale, e con un solo workspace non c'è motivo di impostarla.
• Impostata: una pipeline del catalogo non elencata è nascosta in tutti i workspace dell'istanza.
• etl serve espone su GET /graphs solo le pipeline dei workspace serviti, e il worker annuncia a noeva-server solo quelle del proprio workspace.
• Il worker chiude come failed, senza eseguirlo, un processo la cui pipeline non è assegnata al suo workspace. L'error_message cita pipeline, workspace e variabile.
• Un nome che non è nel catalogo, una voce senza : o senza workspace e una pipeline ripetuta fermano l'avvio con exit 2 e un errore che elenca le pipeline del catalogo.
• Il nome di una sub-pipeline ferma l'avvio col messaggio dedicato: «"<nome>" è una sub-pipeline, si istanzia da un nodo pipeline». Una sub-pipeline non si lancia, quindi non si assegna: è visibile dove lo è una pipeline che la istanzia.
• All'avvio, su stderr, una riga per workspace elenca le pipeline ammesse:

Pipeline per workspace (ETL_PIPELINE_WORKSPACES):
  <uuid-produzione>: orders-enriched-daily

etl list, etl plan ed etl run non applicano il filtro. Le sub-pipeline non sono nel catalogo del worker né in etl list; etl run <sub-pipeline> esce con codice 2 e lo stesso messaggio dedicato.

Inserire un processo di test (da SQL diretto su Supabase, o da uno script):

INSERT INTO processes (workspace_id, worker_type, function, input_data, status)
VALUES ('<workspace-id-uuid>', 'etl', 'orders-enriched-daily', '{}'::jsonb, 'pending');

function deve corrispondere al nome di una pipeline del catalogo di etl.config.ts (stesso nome usato con etl run <pipeline>); se non c'è, o se ETL_PIPELINE_WORKSPACES non la assegna al workspace della riga, il worker marca la riga failed con un error_message che lo dice esplicitamente, senza eseguire nulla. Lo stesso vale per una riga con la chiave trace in input_data: il worker non esegue run tracciate, che restano locali (etl run --trace o l'app desktop su etl serve).

Se la pipeline dichiara parametri di lancio, input_data deve rispettarli: un obbligatorio mancante o un valore del tipo sbagliato chiude la riga failed prima di qualunque nodo, con un error_message che nomina ogni parametro (Parametri di lancio non validi per "<pipeline>": …). Il worker annuncia la dichiarazione all'API insieme al catalogo, all'avvio: l'API la usa per rifiutare con 400 un lancio dalla UI con parametri sbagliati. Dopo un rilascio che cambia i parametri di una pipeline, la validazione dell'API si allinea al primo riavvio del worker.

Output atteso: log umani su stderr (uno per claim, uno per completamento/ fallimento del dispatch); l'esito del processo si legge dalla riga stessa in processes, non da stdout (a differenza di etl run, qui non c'è un report su stdout stampato — l'unico "risultato" persistito è la riga DB aggiornata).

Arresto e deploy

SIGTERM, SIGINT e il messaggio shutdown di PM2 fermano il worker in modo cooperativo (sotto PM2 è il messaggio, perché ecosystem() del pacchetto dichiara shutdown_with_message: true: su Windows i segnali non esistono), e annullano la run in corso invece di aspettarla:

• l'annullamento arriva ai nodi su ctx.signal: un'attesa (un backoff di retry, il sleep di un connettore o di un task) si ferma subito, e il guard non conta l'interruzione fra gli errori catturati;
• il processo chiude failed, con error_message che inizia per Run interrotta dall'arresto del worker: seguito dall'errore del nodo interrotto, e con il report parziale in output_data;
• una run che riesce a finire bene mentre arriva l'arresto resta completed.

Un passo che il segnale non interrompe (un commit già partito, un DDL, una query il cui driver non accetta il segnale) viene comunque atteso. Il tempo concesso dal supervisore deve coprire quel passo più la chiamata finale a noeva-server-api: kill_timeout in ecosystem() (10 s), e docker stop -t / stopTimeout di ECS per il container (10 s il default di Docker). Oltre quel tempo il processo viene ucciso, e la riga resta running (vedi § Run fallita a metà).

Le run interrotte da un deploy non tornano in coda da sole: rilanciarle è una decisione di chi opera, perché la run può aver già prodotto effetti esterni (una mail mandata). Per trovarle e rimetterle in coda:

SELECT id, function, error_message, updated_at
FROM processes
WHERE workspace_id = '<workspace-id-uuid>' AND worker_type = 'etl' AND status = 'failed'
  AND error_message LIKE 'Run interrotta dall''arresto del worker%'
ORDER BY updated_at DESC;

UPDATE processes SET status = 'pending', progress = 0, error_message = NULL
WHERE id = '<process-id>' AND status = 'failed';

Come distribuire un aggiornamento: Deploy.

Monitoraggio

• Query di stato rapido:

  SELECT id, function, status, progress, error_message, updated_at
  FROM processes
  WHERE workspace_id = '<workspace-id-uuid>' AND worker_type = 'etl'
  ORDER BY updated_at DESC
  LIMIT 20;

• Un processo bloccato a lungo su running con updated_at non recente indica o un'istanza worker ferma/crashata, o un dispatch che sta ancora eseguendo una pipeline molto lunga (qui non c'è timeout applicativo — a differenza del worker principale noeva-server-worker, questo poller non forza un limite di durata sul singolo dispatch).
• Log applicativi: cercare le righe component: "etl-worker" (bindings del logger) per isolare l'attività di questa istanza da altri log del processo.

Procedure di recovery

Run fallita a metà (pipeline lanciata, poi crash del processo worker)

Se l'intero processo Node muore durante un dispatch (non un errore applicativo gestito, ma un crash, o un arresto che ha superato kill_timeout — vedi § Arresto e deploy), la riga resta running indefinitamente — nessun altro worker la rivendica finché non torna pending. Verificare che la pipeline non abbia effettivamente scritto dati parziali (le pipeline devono essere idempotenti), poi:

UPDATE processes SET status = 'pending', progress = 0
WHERE id = '<process-id>' AND status = 'running';

Riavviare il worker (o attendere che l'istanza esistente la rivendichi al prossimo ciclo, se il crash era transitorio e il processo Node è ancora vivo).

Troppi record in failed

SELECT function, error_message, count(*)
FROM processes
WHERE workspace_id = '<workspace-id-uuid>' AND worker_type = 'etl' AND status = 'failed'
GROUP BY function, error_message
ORDER BY count(*) DESC;

Se il pattern è lo stesso error_message ripetuto per una function sconosciuta, è quasi sempre un refuso nel nome pipeline lato produttore dell'INSERT, non un bug del worker. Correggere alla fonte e re-inserire (o UPDATE processes SET status='pending', function='<nome-corretto>' WHERE id=...).

Backfill / rilancio massivo

Non c'è un comando dedicato: per rilanciare N processi falliti, UPDATE processes SET status='pending', progress=0, error_message=NULL WHERE workspace_id=... AND worker_type='etl' AND status='failed' AND function='<nome>' li rimette in coda; il worker li rivendica al prossimo ciclo di polling nell'ordine di created_at (FIFO), senza bisogno di riavviarlo.

Escalation

Se il worker non rivendica nulla nonostante righe pending visibili con workspace_id/worker_type corretti: verificare prima che l'istanza sia effettivamente in esecuzione (non solo che il container esista) e che NOEVA_WORKSPACE_ID/ETL_WORKER_TYPE nel suo ambiente combacino esattamente coi valori nella tabella (case-sensitive, nessuno spazio). Se il problema persiste, coinvolgere il team ETL con: query di stato (sopra), log recenti dell'istanza, e l'id di un processo pending non rivendicato da >5 minuti.