Nodo: sink

Il punto d'uscita: carica i record su un sistema di destinazione. Ogni grafo ne ha almeno uno, e ogni ramo finisce in un sink. Il caricamento deve essere idempotente: rieseguire la run non duplica dati.

0/3

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

Metodo	.sink(from, sink, { fromPort?, purpose?, retry?, rateLimit?, writeReport? })
Ingressi	1
Uscite	0
Dead-letter	no; isolabile per record sotto un try-catch se implementa loadOne
Codice	Sink<T> in src/core/connector.ts, retry in src/core/sink-retry.ts

⚠️ .sink() non prende un id: l'id del nodo è sink.name. Due sink con lo stesso name collidono con «Id nodo duplicato».

Quando usarlo

Per ogni scrittura che è l'ultima cosa che succede al record: una tabella, un file, un workspace Noeva, una mail di notifica, e le dead-letter. Se a valle della scrittura serve il suo risultato, è un'action.

Il contratto

interface Sink<T> {
  readonly name: string; // è anche l'id del nodo
  load(
    records: AsyncIterable<T>,
    ctx: RunContext,
  ): Promise<{
    written: number;
    // solo i sink in modalità SCD2
    versions?: {
      opened: number;
      closed: number;
      corrected: number;
      unchanged: number;
    };
  }>;
  loadOne?(record: T, ctx: RunContext): Promise<void>; // isolabile per record sotto un guard
  readonly retrySafe?: boolean; // regge retry e scrittura a blocchi
  readonly held?: (ctx: RunContext) => number; // righe nel batch non ancora scritte
  readonly descriptor?: ConnectorDescriptor;
  readonly columns?: readonly PlanColumn[];
}

Sink pronti

Funzione	Sistema	retrySafe	loadOne	Scheda
postgresSink	PostgreSQL, upsert ON CONFLICT	✅	—	postgres
postgresSink con mode: "scd2"	PostgreSQL, dimensione storicizzata	✅	—	postgres
mssqlSink	SQL Server, MERGE	✅	—	sqlserver
mysqlSink	MySQL, MariaDB, ON DUPLICATE KEY UPDATE	✅	—	mysql
mssqlSink con mode: "scd2"	SQL Server, dimensione storicizzata	✅	—	sqlserver
fileSink	File (storage × codec)	—	—	file
excelSink con mode: "upsert"	Excel, aggiorna per chiave	✅	—	excel
excelSink con mode: "append" o "replace"	Excel, aggiunge o sostituisce	—	—	excel
noevaFileSink, noevaFilesSink	Workspace Noeva	—	noevaFilesSink	noeva
smtpSink	Una mail per record	—	✅	smtp
mailSink	smtpSink o sendgridSink, dalla configurazione	—	✅	sendgrid
sendgridSink	Una mail per record via SendGrid	—	✅	sendgrid
arraySink(name)	In memoria: test	—	✅	—

Esempi

.sink("contratto-fatto", sinks.fatti)
.sink("smista", sinks.esteri, { fromPort: "esteri" }) // da una porta di un route
.sink("contratto-fatto", dwhSink, {
  retry: { attempts: 4, batchSize: 500 }, // solo su un sink retrySafe
  rateLimit: { perSecond: 5 },
})

Una dead-letter è un sink come gli altri, di tipo Sink<DeadLetter>: si passa a { deadLetter } di un nodo, oppure si collega al ramo catch di un try-catch.

Dimensioni storicizzate: mode: "scd2"

I sink Postgres e SQL Server hanno una seconda modalità, per le dimensioni che devono conservare la storia (SCD tipo 2): quando un'entità cambia, il sink chiude la versione corrente e ne apre una nuova invece di sovrascrivere. È lo stesso nodo nel DAG, con mode fra i dettagli.

.transform("clienti", "chiave", surrogateKey("chiave", {
  from: ["codice"], entity: "cliente", as: "cliente_sk",
}))
.transform("chiave", "hash", hashRow("hash", {
  columns: ["nome", "citta"], as: "riga_hash",
}))
.sink("hash", postgresSink({
  pool,
  table: "dwh.dim_cliente",
  mode: "scd2",
  columns: ["codice", "cliente_sk", "nome", "citta", "riga_hash"],
  naturalKey: ["codice"],
  hashColumn: "riga_hash",
  validFrom: "valid_from",
  validTo: "valid_to",
  versionKey: { column: "versione_sk", entity: "cliente" },
  executionIdColumn: "execution_id",
}))

• La run deve passare referenceDate (YYYY-MM-DD) fra i parametri: è l'inizio di validità delle versioni aperte. Non c'è fallback all'orologio, quindi rieseguire la stessa data dà lo stesso risultato.
• conflictTarget insieme a mode: "scd2" non compila.
• A monte servono surrogateKey (chiave durevole) e hashRow (hash di riga): il sink confronta l'hash, non lo calcola.
• Il sink è retrySafe ma non ha loadOne.
• scd2Rehash: true fra i parametri trasforma la run in un rehash.

Opzioni, DDL di riferimento, errori e i due avvisi che il sink non può verificare: schede postgres e sqlserver.

Retry e rate limit

Solo su un sink che dichiara retrySafe: true (oggi i sink Postgres, SQL Server e MySQL, e excelSink in upsert): su uno che riscrive un file intero o manda messaggi build() rifiuta la policy, perché ripetere perderebbe o duplicherebbe dati.

• Con una policy il runner scrive a blocchi di batchSize (default 1000) e ritenta il blocco intero sugli errori transitori (deadlock, timeout, rete).
• Default del retry: attempts: 3 (primo tentativo compreso), baseMs: 200, maxMs: 5000, jitter: 0.5.
• L'atomicità diventa per blocco: dopo un errore definitivo i blocchi già confermati restano scritti. È sicuro perché il sink è idempotente.

Errori

load è un contratto opaco: non sa dire quale record del batch ha fallito. Fuori da un guard, un LoadError ferma la run. Dentro lo scope di un try-catch in modalità per-record, un sink con loadOne viene chiamato record per record e un fallimento arriva al ramo catch con il record in raw; senza loadOne arriva al catch senza raw.

Nel DAG e nel report

Il nodo mostra il descriptor (destinazione, mai segreti). Le colonne arrivano dal nodo a monte, o da columns se il sink le dichiara. Nel report, written conta i record scritti e retries i tentativi ripetuti. Un sink in modalità SCD2 riporta anche versions, con le versioni aperte, chiuse, corrette e invariate: lì written è opened + corrected, quindi una run con tutto invariato ha written: 0.

Un sink può risultare saltato (skipped: { because } nel report, success: false). In quel caso load non viene chiamata: un sink che riscrive un file intero lo riscriverebbe vuoto. Succede in due casi:

• non gli arriva nessun record perché un nodo a monte, dentro un try-catch, è fallito senza aver emesso niente. Lo stream si è chiuso pulito, ma non è un dataset vuoto. Un input vuoto di suo resta un load su zero record, e se il nodo a monte fallisce dopo aver emesso dei record il sink li ha già ricevuti e li scrive;
• una delle source a monte ha un edge di controllo (after) e il suo predecessore non è riuscito: la source è saltata, e con lei tutto ciò che ne discende.

Dal vivo

Nei contatori dal vivo gli usciti di un sink salgono quando il sink dice di aver scritto, e l'in lavorazione dice quante righe aspettano di esserlo. C'è in due casi:

• con retry o rateLimit: il blocco del runner, sia mentre si riempie sia mentre il load lo scrive, retry compresi; sotto un guard per-record, il record di un loadOne in volo;
• senza policy, su un sink che dichiara held: le righe accumulate nel suo batch e non ancora scritte. Lo dichiarano i sink Postgres, SQL Server e MySQL, anche in SCD2 dove c'è. Il batch conta finché lo statement non risponde. Le righe già scritte aspettano il commit a fine load: non sono più in lavorazione, e non sono ancora uscite.

Con una policy il blocco del runner comprende già il batch del sink, quindi il runner conta il proprio e non somma l'altro. Un sink che non accumula (arraySink, un file riscritto intero) non ha il campo.

Un sink custom che accumula lo dichiara come una transform: const held = heldCounter() nella factory, held: held.read sul sink, held.set(ctx, n) dentro load a ogni riga accumulata e a ogni batch scritto, e zero in un finally (convenzione del report).

Edge di controllo

Un sink può essere l'origine di un edge di controllo: un task o una source con after: ["<nome del sink>"] parte solo dopo che il sink è riuscito. È il modo di scambiare le tabelle dopo il caricamento, o di caricare le dimensioni prima dei fatti.

• Il sink non è riuscito se è fallito (anche con l'errore catturato da un guard), se è stato saltato, o se il suo input è stato troncato: un guard ha catturato l'errore di un nodo a monte. In quest'ultimo caso il sink ha scritto e nel report ha success: true, ma i suoi successori sono saltati.
• Un sink con una source con after a monte chiama load solo quando i gate di tutte le source a monte sono aperti: finché aspetta non scrive nulla, nemmeno la parte che arriverebbe dalle altre source.
• Un sink non può essere la destinazione di un edge di controllo: per fare qualcosa prima della scrittura si mette after sulla source che lo alimenta.

Fail-fast. Quando un altro nodo fa fallire la run, il sink viene annullato: il suo stream d'ingresso rigetta subito e load deve chiudere. Un sink SQL fa rollback. Il runner aspetta che ogni load si concluda prima di riportare la run failed.

Anti-pattern

• Un INSERT semplice senza chiave naturale: una seconda run duplica tutto.
• Più sink con name: "memory" nello stesso grafo: collidono.
• Una scrittura il cui risultato serve a valle, fatta con un sink e riletta dopo: è un'action.
• Un upsert su una dimensione che deve conservare la storia: sovrascrive la versione precedente. Serve mode: "scd2".