> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sqd.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# Background and scheduled jobs

> Run scheduled database jobs without blocking an SQD Cloud processor.

Run expensive, wall-clock-driven work in a separate Cloud processor service. Do not run it inside the indexer's batch transaction.

<Warning>
  A query started from the batch handler shares the processor transaction. A
  lock wait, timeout, or serialization failure can roll back the whole batch
  and stop indexing until the query completes or fails.
</Warning>

This pattern is useful for:

* leaderboards and rankings recalculated every few minutes;
* materialized summaries and exports;
* retention or cleanup jobs;
* reconciliation against an external system.

It is not needed for lightweight calculations that naturally belong to one indexing batch.

## Architecture

Use two named processor services that share the same Cloud Postgres add-on:

1. The indexer writes canonical blockchain data.
2. The background service opens its own database pool, waits for the next wall-clock interval, and runs the aggregate in its own transaction.
3. An advisory lock prevents two copies of the job from overlapping.
4. The job exposes Prometheus metrics so its runs and failures are observable.

Both services receive the Cloud-provided `DB_URL`. They run as separate processes, so an aggregation failure does not roll back the indexer's batch.

## Add a background command

Add a command for the job to `commands.json`:

```json title="commands.json" theme={"system"}
{
  "commands": {
    "process:prod": {
      "deps": ["migration:apply"],
      "cmd": ["node", "lib/main.js"]
    },
    "aggregate:prod": {
      "cmd": ["node", "lib/jobs/aggregate.js"]
    }
  }
}
```

The deployment-level [`init` command](/en/cloud/reference/manifest#init) applies migrations before either processor service starts. The background command should not run migrations itself.

## Declare both processor services

```yaml title="squid.yaml" theme={"system"}
manifest_version: subsquid.io/v0.1
name: account-leaderboard

build:

deploy:
  addons:
    postgres:
  init:
    cmd: ["sqd", "migration:apply"]
  processor:
    - name: indexer
      cmd: ["sqd", "process:prod"]
    - name: leaderboard
      cmd: ["sqd", "aggregate:prod"]
  api:
    cmd: ["sqd", "serve:prod"]

scale:
  dedicated: true
  processor:
    profile: medium
  addons:
    postgres:
      profile: medium
      storage: 100Gi
```

Processor names must be unique. Each named processor runs and is billed as a separate service. The `scale.processor.profile` setting applies to the processor services in the deployment.

## Implement the scheduler safely

Install the database and metrics dependencies used by the job:

```bash theme={"system"}
pnpm add pg prom-client@^14.2.0 @subsquid/util-internal-prometheus-server
pnpm add -D @types/pg
```

The following skeleton uses one Postgres session for the advisory lock and transaction. Replace the aggregate query with your own:

```typescript title="src/jobs/aggregate.ts" expandable theme={"system"}
import {Pool} from 'pg'
import {Counter, Gauge, Registry} from 'prom-client'
import {createPrometheusServer} from '@subsquid/util-internal-prometheus-server'

const intervalMs = 5 * 60 * 1000
const lockId = 742019
const pool = new Pool({connectionString: process.env.DB_URL})

const registry = new Registry()
const runs = new Counter({
  name: 'sqd_background_job_runs_total',
  help: 'Background job attempts',
  labelNames: ['result'],
  registers: [registry],
})
const lastSuccess = new Gauge({
  name: 'sqd_background_job_last_success_unixtime',
  help: 'Unix timestamp of the last successful run',
  registers: [registry],
})

async function main(): Promise<void> {
  await createPrometheusServer(
    registry,
    process.env.PROCESSOR_PROMETHEUS_PORT ?? process.env.PROMETHEUS_PORT
  )

  while (true) {
    const waitMs = intervalMs - (Date.now() % intervalMs)
    await new Promise((resolve) => setTimeout(resolve, waitMs))

    const client = await pool.connect()
    let locked = false
    let inTransaction = false

    try {
      const result = await client.query<{locked: boolean}>(
        'select pg_try_advisory_lock($1) as locked',
        [lockId]
      )
      locked = result.rows[0]?.locked === true

      if (locked) {
        await client.query('begin isolation level read committed')
        inTransaction = true
        await client.query("set local statement_timeout = '4min'")
        await client.query(`
          insert into account_rank (account_id, rank)
          select account_id, row_number() over (order by score desc)
          from account_stat
          on conflict (account_id) do update set rank = excluded.rank
        `)
        await client.query('commit')
        inTransaction = false
        runs.inc({result: 'success'})
        lastSuccess.setToCurrentTime()
      } else {
        runs.inc({result: 'skipped'})
      }
    } catch (error) {
      if (inTransaction) {
        await client.query('rollback').catch(() => undefined)
      }
      runs.inc({result: 'error'})
      console.error(error)
    } finally {
      if (locked) {
        await client.query('select pg_advisory_unlock($1)', [lockId])
      }
      client.release()
    }
  }
}

void main().catch((error) => {
  console.error(error)
  process.exit(1)
})
```

<Note>
  `READ COMMITTED` is an example, not a universal default. Choose the weakest
  isolation level that preserves your aggregate's correctness, and test it
  against concurrent indexer writes.
</Note>

## Production checklist

* Make the query safe to retry.
* Prevent overlapping runs with an advisory lock or equivalent mechanism.
* Keep the job's database pool separate from the indexer's pool.
* Set a statement timeout appropriate for the job.
* Record success, failure, duration, and last-success metrics.
* Alert when the last-success timestamp becomes stale.
* Test the query on production-sized data before enabling the schedule.
* Review the additional processor and database load in [Billing and Usage](https://app.subsquid.io/billing/usage).

If the job needs to replay blockchain history rather than aggregate existing database rows, follow [Schema changes and backfills](/en/cloud/resources/schema-changes-and-backfills).