# How to Wait for All Jobs to Finish in VarMQ: Complete Guide

> Learn how to wait for all jobs to finish in VarMQ using worker WaitUntilFinished or WaitAndStop. Complete guide for efficient job management in goptics varmq.

- Repository: [Goptics/varmq](https://github.com/goptics/varmq)
- Tags: how-to-guide
- Published: 2026-03-02

---

**Use `worker.WaitUntilFinished()` to block until every queued job is dequeued and processed, or call `worker.WaitAndStop()` to finish all work and shut down the worker cleanly.**

VarMQ is a Go job queue library that processes jobs concurrently through pool-based workers. When you need to ensure all enqueued tasks complete before proceeding—such as during graceful shutdowns or test assertions—the library provides built-in synchronization primitives that block until the worker becomes idle.

## Core Blocking Methods in VarMQ

The `Worker` type in [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go) exposes three primary APIs for waiting on job completion. Each serves a distinct use case depending on whether you want to continue processing, pause temporarily, or stop entirely.

### WaitUntilFinished

The `WaitUntilFinished()` method blocks the calling goroutine while the worker remains busy. It returns only when the internal condition indicates no pending jobs and zero in-flight processing.

In [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go), this method implements a condition variable pattern:

```go
// Simplified logic from worker.go#L209-L227
func (w *Worker) WaitUntilFinished() error {
    w.mu.Lock()
    defer w.mu.Unlock()
    
    for w.condition() {
        w.waiters.Wait() // Blocks on sync.Cond
    }
    return nil
}

```

The underlying `condition` closure evaluates to `true` while the worker is `running` and (`queues.Len() > 0` or `curProcessing > 0`), or while the worker is `paused`/`stopped` but `curProcessing > 0`. When `releaseWaiters` broadcasts—triggered when `curProcessing` drops to zero—the blocked caller wakes and returns.

### PauseAndWait

When you need to stop accepting new jobs but allow current work to finish, use `PauseAndWait()`. This atomic operation first transitions the worker to `paused` status, then invokes `WaitUntilFinished()`.

According to the source implementation, pausing prevents the worker from pulling new jobs from the queue set, while the wait ensures all `curProcessing` goroutines complete. After returning, you can safely call `Resume()` to continue processing or perform maintenance without data loss.

### WaitAndStop

For graceful shutdown scenarios, `WaitAndStop()` combines waiting with termination. This method waits for all in-flight jobs to finish using the same condition logic as `WaitUntilFinished()`, then stops the worker cleanly by terminating internal goroutines.

## How the Waiting Mechanism Works Internally

The synchronization relies on a `sync.Cond` (`w.waiters`) coupled with atomic counters. Understanding this flow helps diagnose blocking behavior in production:

1. **Condition Evaluation**: The `condition` closure in [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go) checks `w.status`, `queues.Len()`, and the atomic `curProcessing` counter. It returns `true` while work remains active.

2. **Efficient Blocking**: Rather than polling, `WaitUntilFinished()` locks the worker mutex and calls `w.waiters.Wait()`, yielding the CPU until broadcast.

3. **Release Trigger**: The private `releaseWaiters()` method (located around `worker.go#L183-L194`) broadcasts on the condition variable whenever `curProcessing` reaches zero and either the worker is paused or the queue set is empty.

4. **Completion Guarantee**: Upon return from `WaitUntilFinished()`, the mutex guarantees that `curProcessing == 0` and `queues.Len() == 0` (or the worker is stopped), ensuring all jobs have finished.

## Practical Code Examples

### Wait for All Jobs After Enqueuing

After starting a worker and adding payloads, block until the queue drains:

```go
package main

import (
    "fmt"
    "log"
    
    "github.com/goptics/varmq"
)

func main() {
    // Initialize queue and worker
    queue := varmq.NewQueue()
    worker := varmq.NewWorker(queue)
    
    if err := worker.Start(); err != nil {
        log.Fatalf("start failed: %v", err)
    }
    
    // Enqueue work
    for i := 0; i < 100; i++ {
        queue.Add(fmt.Sprintf("task-%d", i))
    }
    
    // Block until queue empty and all jobs processed
    if err := worker.WaitUntilFinished(); err != nil {
        log.Fatalf("wait failed: %v", err)
    }
    
    fmt.Println("All 100 jobs completed")
}

```

This pattern appears frequently in [`worker_test.go`](https://github.com/goptics/varmq/blob/main/worker_test.go) (lines 87-105), where tests verify queue drainage after batch inserts.

### Pause Processing and Wait

Use this pattern when you need to perform atomic operations or maintenance without shutting down:

```go
// Prevent worker from taking new jobs
if err := worker.Pause(); err != nil {
    log.Fatal(err)
}

// Wait for current batch to finish
if err := worker.WaitUntilFinished(); err != nil {
    log.Fatal(err)
}

// At this point: no new jobs started, all active jobs done
performMaintenance()

// Resume normal operation
if err := worker.Resume(); err != nil {
    log.Fatal(err)
}

```

Alternatively, use the convenience method:

```go
if err := worker.PauseAndWait(); err != nil {
    log.Fatal(err)
}

```

### Graceful Shutdown with WaitAndStop

For application shutdown hooks, ensure all jobs complete before exiting:

```go
func shutdownWorker(worker *varmq.Worker) {
    log.Println("Initiating graceful shutdown...")
    
    if err := worker.WaitAndStop(); err != nil {
        log.Printf("Error during shutdown: %v", err)
        return
    }
    
    log.Println("Worker stopped after finishing all jobs")
}

```

`WaitAndStop()` handles the synchronization internally, making it safer than manually calling `Stop()` which might interrupt running jobs.

### Wait for a Single Specific Job

Each `Job` struct embeds its own `sync.WaitGroup`. To block on an individual job rather than the entire worker:

```go
// AddJob returns *Job immediately
job := queue.AddJob(myPayload)

// Process other logic...

// Block only until this specific job completes
job.Wait()

// Implementation reference: job.go#L179-L183

```

This is useful when you need results from a specific task while allowing the worker to continue processing unrelated jobs.

## Summary

- **`WaitUntilFinished()`** blocks until the queue empties and all in-flight jobs complete, using an efficient `sync.Cond` mechanism defined in [`worker.go`](https://github.com/goptics/varmq/blob/main/worker.go).
- **`PauseAndWait()`** atomically pauses the worker and waits for current jobs, preventing new work from starting during the wait.
- **`WaitAndStop()`** provides graceful shutdown by waiting for completion then stopping the worker cleanly.
- **Individual job waiting** is available via `job.Wait()` for per-task synchronization.
- The waiting logic relies on atomic `curProcessing` counters and condition variables rather than busy-waiting, ensuring efficient CPU usage.

## Frequently Asked Questions

### What is the difference between WaitUntilFinished and PauseAndWait?

`WaitUntilFinished()` blocks until current work finishes while the worker remains in its current state—if running, it will continue pulling new jobs from the queue. `PauseAndWait()` first transitions the worker to `paused` status (preventing new job acquisition) and then waits. Use `PauseAndWait()` when you need to guarantee no new jobs start during the wait period, such as during maintenance windows or when preparing for a controlled shutdown.

### Is WaitUntilFinished safe to call from multiple goroutines?

Yes, `WaitUntilFinished()` is safe for concurrent use. The implementation locks the worker's mutex (`w.mu`) before checking the condition and waiting on the `sync.Cond`. Multiple goroutines can safely call this method; all will block until the completion condition is met, at which point the condition variable broadcasts to wake all waiters simultaneously.

### How do I wait for a specific job instead of all jobs?

Use the `Wait()` method on the individual `Job` pointer returned by `queue.AddJob()`. As implemented in [`job.go`](https://github.com/goptics/varmq/blob/main/job.go) (lines 179-183), each job maintains its own `sync.WaitGroup` that completes when the job handler finishes. This allows fine-grained synchronization without blocking the entire worker or affecting other concurrent jobs.

### What happens if I call WaitAndStop on an already stopped worker?

`WaitAndStop()` first checks the worker's status. If the worker is already stopped, the method returns immediately without error. The implementation ensures idempotent behavior: if `curProcessing` is already zero and the worker status is `stopped`, the condition evaluates to false instantly, and the method proceeds to the stop logic, which safely handles the already-stopped state.