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 e ADR-0021.
• 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 (precarica .env)
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
NOEVA_API_KEY	sì	—	API key service-role del workspace, header Authorization: Bearer
ETL_WORKER_TYPE	no	etl	valore di processes.worker_type filtrato
ETL_POLL_INTERVAL_MS	no	2000	intervallo tra un ciclo di polling e il successivo
ETL_ENV / ETL_RELEASE / ETL_LOG_LEVEL	no	dev / null / info	propagate come RunContext alle pipeline dispatchate

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

¹ 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).

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 (vedi ADR-0021 per il razionale e i limiti).
• 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.

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 a un nome registrato in src/pipelines/registry.ts (stesso nome usato con etl run <pipeline>); se non esiste, il worker marca la riga failed con un error_message che lo dice esplicitamente, senza eseguire nulla.

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).

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), 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, §3/§5 di CLAUDE.md), 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.