Job heartbeat support

[meyer9]

If we had a field last_heartbeat, we could more quickly detect when a job fails due to process crashes, etc where it may not properly update to failed.

One example would be to send a heartbeat every 15 seconds (setting the column to NOW()) and marking a job as failed/expired if the heartbeat is over 60 seconds old.

[Crispy1975]

@timgit I was thinking about this exact addition to pg-boss. We have a scenario as the OP describes where a worker process exits in an uncontrolled way leaving a job in the active state. There doesn't seem to be a way for other workers to know about this and therefore we end up with jobs stuck in limbo.

As the OP mentioned a heartbeat column in the jobs table would allow for a maintenance task or other workers to detect the stuck job and perhaps move it into the retry state so it can again be processed.

What are your thoughts on this? I am happy to work on this update/feature as we have a a fairly urgent need. It should also benefit other pg-boss users too. 😄

[schester44]

+1 to this, would love to see pg-boss better recognize when a job has stopped for reasons such as the worker process crashing.

[timgit]

Internal maintenance guarantees active jobs are retried or failed after their timeout/expiration. This behaves similarly to the visibility timeout in SQS. You should tune the expiration to what should be considered normal execution time as well.

[KristjanTammekivi]

Internal maintenance guarantees active jobs are retried or failed after their timeout/expiration. This behaves similarly to the visibility timeout in SQS. You should tune the expiration to what should be considered normal execution time as well.

Well, you can but at the same time sometimes some task might take longer, then you can either increase the expected time to way longer than the normal execution time or risk same task be run multiple times. Heartbeat should reasonably easy to implement and would solve that issue at the cost of an extra column in the job table.

[KristjanTammekivi]

Hi @timgit I see that this is marked as completed but I can't seem to find it in the source code or documentation. Can you please give more information on the subject?

[timgit]

I didn't mean to mark this as completed. I was resolving lot of old issues during the v10 release.

[afonsomatos]

Hey, I have this issue as well.

My situation is that our jobs can take 60 seconds or 15 minutes to complete, depending of the configuration of that particular job.

We want our users to get the job done as quickly as possible.

If the worker dies at 60 seconds, it's too much to wait another 14 minutes (for the expiration timeout).
Job heartbeat support would fix this.

[kibertoad]

@timgit Would you be interested in PR implementing heartbeat support?

[timgit]

I am. What is your idea of an implementation?

[kibertoad]

@timgit give me a bit of time to put together a proposal

[kibertoad]

@timgit OK, so here comes. There are two useful precedents - BullMQ and sqs-consumer.

Bullmq

BullMQ implements a LockManager that automatically renews job locks:

• Automatic Lock Renewal: Workers start a lock renewal process via lockManager.start() unless skipLockRenewal is configured
• Lock Extension API: extendJobLocks(jobIds, tokens, duration) extends locks in Redis
• Configuration:
  • lockDuration: Total time a job lock is valid (default: 30000ms)
  • lockRenewTime: Interval for automatic renewal (default: lockDuration / 2)
  • stalledInterval: How often stalled jobs are checked (default: 30000ms)
• Job Tracking: Each job is tracked with { token, timestamp, abortController }
• Renewal Loop: Runs every lockRenewTime / 2 milliseconds, extending locks for jobs whose timestamp exceeds the threshold
• AbortController Integration: Jobs can be aborted if lock renewal fails

From my understanding a renewal loop is basically a setInterval that is started at a job execution start.

SQS Consumer (AWS SQS)

SQS Consumer implements heartbeat through visibility timeout extension:

• Single Message: changeVisibilityTimeout() sends ChangeMessageVisibilityCommand
• Batch Messages: changeVisibilityTimeoutBatch() for multiple messages
• Heartbeat Interval: setInterval() periodically refreshes visibility timeouts
• Configuration:
  • heartbeatInterval: Refresh frequency (in seconds)
  • visibilityTimeout: Initial and extended timeout duration
• Lifecycle: Heartbeat starts during processMessage(), clears in finally block

As we can see, in both cases timeout extension is automatic, without extra actions needed from user side, based purely on configuration.

Now here is my proposal for the pg-boss:

Add automatic heartbeat functionality to workers, similar to BullMQ's LockManager

API Design

interface WorkOptions {
  // ... existing options

  // Enable automatic heartbeat
  heartbeat?: boolean | HeartbeatOptions
}

interface HeartbeatOptions {
  // Interval in seconds for automatic heartbeat (default: expireInSeconds / 2)
  intervalSeconds?: number

  // Whether to abort the job if heartbeat fails (default: true)
  abortOnFailure?: boolean
}

Usage Examples

// Simple automatic heartbeat
await boss.work('long-task', { heartbeat: true }, async ([job]) => {
  await performLongOperation(job.data)
})

// Custom heartbeat configuration
await boss.work('batch-job', {
  heartbeat: {
    intervalSeconds: 60,  // Touch every 60 seconds
    abortOnFailure: true
  }
}, async ([job]) => {
  await processBatch(job.data)
})

Implementation Details

When heartbeat is enabled in WorkOptions:

1. After fetching jobs, the Manager starts an interval timer for each job
2. Timer calls touchJob() at the configured interval
3. Timer is cleared when the job completes (success, failure, or abort)
4. If heartbeat fails (e.g., job was already timed out by another process), the AbortController is triggered

// In Manager.ts - pseudocode
private async startHeartbeat(job: Job, options: ResolvedHeartbeatOptions) {
  const intervalMs = (options.intervalSeconds ?? job.expireInSeconds / 2) * 1000

  const timer = setInterval(async () => {
    try {
      const touched = await this.boss.touchJob(job.name, job.id)
      if (!touched && options.abortOnFailure) {
        job.abortController.abort(new Error('Heartbeat failed - job may have timed out'))
        clearInterval(timer)
      }
    } catch (error) {
      this.emit('error', error)
    }
  }, intervalMs)

  // Store timer reference for cleanup
  this.activeHeartbeats.set(job.id, timer)
}

touchJob implementation

SQL Implementation

-- New function: touch_jobs
-- Extends the effective timeout by resetting started_on or adding to expire_seconds

-- Option A: Reset started_on (simpler, resets the full timeout window)
UPDATE {schema}.{table}
SET started_on = now()
WHERE name = $1
  AND id IN (SELECT UNNEST($2::uuid[]))
  AND state = 'active'
RETURNING id

-- Option B: Add to expire_seconds (more flexible, accumulative)
UPDATE {schema}.{table}
SET expire_seconds = COALESCE($3, expire_seconds) +
    GREATEST(0, EXTRACT(EPOCH FROM (now() - started_on))::int)
WHERE name = $1
  AND id IN (SELECT UNNEST($2::uuid[]))
  AND state = 'active'
RETURNING id

I don't really like option A, because it messes up with job execution history, and if something relies on that column for observability, they will get data that is simply not true. I would prefer the option B here, but maybe I'm missing some possible risks with bumping the expiration time incrementally?

Alternative is expecting user to call touchJob manually over the course of the job, but it feels like a very poor DX, and I don't expect users to use it consistently or well, or at all. I don't see drawbacks from a fully automated approach at this point, as long as Node process is still alive and not blocking the event loop, heartbeat could and should pass. If we want to protect against potentialy eternal loops and mark them as a timeout nevertheless, maybe we can put a sanity cap on how long the expire_seconds could be (but it should be something significant, as jobs running for hours is not impossible)

[timgit]

Thanks for the research on this.

It seems like there are 2 different requests in this issue showing up.

1. Fail an active job if a worker that fetched it failed its health check
2. Extend expiration of an active job instead of failing it if it's taking longer than usual. (what I think you're proposing above)

If we go with option 1, it seems like most would want this feature to always be on and not have to opt in, assuming it's part of the batteries included with a polling worker.

For option 2, I agree with your opinion on avoiding changing started_on, but if we change expire_seconds, we'll need to mess with the logic in resolveWithinSeconds in order to add a new extend() function on delay() to reset the timeout.

[kibertoad]

@timgit 1) seems to be already addressed by abortOnFailure param.

On topic of resolveWithinSeconds, how about this?

Changes to src/tools.ts

• Add an extend() method to the delay promise:

export interface ExtendableAbortablePromise<T> extends Promise<T> {
  abort: () => void
  extend: (newDurationMs: number) => void  // Reset timeout to new duration from now
}

function delay(ms: number, error?: string, abortController?: AbortController): ExtendableAbortablePromise<void> {
  const ac = abortController || new AbortController()
  let currentTimeoutId: ReturnType<typeof setTimeout>
  let settled = false  // Track if promise has already resolved/rejected

  const scheduleReject = (resolve: () => void, reject: (err: Error) => void, timeoutMs: number) => {
    currentTimeoutId = setTimeout(() => {
      if (settled) return  // Guard against calling resolve/reject after settlement
      settled = true
      if (error) {
        reject(new Error(error))
      } else {
        resolve()
      }
    }, timeoutMs)
  }

  let resolveRef: () => void
  let rejectRef: (err: Error) => void

  const onAbort = () => {
    if (settled) return
    settled = true
    clearTimeout(currentTimeoutId)
    resolveRef()
  }

  const promise = new Promise<void>((resolve, reject) => {
    resolveRef = resolve
    rejectRef = reject
    scheduleReject(resolve, reject, ms)

    ac.signal.addEventListener('abort', onAbort, { once: true })  // Auto-removes after firing
  }) as ExtendableAbortablePromise<void>

  promise.abort = () => {
    if (!ac.signal.aborted) {
      ac.abort()
    }
  }

  promise.extend = (newDurationMs: number) => {
    if (settled) return  // No-op if already settled
    clearTimeout(currentTimeoutId)
    scheduleReject(resolveRef, rejectRef, newDurationMs)
  }

  return promise
}

Changes to resolveWithinSeconds

export interface ExtendableTimeout<T> {
  promise: Promise<T | void>
  abort: () => void
  extend: (newDurationMs: number) => void
}

function resolveWithinSeconds<T>(
  promise: Promise<T>,
  seconds: number,
  message?: string,
  abortController?: AbortController
): ExtendableTimeout<T> {
  const timeout = Math.max(1, seconds) * 1000
  const timeoutPromise = delay(timeout, message, abortController)

  const racePromise = (async () => {
    let result
    try {
      result = await Promise.race([promise, timeoutPromise])
    } finally {
      timeoutPromise.abort()
    }
    return result
  })()

  return {
    promise: racePromise,
    abort: () => timeoutPromise.abort(),
    extend: (newDurationMs: number) => timeoutPromise.extend(newDurationMs)
  }
}

Integration with Heartbeat in src/manager.ts

When heartbeat succeeds, extend the local timeout:

private activeHeartbeats = new Map<string, ReturnType<typeof setInterval>>()

private startHeartbeat(job: Job, timeout: ExtendableTimeout, options: ResolvedHeartbeatOptions) {
  const intervalMs = (options.intervalSeconds ?? job.expireInSeconds / 2) * 1000
  const extensionMs = job.expireInSeconds * 1000  // Full timeout window

  const timer = setInterval(async () => {
    try {
      const touched = await this.boss.touchJob(job.name, job.id)
      if (touched) {
        // Successfully touched in DB - extend local timeout to match
        timeout.extend(extensionMs)
        this.emit('heartbeat', { name: job.name, jobId: job.id })
      } else if (options.abortOnFailure) {
        this.stopHeartbeat(job.id)
        job.abortController.abort(new Error('Heartbeat failed - job may have timed out'))
      }
    } catch (error) {
      this.emit('error', error)
    }
  }, intervalMs)

  this.activeHeartbeats.set(job.id, timer)
}

private stopHeartbeat(jobId: string) {
  const timer = this.activeHeartbeats.get(jobId)
  if (timer) {
    clearInterval(timer)
    this.activeHeartbeats.delete(jobId)
  }
}

// Must be called in finally block after job handler completes
private cleanupJob(jobId: string) {
  this.stopHeartbeat(jobId)
  // ... other cleanup
}

Note that the heartbeat interval MUST be cleaned up when the job completes. In the fetch/execute flow:

// In the job execution flow
try {
  const result = await timeout.promise  // resolveWithinSeconds wrapper
  await this.complete(name, jobIds, result)
} catch (err) {
  await this.fail(name, jobIds, err)
} finally {
  // Always clean up heartbeat timer, regardless of success/failure
  for (const jobId of jobIds) {
    this.stopHeartbeat(jobId)
  }
  timeout.abort()  // Also abort the timeout promise if still pending
}

This ensures that when a heartbeat successfully extends the database timeout, the local JavaScript timeout is also reset, keeping them in sync.

Worker Shutdown Cleanup

When a worker is stopped (via boss.stop() or boss.offWork()), all active heartbeat timers must be cleaned up:

// In Worker.stop() or Manager cleanup
async stopAllHeartbeats() {
  for (const [jobId, timer] of this.activeHeartbeats) {
    clearInterval(timer)
  }
  this.activeHeartbeats.clear()
}