Background Tasks

Some work has to happen on a schedule instead of in response to a request: checking a ledger for drift every fifteen minutes, retrying failed payouts, closing out stale sessions. Dyrected ships a small task runner for exactly this, so you don't need a separate scheduler service for the common cases.

A task is a name, a schedule, and a function. The runner takes care of the part that is easy to get wrong: making sure that when you run several instances of your app, each scheduled run still happens only once.

Define a task

A task is what you hand to defineTask, and you list it under tasks in your config. This one runs every fifteen minutes and expires payments that have sat in pending for over an hour, using only the db the runner gives it:

import { defineCollection, defineConfig, defineSelectField, defineTask } from '@dyrected/core'
import { SqliteAdapter } from '@dyrected/db-sqlite'

const Payments = defineCollection({
  slug: 'payments',
  fields: [
    defineSelectField({ name: 'status', label: 'Status', options: ['pending', 'paid', 'expired'] }),
  ],
})

export const expireStalePayments = defineTask({
  name: 'payments:expire-stale',
  cron: '*/15 * * * *',
  lockTimeout: 300,
  async run({ db, logger }) {
    const cutoff = new Date(Date.now() - 60 * 60 * 1000).toISOString()
    const stale = await db.find({
      collection: 'payments',
      where: { status: { equals: 'pending' }, createdAt: { less_than: cutoff } },
      limit: 100,
    })
    for (const payment of stale.docs) {
      await db.update({ collection: 'payments', id: payment.id, data: { status: 'expired' } })
    }
    logger.info({ expired: stale.docs.length }, 'Expired stale payments')
  },
})

export default defineConfig({
  collections: [Payments],
  globals: [],
  db: new SqliteAdapter({ filename: './dyrected.db' }),
  tasks: [expireStalePayments],
})

The task gets db for reading and writing, logger for output that lands in your normal logs, and a signal you'll meet in the lock timeout section below.

defineTask checks the name, the cron expression and the timeout right away, so a typo fails when the server starts, not at 3am. If you leave out cron, the task only runs when you call it yourself.

The cron expression has the usual five fields (minute hour day-of-month month day-of-week) and is evaluated in UTC. Lists, ranges, steps and month or weekday names all work, for example 0 3 * * mon.

Start the runner

Dyrected doesn't start the runner for you, because a timer is only reliable in a process that stays alive. You choose how it runs.

On a long-lived Node server, create the runner and start it. start() returns immediately, so it never delays your server booting:

import { createTaskRunner } from '@dyrected/core'

const runner = createTaskRunner(config)
runner.start()

// on shutdown
await runner.stop()

On Nuxt, set runTasks: true in the module options and Dyrected starts and stops the runner with the server.

On serverless, there is no process to keep a timer alive. Call runDue() from your platform's cron (Vercel Cron, Cloudflare Cron Triggers) every minute:

const results = await createTaskRunner(config).runDue()

You can also run one task on demand, ignoring its schedule, with runner.runTask('wallet:ledger-drift').

How duplicate runs are prevented

Every task has a lock row in the database. Before running, an instance claims the lock with a single conditional update, and only the instance whose update matched goes ahead. For scheduled runs it also claims the specific time slot, so a slow instance can't run a slot that another instance already handled.

That means you can run as many instances as you like, and runDue() is safe to call from all of them at once.

If a task throws, the run is recorded as failed, the lock is released, and the next slot runs normally. The last status and error are stored on the task's lock row.

Lock timeout

lockTimeout is how many seconds a run may hold its lock, and it defaults to 300. If a run goes past it, Dyrected records it as failed and aborts the signal passed to your function. If the process dies mid-run, the lock simply expires after the timeout and another instance takes over.

Pass signal to anything cancellable, such as fetch, or check it inside long loops, so a run that has lost its lock stops writing data:

async run({ db, signal }) {
  for (const account of accounts) {
    if (signal.aborted) return
    await reconcile(db, account)
  }
}

Choose a timeout comfortably longer than the task's normal runtime.

Good to know

• A missed slot isn't replayed. After downtime the task runs once for the current slot, not once per missed slot.
• Runs are checked about every 30 seconds, so a task starts within that window of its scheduled minute. Pass intervalMs to start() to change it.
• Make tasks safe to run twice. The lock prevents overlap, but a run that fails halfway can be picked up again on the next slot.
• If a task needs several writes to succeed or fail together, wrap them in a transaction, and keep slow external calls outside it.