guide · worker

Stopping without dropping work.

Deploys are the most common cause of stalled jobs. Closing a worker properly lets in-flight jobs finish and be acknowledged before the process exits.

Control and shut down

Bun

worker.run();     // Start processing (if created with autorun: false)
worker.pause();   // Stop pulling new jobs
worker.resume();  // Resume

await worker.close();      // Wait for active jobs, then stop
await worker.close(true);  // Force close immediately

Node.js / Deno

worker.run();     // Start processing (if created with autorun: false)
worker.pause();   // Stop pulling new jobs
worker.resume();  // Resume

await worker.close();      // Wait for active jobs, then stop
await worker.close(true);  // Force close immediately

Python

worker.run()             # Blocking loop (or autorun starts it in the background)
worker.pause()           # Stop pulling new jobs
worker.resume()          # Resume

worker.close()           # Stop pulling, wait for in-flight jobs to drain
worker.close(timeout=5)  # Bound the wait; the drain continues in the background

PHP

$worker->run();      // Blocking loop: pull, process, repeat
$worker->runOnce();  // Pull and process one batch (cron-friendly)

$worker->stop();     // Finish the in-flight job, then return from run()
$worker->close();    // Unregister and close the connection

Go

worker.Run()   // Blocking pull loop
worker.Stop()  // Stop pulling; in-flight jobs finish
worker.Close() // Unregister and close the connection

Rust

worker.run()?;      // Blocking pull loop; returns after stop()
worker.run_once()?; // Pull and process one batch
worker.stop();      // Ask the loop to exit
worker.close();     // Stop, unregister and close the connection

Elixir

Bunqueue.Worker.run(worker)       # Blocking pull loop; returns after stop
Bunqueue.Worker.run_once(worker)  # Pull and process one batch
Bunqueue.Worker.stop(worker)      # Drain, unregister and close

Pausing stops new pulls; it does not interrupt processors already running or release jobs already buffered by the Worker. Lease-renewal and worker-registration heartbeats therefore continue while paused. resume() reuses those heartbeat loops, and repeated pause/resume cycles do not create additional timers.

close(true) stops owning the active delivery immediately; it does not forcibly cancel JavaScript already running inside the processor. If that processor later returns or throws, the Worker discards the late outcome without sending ACK or FAIL. The broker can therefore recover the unfinished job through disconnect, lock-expiry, or stall handling instead of accepting a result from a closed worker. Use plain close() when the current result must be committed before shutdown.

Processing timeouts follow the same generation rule during normal operation. The broker owns the deadline and claims the active generation atomically. If a processor returns or throws after that claim, the Worker receives an explicit ignored outcome and suppresses its local completed, failed, and ACK/FAIL error events. A retry’s newer lease is independent and can complete normally.

Cancel active processors cooperatively (Bun)

Every Bun processor receives an AbortSignal in its second argument. Cancel one delivery or every active delivery owned by this Worker:

const worker = new Worker('downloads', async (job, context) => {
  return await fetch(job.data.url, { signal: context?.signal });
});

worker.cancelJob(jobId, 'request withdrawn'); // false unless this Worker is currently executing that job (pulled-but-not-started jobs return false too)
worker.cancelAllJobs('service is shutting down');
worker.isJobCancelled(jobId); // true while its cancelled delivery remains active here

Cancellation aborts the signal and emits cancelled({ jobId, reason }). A Promise processor must pass that signal to fetch or another cancellable API, or check signal.aborted; JavaScript cannot forcibly stop a Promise that ignores it. Structural Observable processors are unsubscribed automatically. The resulting processor rejection follows normal failure/retry handling.

close(true) is different: it relinquishes active delivery ownership and suppresses late outcomes so broker recovery can redeliver. Use cancellation when the processor should observe and handle an abort; use forced close when the process must stop owning work immediately.

For a clean process exit, hook your runtime’s shutdown signal; in the Bun client, also shut down the shared machinery:

Bun

import { shutdownManager, closeSharedTcpClient } from 'bunqueue/client';

process.on('SIGINT', async () => {
  await worker.close();
  shutdownManager();       // Embedded mode: flush writes, close SQLite
  closeSharedTcpClient();  // TCP mode: close the shared connection pool
  process.exit(0);
});

Node.js / Deno

import { closeSharedTcpClient } from 'bunqueue-client';

process.on('SIGINT', async () => {
  await worker.close();
  closeSharedTcpClient();  // TCP mode: close the shared connection pool
  process.exit(0);
});

Python

try:
    worker.run()
except KeyboardInterrupt:
    worker.close()  # drains in-flight jobs

PHP

$worker->installSignalHandlers(); // SIGTERM / SIGINT -> graceful stop
$worker->run();

Go

go func() {
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
    <-sig
    worker.Stop() // Run() drains in-flight jobs, then closes
}()
worker.Run()

Rust

// `ctrlc` is the signal-hook crate used by this application.
let shutdown_worker = worker.clone();
ctrlc::set_handler(move || shutdown_worker.stop())?;

worker.run()?; // returns after the handler calls stop()
worker.close();

Elixir

task = Task.async(fn -> Bunqueue.Worker.run(worker) end)

receive do
  :shutdown ->
    :ok = Bunqueue.Worker.stop(worker) # waits for the active run_once batch
    Task.await(task, :infinity)
end

In Rust and Elixir, calling worker.stop() / Bunqueue.Worker.stop(worker) from another thread or process makes the blocking run loop drain and return; there is no shared client machinery to tear down.

Where to go next

Guide	What it covers
Worker	Create a worker and process your first job
Worker Concurrency and Batch Pulling	Run jobs in parallel and pull them in batches
The Job Object Inside a Worker Processor	Everything the processor receives and can do
Worker Events	completed, failed, stalled and the rest
Worker Error Handling, Retries and Backoff	Retries, backoff, timeouts and giving up
Heartbeats, Stall Detection and Lock Ownership	Heartbeats, stall recovery and lock ownership
SandboxedWorker	Experimental isolation for CPU-heavy handlers
WorkerOptions Reference	Every WorkerOptions field, with defaults