Background and scheduled jobs

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

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.

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:

The indexer writes canonical blockchain data.
The background service opens its own database pool, waits for the next wall-clock interval, and runs the aggregate in its own transaction.
An advisory lock prevents two copies of the job from overlapping.
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:

commands.json

{
  "commands": {
    "process:prod": {
      "deps": ["migration:apply"],
      "cmd": ["node", "lib/main.js"]
    },
    "aggregate:prod": {
      "cmd": ["node", "lib/jobs/aggregate.js"]
    }
  }
}

The deployment-level init command applies migrations before either processor service starts. The background command should not run migrations itself.

Declare both processor services

squid.yaml

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:

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:

src/jobs/aggregate.ts

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)
})

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.

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.

If the job needs to replay blockchain history rather than aggregate existing database rows, follow Schema changes and backfills.

Managed

Self-Hosting

Architecture

Add a background command

Declare both processor services

Implement the scheduler safely

Production checklist

​Architecture

​Add a background command

​Declare both processor services

​Implement the scheduler safely

​Production checklist

Architecture

Add a background command

Declare both processor services

Implement the scheduler safely

Production checklist