Implementare una coda di lavoro con PostgreSQL

Quante volte, di fronte alla necessità di processare dei job asincroni in background, la risposta automatica che hai sentito è stata “aggiungiamo Redis” o “mettiamo su RabbitMQ”? Molti danno per scontato che una coda di lavoro richieda un message broker dedicato. In realtà, non è così.

In questo articolo voglio condividere con te un approccio diverso, implementiamo una coda di lavoro con PostgreSQL! PostgreSQL offre tutto quello che ci serve: transazioni, lock a livello di riga e garanzie ACID collaudate da decenni. Vediamo come metterli insieme.

Il problema da risolvere

Partiamo dalla domanda fondamentale, cosa deve garantire una coda di lavoro?

Essenzialmente tre cose:

1. Più worker devono poter processare task in parallelo, senza pestarsi i piedi.
2. Un task non deve mai essere eseguito due o più volte.
3. Un task non deve mai essere assegnato a più di un worker contemporaneamente.

Immagina, ad esempio, un sistema che invia email transazionali. Se due worker prendono in carico lo stesso task, l’utente riceve la stessa email due volte. Non è un disastro, ma non è bello. Con task più critici, come un pagamento, il problema diventa serio.

Normalmente queste garanzie arrivano da un broker dedicato, ma un database relazionale con gestione dei lock può offrirle altrettanto bene.

Un lock è un meccanismo che controlla l’accesso concorrente a una risorsa del database, funziona un po’ come un semaforo. Quando una transazione acquisisce un lock su una risorsa, le altre trovano il rosso e aspettano che si liberi, evitando condizioni di race e garantendo l’integrità dei dati. PostgreSQL offre lock a diversi livelli di granularità, noi useremo i lock a livello di riga. Per un dettaglio sui diversi tipi di lock ti rimando a questo articolo sui lock in PostgreSQL.

Vediamo dunque come risolvere il problema con un database PostgreSQL e dei worker producer e consumer in Node.js.

Lo schema della tabella

Per implementare la coda definiamo la tabella tasks.

CREATE TYPE t_status AS ENUM ('pending', 'in_progress', 'done', 'failed');

CREATE TABLE tasks (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  payload JSONB NOT NULL,
  status t_status NOT NULL DEFAULT 'pending',
  created_at TIMESTAMP NOT NULL DEFAULT NOW(),
  picked_at TIMESTAMP,
  executed_at TIMESTAMP
);

Vediamo i campi uno per uno:

• id: identificativo univoco del task. Usiamo un UUID generato dal database, così il producer di task non deve preoccuparsi di generare chiavi e possiamo referenziare il task nei log senza ambiguità.
• payload: campo JSONB contenente i dati necessari al worker consumer per eseguire il task. La coda non impone una struttura, è un contratto tra producer e consumer.
• status: lo stato del task nel suo ciclo di vita. Un task nasce pending, passa a in_progress quando un worker lo prende in carico, e termina in done o failed. Usare un ENUM invece di una stringa libera ci protegge da stati inventati (in-progress? completed?) direttamente a livello di database.
• created_at: il momento in cui il task è stato inserito in coda. Useremo questo campo per garantire che i task vengano processati in ordine di arrivo (FIFO).
• picked_at: il momento in cui un worker ha preso in carico il task. Utile per il monitoraggio, ad esempio per individuare task rimasti in_progress troppo a lungo, sintomo di un probabile bug nel worker consumer.
• executed_at: il momento in cui il task è stato completato (o è fallito definitivamente). La differenza tra executed_at e created_at ci dà il tempo di esecuzione effettivo.

Riprendendo l’esempio delle email transazionali, un payload potrebbe essere:

{
  "type": "send_email",
  "to": "user@example.com",
  "template": "order_confirmation",
  "params": {
    "order_id": "A-2024-1234"
  }
}

La query che fa tutto il lavoro

Il cuore del sistema è questa query:

SELECT *
FROM tasks
WHERE status = 'pending'
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT 1;

A parole, ecco cosa vogliamo: “dammi il prossimo task in ordine di creazione che sia nello stato pending e che non sia bloccato da un lock di un’altra transazione”.

• FOR UPDATE acquisisce un lock di riga esclusivo sulla riga selezionata, che verrà rilasciato solo al commit (o rollback) della transazione. Avendo modellato un task come una riga della tabella tasks, i lock di riga sono lo strumento perfetto per dire “questo task è mio” senza bloccare gli altri worker.
• SKIP LOCKED è la parte interessante. Stiamo chiedendo a PostgreSQL di ignorare le righe già bloccate da un’altra transazione, e di non aspettare che si liberino.

Il risultato? Se cinque worker eseguono questa query nello stesso istante, ognuno prende un task diverso. Nessuno aspetta, nessuno riceve il task di un altro. Tutte e tre le garanzie di cui parlavamo prima, in una sola query.

Fammi vedere il codice!

Ecco il codice del worker consumer.

export const fetch = async (): Promise<Task | null> => {
  const trx = await db.transaction();
  try {
    const rows = await trx('tasks')
      .select<Task[]>('*')
      .forUpdate()
      .skipLocked()
      .where({ status: 'pending' })
      .orderBy('created_at', 'asc')
      .limit(1);

    if (rows.length === 0) {
      return null;
    }

    await trx('tasks')
      .update({
        status: 'in_progress',
        picked_at: db.fn.now(),
      })
      .where({ id: rows[0].id });

    await trx.commit();

    return rows[0];
  } catch (error) {
    await trx.rollback();
    console.error('Error fetching task:', error);
    return null;
  }
};

Il worker fa partire una transazione, usa la query che abbiamo visto in precedenza per ottenere il lock su un task pending, lo marca in_progress e fa commit. Il lock gli garantisce che nessun altro worker stia operando su quel task e il commit libererà il lock sulla riga. Se qualcosa va storto prima del commit, il rollback rilascia il lock e il task torna disponibile per un altro worker.

Il ciclo del worker consumer, a questo punto, è molto semplice:

1. Ottieni il prossimo task da lavorare.
2. Lavora il task.
3. Marca il task come done.

while (true) {
  const task = await fetch();

  if (!task) {
    await new Promise((resolve) => setTimeout(resolve, 2_000));
    continue;
  }

  try {
    await executeTask(task);
    await db('tasks')
      .update({ status: 'done', executed_at: db.fn.now() })
      .where({ id: task.id });
  } catch (error) {
    // gestione del retry, la vediamo tra poco
  }
}

E se un task fallisce?

Bella domanda. I sistemi a code offrono meccanismi di retry, e la nostra coda non può essere da meno. Anche qui, però, non serve nulla di eccessivamente complesso, bastano tre colonne in più.

ALTER TABLE tasks ADD COLUMN retry_count INTEGER NOT NULL DEFAULT 0;
ALTER TABLE tasks ADD COLUMN max_attempts INTEGER NOT NULL DEFAULT 3;
ALTER TABLE tasks ADD COLUMN error_message TEXT;

Quando l’esecuzione di un task fallisce, il worker decide se ritentare o arrendersi:

const newRetryCount = task.retry_count + 1;

if (newRetryCount >= task.max_attempts) {
  await db('tasks')
    .update({
      status: 'failed',
      retry_count: newRetryCount,
      error_message: error.message,
      executed_at: db.fn.now(),
    })
    .where({ id: task.id });
} else {
  await db('tasks')
    .update({
      status: 'pending',
      retry_count: newRetryCount,
      error_message: error.message,
      picked_at: null,
    })
    .where({ id: task.id });
}

La logica è lineare. Se retry_count è ancora sotto max_attempts, il task torna pending e un altro worker (o lo stesso) lo riprenderà. Superata la soglia, il task viene marcato failed, con l’ultimo errore salvato in error_message per il debug. Di fatto, i task falliti restano nella tabella e fanno da “dead letter queue”, interrogabile con una normale SELECT.

Quando usare questo approccio

• Accodamento transazionale con i dati applicativi:
  • il worker producer inserisce i dati di business e il task nella stessa transazione in maniera atomica. Non esistono situazioni in cui scrivi sul database e poi pubblichi su RabbitMQ (o viceversa) e uno dei due possa fallire.
• Semplicità:
  • riutilizzi il database, un componente già presente nella tua architettura.
  • la base di codice rimane pulita. Non si aggiunge nessuna nuova dipendenza.
  • il codice per la gestione di producer / consumer è lineare, semplice da leggere e manutenere.
• Debuggabilità:
  • per ispezionare il contenuto di un task non serve consumare un messaggio, basta una SELECT sulla tabella.
  • per rimettere un task di nuovo in esecuzione basta una UPDATE sullo stato. Inoltre, mantenendo la created_at iniziale, verrà eseguito con priorità più alta rispetto agli altri.
  • in generale tutta l’analisi del workload si riduce a delle query SQL, di fatto abbiamo a disposizione il linguaggio più efficiente che esista per interrogare e aggregare dati.
• Efficacia a basso costo:
  • per job asincroni, invio di email, elaborazioni in background di volumi moderati, il database è più che sufficiente.

Quando non usare questo approccio

• Carico di lavoro sulla coda eccessivo:
  • il throughput è vincolato alla capacità del database di gestire i lock. Se hai bisogno di processare ~50.000 messaggi al secondo, allora affidati a un sistema a code dedicato.
  • inoltre la coda vive sullo stesso database che serve la produzione, e quindi compete con il carico applicativo.
• Mancanza di feature avanzate:
  • il nostro approccio non consente il pattern publisher / subscriber verso più consumer indipendenti sullo stesso messaggio, e se stai pensando di implementarlo, allora stai per lanciarti in una quest che ti porterà solo a reinventare la ruota.
• Scheduling:
  • non puoi schedulare un lavoro o ritardarlo. Di nuovo, se stai pensando di implementarlo, rileggi il punto precedente.

Conclusioni

Torniamo alla domanda iniziale. Serve davvero “aggiungere Redis” o “mettere su RabbitMQ”? Come abbiamo visto, spesso no. Una tabella, una query con FOR UPDATE SKIP LOCKED e una manciata di righe di TypeScript ci hanno dato una coda con worker concorrenti, retry e dead letter queue, senza aggiungere un singolo componente da monitorare, aggiornare e debuggare. La semplicità è la cosa migliore che ti possa accadere, e quasi sempre ti farà vincere. Parti dal database che hai già, e passa a un broker dedicato quando i numeri (o le feature) te lo richiedono, non prima. Se vuoi metterci le mani, il codice completo del proof of concept lo trovi in questo repository.

Risorse utili

• PostgreSQL Documentation - Explicit Locking
• PostgreSQL Documentation - SELECT (clausola FOR UPDATE/SKIP LOCKED)
• Lock di tabelle con PostgreSQL