Pulizia periodica in Node.js con PostgreSQL e node-cron

Un cron non garantisce che una riga venga elaborata una sola volta: offre soltanto l’occasione di eseguire del codice. Per rendere recuperabile una pulizia periodica in Node.js, il lavoro deve selezionare un piccolo batch, reclamare le righe all’interno di una transazione ed eliminare solo quelle che soddisfano già la regola di conservazione. Qui realizzeremo questo schema con PostgreSQL e node-cron.

È il database a decidere quali righe reclamare; il timer si limita a risvegliare il worker. L’esempio elimina i record scaduti da una tabella chiamata webhook_deliveries. Modifica il nome e la scadenza in base al tuo caso, ma mantieni la reclamazione condizionale e il limite per esecuzione.

Preparare la tabella per la pulizia

Il primo passaggio consiste nel rendere esplicita la condizione di scadenza. La tabella di esempio memorizza una data di scadenza e un contrassegno di reclamazione. Quest’ultimo permette a un’esecuzione di non confondere le righe idonee con quelle già in elaborazione.

CREATE TABLE webhook_deliveries (
  id BIGSERIAL PRIMARY KEY,
  payload JSONB NOT NULL,
  expires_at TIMESTAMPTZ NOT NULL,
  cleanup_claimed_at TIMESTAMPTZ
);

CREATE INDEX webhook_deliveries_cleanup_idx
  ON webhook_deliveries (expires_at, id)
  WHERE cleanup_claimed_at IS NULL;

L’indice parziale supporta la ricerca delle righe scadute che non sono ancora state reclamate. Non trasformare il contrassegno di reclamazione in un permesso eterno: se un processo si arresta, un contrassegno vecchio deve poter scadere e la riga deve tornare idonea. In questo esempio, la reclamazione scade dopo dieci minuti; il batch massimo è di cento righe.

Installare l’esecutore e l’accesso a PostgreSQL

La guida rapida di node-cron indica che il pacchetto richiede Node.js 20 o versioni successive. Installiamo lo scheduler e il client PostgreSQL; la connessione viene configurata tramite la variabile d’ambiente DATABASE_URL.

npm init -y
npm install node-cron pg

Useremo i moduli ES. Salva il file seguente come cleanup.js ed esegui node cleanup.js dopo aver configurato DATABASE_URL. L’esempio pianifica l’attività ogni minuto; l’espressione cron non determina se due processi distinti possano reclamare la stessa riga.

import cron from 'node-cron';
import pg from 'pg';

const { Pool } = pg;

if (!process.env.DATABASE_URL) {
  throw new Error('Falta DATABASE_URL');
}

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

console.log('Conexión PostgreSQL preparada');

Reclamare un batch con una transazione

La reclamazione e l’eliminazione devono essere decise dal database, non da un elenco caricato in memoria prima di iniziare. PostgreSQL consente di selezionare le righe con FOR UPDATE SKIP LOCKED: le righe bloccate da un’altra esecuzione vengono ignorate, invece di costringere gli altri worker ad attendere. La query seguente blocca un batch, elimina quelle righe e ne restituisce gli identificativi.

async function deleteExpiredBatch() {
  const client = await pool.connect();

  try {
    await client.query('BEGIN');

    const result = await client.query(`
      WITH candidates AS (
        SELECT id
        FROM webhook_deliveries
        WHERE expires_at <= NOW()
          AND (
            cleanup_claimed_at IS NULL
            OR cleanup_claimed_at < NOW() - INTERVAL '10 minutes'
          )
        ORDER BY expires_at, id
        LIMIT 100
        FOR UPDATE SKIP LOCKED
      )
      DELETE FROM webhook_deliveries AS d
      USING candidates AS c
      WHERE d.id = c.id
      RETURNING d.id
    `);

    await client.query('COMMIT');
    return result.rowCount;
  } catch (error) {
    await client.query('ROLLBACK');
    throw error;
  } finally {
    client.release();
  }
}

Il limite mantiene circoscritto ogni tentativo; se restano altre righe, le esecuzioni successive proseguiranno il lavoro. In questo esempio, la pulizia avviene interamente nel database e l’eliminazione viene confermata all’interno della transazione. Se devi coordinare un’operazione esterna, ad esempio chiamare una destinazione HTTP, non mantenerla all’interno di questa transazione: la chiamata di rete e il commit non costituiscono un’operazione atomica.

Collegare il batch alla pianificazione

Ora il callback pianificato esegue la funzione e registra quante righe ha eliminato. node-cron avvia immediatamente le attività create con schedule. La sua opzione noOverlap impedisce l’avvio di una seconda esecuzione della stessa attività mentre la precedente è ancora in corso nello stesso processo.

async function runCleanup() {
  const deleted = await deleteExpiredBatch();
  console.log(`${new Date().toISOString()} limpieza: ${deleted} filas`);
}

const task = cron.schedule('* * * * *', async () => {
  try {
    await runCleanup();
  } catch (error) {
    console.error('Falló la limpieza:', error);
  }
}, {
  name: 'limpieza-webhooks',
  timezone: 'UTC',
  noOverlap: true,
});

console.log(`Tarea activa: ${task.name}`);

noOverlap protegge solo questo processo. Se sono in esecuzione più copie dell’applicazione, ognuna può avere il proprio timer. La selezione con blocco e SKIP LOCKED evita che competano per le stesse righe finché dura la transazione, ma non trasforma un cron in un sistema di coordinamento distribuito e non risolve da sola il problema di un’attività esterna lenta.

flowchart TD
  A["node-cron despierta"] --> B["abrir transaccion"]
  B --> C["seleccionar lote vencido"]
  C --> D["bloquear filas disponibles"]
  D --> E["borrar hasta cien filas"]
  E --> F["confirmar o revertir"]

Interrompere l’attività allo spegnimento del processo

Durante una chiusura normale, smetti di accettare nuove esecuzioni e chiudi il pool di connessioni. Il gestore seguente arresta l’attività e attende che il pool termini la chiusura prima di uscire.

let shuttingDown = false;

async function shutdown(signal) {
  if (shuttingDown) return;
  shuttingDown = true;

  console.log(`Recibida señal ${signal}; deteniendo tarea`);
  task.stop();

  try {
    await pool.end();
    process.exit(0);
  } catch (error) {
    console.error('Error al cerrar PostgreSQL:', error);
    process.exit(1);
  }
}

process.on('SIGINT', () => void shutdown('SIGINT'));
process.on('SIGTERM', () => void shutdown('SIGTERM'));

Combina questi frammenti in un unico file, inserendo la configurazione del pool prima delle funzioni e la registrazione dei segnali dopo la creazione di task. In questo modo, il gestore può arrestare l’attività e chiudere lo stesso pool usato dalla pulizia.

Problemi da verificare

Un batch piccolo limita il lavoro in sospeso, ma non sostituisce il monitoraggio. Verifica cosa succede se due worker partono contemporaneamente, se il processo si arresta durante un’esecuzione e se la pulizia non riesce a svuotare la tabella in un ciclo. Registra almeno le righe elaborate e l’esito del tentativo: un’applicazione attiva non dimostra che il lavoro stia progredendo.

Se la pulizia richiede più fasi o lo stato deve sopravvivere a un arresto tra una fase e l’altra, non contrassegnare le righe e poi eliminarle separatamente senza una strategia di recupero. Per attività brevi come l’eliminazione mostrata, la transazione raggruppa selezione ed eliminazione. Per elaborazioni lunghe o tentativi operativi, può essere più adatto il modello di una coda persistente, a costo di mantenere più stato e coordinamento.

Infine, considera la scadenza e il fuso orario come decisioni esplicite. L’attività viene valutata in UTC e la condizione di scadenza usa l’ora di PostgreSQL. Se il requisito prevede di conservare i dati per un periodo diverso, modifica la regola di scadenza, non solo la frequenza del cron. Il ciclo successivo riprenderà fino a cento righe disponibili.

Fonti: DEV Community, node-cron, node-cron, Documentazione PostgreSQL