Job in background

Panoramica

Un job è una riga in rebase.jobs. Viene acquisito da esattamente un worker, ritentato con un ritardo progressivo se il suo handler genera un errore, e lasciato nella tabella quando alla fine si arrende, in modo che qualcuno possa esaminarlo.

Non c’è nulla da installare e nulla da eseguire insieme a Postgres. Un job accodato all’interno di una transazione che esegue un rollback non è mai stato accodato.

Usalo per attività che non devono essere perse e che non devono essere eseguite all’interno di una richiesta: inviare email, chiamare terze parti, generare un file, riconciliare dati con un sistema esterno.

	Esecuzione	Sopravvive a un riavvio
Cron	Pianificata	Sì — la pianificazione è nel codice
Job	Una volta, non appena un worker è libero	Sì — il job è una riga
Un setTimeout in un callback	Una volta, in questo processo	No

Abilitazione

await initializeRebaseBackend({
    jobs: {
        enabled: true,
        tasks: {
            "send-welcome": async ({ payload }) => {
                await sendEmail((payload as { email: string }).email);
            }
        }
    }
});

Disabilitato se non richiesto esplicitamente: un worker continua a cercare lavoro finché il processo è in vita, un comportamento che non è una scelta predefinita desiderabile per tutti. Richiede un driver in grado di eseguire SQL — su uno che non può farlo (MongoDB), la coda non è disponibile e ti viene segnalato all’avvio anziché al primo inserimento in coda.

Da 0.25 Un worker inattivo non esegue il polling ogni due secondi. Ogni inserimento in coda invia un NOTIFY sul canale rebase_jobs al momento del commit, e un worker che tiene un LISTEN su di esso acquisisce subito il job — un job accodato da un altro processo (il ruolo api, un’altra replica) parte in pochi millisecondi, e uno differito quando scade. Dietro a questo, controlla comunque ogni 30 secondi. Un job che questo processo accoda, un retry che ha pianificato e un job che ha recuperato lo svegliano senza una notifica.

Il LISTEN richiede una connessione propria, trovata nello stesso modo in cui la trova il realtime: DATABASE_DIRECT_URL se impostato, altrimenti DATABASE_URL. Dietro un pooler in modalità transazione come PgBouncer, imposta DATABASE_DIRECT_URL: un pooler non mantiene un LISTEN, e i job di altri processi attenderebbero allora il controllo dei 30 secondi. Dove non esiste alcuna connessione di questo tipo (il database di sviluppo in-process), un worker inattivo rallenta: controlla dopo pollIntervalMs (2 secondi), poi il doppio dopo ogni controllo che non ha trovato nulla, fino a 30 secondi, e ricomincia ogni volta che trova lavoro. Un worker inattivo fa circa 2.900 acquisizioni al giorno invece di 43.200.

Accodamento

const { jobQueue } = await initializeRebaseBackend({ jobs: { enabled: true, tasks } });

await jobQueue?.enqueue("send-welcome", { email: "ada@example.com" });

Opzioni

await jobQueue?.enqueue("send-digest", { userId: "u7" }, {
    delayMs: 60_000,               // not before a minute from now
    maxAttempts: 5,                // default 3
    idempotencyKey: "digest:u7"    // at most one *unfinished* job with this key
});

idempotencyKey raggruppa un doppio clic, una richiesta ritentata e due istanze che reagiscono allo stesso evento in un unico job. Il suo ambito è limitato al lavoro non completato, quindi la chiave diventa riutilizzabile una volta completato il job — altrimenti “il digest notturno per l’utente 7” sarebbe inviabile esattamente una sola volta in assoluto. Un accodamento duplicato si risolve in null anziché generare un errore: il lavoro richiesto è in coda, che è il risultato desiderato.

Fallimenti

Un handler fallisce sollevando un’eccezione (throwing). Non esiste alcun return false — un booleano verrebbe ignorato silenziosamente da ogni handler che si dimentica di restituirne uno, e il fallimento deve essere il comportamento predefinito in caso di errore.

• Tentativi rimanenti → torna a pending, con run_at posticipato dal backoff (1s, 5s, 25s … con un limite massimo di un’ora; sovrascrivibile con backoff).
• Tentativi esauriti → failed, e la riga rimane. Una coda che elimina silenziosamente ciò che non è riuscita a consegnare è indistinguibile da una che non ha nulla da fare.

SELECT task, attempts, last_error, updated_at
FROM rebase.jobs WHERE status = 'failed'
ORDER BY updated_at DESC;

Le righe con esito negativo vengono conservate per 30 giorni; quelle riuscite per 3. Ogni worker elimina le righe scadute quando si avvia e poi ogni ora, a lotti, fuori dal percorso delle richieste.

Cosa succede quando un worker si arresta

Un processo terminato durante l’esecuzione di un job non può rilasciare la presa in carico (claim), quindi solo un timeout potrà liberare la riga. Mentre un handler è in esecuzione, il suo worker rinnova la presa in carico ogni terzo di visibilityTimeoutMs (predefinito: 5 minuti). Una presa in carico rimasta senza rinnovo per tutto quel tempo viene recuperata — tornando a pending se il job ha ancora tentativi disponibili, altrimenti spostata in dead-letter con un errore che spiega l’accaduto.

Quindi un handler lento su un worker attivo mantiene il proprio job, per quanto tempo impieghi, e il timeout decide soltanto quanto presto tornano disponibili i job di un worker che si è arrestato. Neppure un tentativo che ha perso la propria presa in carico può sovrascrivere l’esito del job: il suo successo o fallimento viene registrato nei log e scartato, e decide il tentativo che detiene la presa in carico.

jobs: {
    enabled: true,
    concurrency: 5,              // jobs at once, per instance
    pollIntervalMs: 2_000,       // the first wait after a look that found nothing; it doubles up to 30 s
    visibilityTimeoutMs: 300_000 // how long an unrenewed claim holds
}

Un handler che non finisce mai

Il rovescio della medaglia: un handler in attesa di una fetch che non risponde mai tiene occupato il suo slot di worker, e la sua presa in carico, finché il processo è in vita — con cinque di questi il worker non esegue nient’altro, mentre /health resta verde. Da 0.25 timeoutMs limita un tentativo. Alla scadenza il tentativo fallisce — ritentato finché restano tentativi, come qualsiasi throw — il suo slot viene liberato, e il signal dell’handler viene interrotto. Passa il signal a chi di dovere, così anche il lavoro si ferma; un handler che lo ignora continua a girare in background senza che nulla ne registri l’esito.

Se omesso, vale 15 minuti (900_000), e il worker lo dice con una riga di log all’avvio. Un job che legittimamente dura di più imposta un proprio valore, e timeoutMs: 0 rimuove il limite.

jobs: {
    enabled: true,
    timeoutMs: 120_000,          // one attempt; 15 minutes when left out, 0 for no limit
    tasks: {
        "sync-crm": async ({ payload, signal }) => {
            await fetch(crmUrl, { method: "POST", body: JSON.stringify(payload), signal });
        }
    }
}

Gli handler della coda e le sottoscrizioni ai topic ricevono lo stesso signal nel loro contesto.

Istanze multiple

Sicuro per progettazione. I worker acquisiscono i job con SELECT … FOR UPDATE SKIP LOCKED, quindi ogni job viene assegnato a uno solo di essi e gli altri passano alla riga successiva invece di mettersi in coda dietro di essa. Non è necessario eleggere alcun leader.

Un’istanza acquisisce solo i task per cui ha un handler. Durante un rolling deploy, un’istanza che esegue codice precedente lascia stare i job il cui task è implementato solo dal codice più recente: restano pending, senza consumare tentativi, finché un’istanza aggiornata non li acquisisce. Anche un task che nessuna istanza implementa resta pending, invece di finire in dead-letter.

Webhook persistenti

Per impostazione predefinita, WebhookDispatcher accoda i suoi invii in memoria; questo significa che un crash o un deploy tra la modifica e l’invio comporterebbe la perdita dell’evento. Assegnagli la coda e ogni invio diventerà una riga:

import { WebhookDispatcher, WEBHOOK_DELIVERY_TASK } from "@rebasepro/server";

const { jobQueue } = await initializeRebaseBackend({ jobs: { enabled: true } });

const dispatcher = new WebhookDispatcher({ jobQueue });
dispatcher.setWebhooks(myWebhooks);

jobQueue?.register(WEBHOOK_DELIVERY_TASK, ctx => dispatcher.deliverQueuedJob(ctx.payload as never, ctx));

Sul job viene memorizzato solo l’id del webhook, mai il webhook stesso — altrimenti il suo segreto di firma rimarrebbe in chiaro in rebase.jobs per tutto il periodo di retention della riga, e un webhook modificato tra l’accodamento e l’invio deve essere inviato con la configurazione attuale. Passare ctx come secondo argomento fa sì che ogni tentativo invii il proprio X-Webhook-Attempt; tutti i tentativi di una stessa consegna inviano lo stesso X-Webhook-Delivery, così un destinatario può deduplicare i retry.

Un invio che il dispatcher si rifiuta di effettuare (una destinazione loopback, link-local, privata o non http(s)), o a cui il destinatario risponde con un redirect, fallisce ogni volta allo stesso modo. Finisce in dead-letter dopo un solo tentativo, con il motivo in last_error. Qualsiasi altro errore viene ritentato come per ogni job.

Arresto

shutdown() impedisce al worker di acquisire nuovi job e attende il completamento di quelli in corso (in flight), evitando che un deploy esegua due volte la coda di un batch. L’attesa è limitata a due terzi del timeout di arresto (10 secondi dei 15 predefiniti), così un handler che non termina mai non può bloccare l’arresto. Un job ancora in esecuzione dopo questo limite mantiene il proprio blocco e viene recuperato dal timeout di visibilità.

Passaggi successivi

• Cron Jobs — lavoro pianificato
• Webhooks — invia notifiche ad altri sistemi in caso di modifiche