Background job notifications

Background jobs are invisible by default. They start, they run for minutes or hours, they finish — or they don’t — and you find out by checking the queue dashboard or tailing logs.

Here’s how to make them tell you instead.

The basic pattern

Notify at the end of the job:

import fyi from "trigger.fyi"
 
async function processReport(jobId, userId) {
  const start = Date.now()
 
  try {
    const result = await generateReport(userId)
    const elapsed = Math.round((Date.now() - start) / 1000)
 
    fyi("Report complete", {
      body: `${result.rows} rows · ${elapsed}s`,
      userId,
      jobId
    })
 
    return result
  } catch (err) {
    fyi.critical("Report failed", {
      body: err.message,
      userId,
      jobId
    })
    throw err
  }
}

The notify call fires after the work is done. If the job fails, the critical notification fires before re-throwing so the queue can retry.

BullMQ

import { Worker } from "bullmq"
import fyi from "trigger.fyi"
 
const worker = new Worker("reports", async (job) => {
  const start = Date.now()
  const result = await processReport(job.data)
  const elapsed = Math.round((Date.now() - start) / 1000)
 
  fyi(`Report complete`, {
    body: `${result.rows} rows · ${elapsed}s`,
    queue: "reports",
    attempt: job.attemptsMade + 1
  })
 
  return result
})
 
worker.on("failed", (job, err) => {
  fyi.critical(`Report failed`, {
    body: err.message,
    queue: "reports",
    attempt: job.attemptsMade
  })
})

Celery (Python)

from celery import Celery
from trigger_fyi import fyi
import time
 
app = Celery("tasks")
 
@app.task(bind=True, max_retries=3)
def process_report(self, user_id):
    start = time.time()
 
    try:
        result = generate_report(user_id)
        elapsed = time.time() - start
        fyi("Report complete", body=f"{result['rows']} rows · {elapsed:.0f}s", user_id=str(user_id))
        return result
    except Exception as exc:
        fyi.critical("Report failed", body=str(exc), user_id=str(user_id))
        raise self.retry(exc=exc, countdown=60)

Sidekiq (Ruby)

require "trigger_fyi"
 
class ReportWorker
  include Sidekiq::Worker
 
  def perform(user_id)
    start = Time.now
    result = ReportGenerator.run(user_id)
    elapsed = (Time.now - start).round
 
    Fyi.call("Report complete",
      body: "#{result[:rows]} rows · #{elapsed}s",
      user_id: user_id.to_s
    )
  rescue => e
    Fyi.critical("Report failed",
      body: e.message,
      user_id: user_id.to_s
    )
    raise
  end
end

Long-running jobs: progress notifications

For jobs that take minutes, notify at milestones:

from trigger_fyi import fyi
 
def process_large_dataset(dataset_id, total):
    processed = 0
 
    for batch in get_batches(dataset_id):
        process_batch(batch)
        processed += len(batch)
 
        # Notify every 25%
        pct = int((processed / total) * 100)
        if pct in (25, 50, 75):
            fyi.log(f"Progress: {pct}%",
                body=f"{processed:,} of {total:,} records",
                dataset_id=str(dataset_id)
            )
 
    fyi("Dataset processed",
        body=f"{total:,} records complete",
        dataset_id=str(dataset_id)
    )

fyi.log() records in the feed without pushing. Progress milestones go to the feed; completion pushes to your phone.

Jobs that time out

If a job is supposed to complete in under 5 minutes:

const TIMEOUT_MS = 5 * 60 * 1000
 
async function withTimeout(jobFn, jobId) {
  const timer = setTimeout(() => {
    fyi.critical("Job timeout", {
      body: `${jobId} exceeded ${TIMEOUT_MS / 60000}m`,
      jobId
    })
  }, TIMEOUT_MS)
 
  try {
    const result = await jobFn()
    clearTimeout(timer)
    return result
  } catch (err) {
    clearTimeout(timer)
    throw err
  }
}

What to include in the notification

Useful things to log with every job notification:

Duration: how long it took
Count: records processed, items completed
Attempt number: is this a retry?
Entity ID: what entity was being processed
Environment: prod vs staging

fyi("Sync complete", {
  body: `${count} records · ${elapsed}s`,
  entity: entityId,
  attempt: job.attemptsMade,
  env: process.env.NODE_ENV
})

The metadata is filterable. Open the feed, filter by env: prod, see only production jobs.

#The basic pattern

#BullMQ

#Celery (Python)

#Sidekiq (Ruby)

#Long-running jobs: progress notifications

#Jobs that time out

#What to include in the notification