Worker lifecycle

A worker is always in one of seven states. Status() returns the current one as a string, and five Is* methods answer yes or no for some of them.

The diagram below is a worker you can drive. Click a state to read about it. Press a method to call it. The worker has three jobs waiting and runs one at a time. Pause and Stop never interrupt a running job, so press Finish to end it.

The seven states

Idle and Running swap on their own as jobs start and finish. You cannot set Running. The worker sets it while jobs run. IsActive is true for either of them, which makes it the check for “started and not paused or stopped”.

The program below creates a worker with WithAutoRun(false), adds three jobs, starts it, and pauses it while job 1 runs.

package main

import (
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(
		func(j varmq.Job[int]) {
			time.Sleep(100 * time.Millisecond)
			fmt.Println("job", j.Data(), "done")
		},
		varmq.WithAutoRun(false),
	)
	queue := worker.BindQueue()

	for i := 1; i <= 3; i++ {
		queue.Add(i)
	}
	fmt.Println("created:", worker.Status())

	worker.Start()
	time.Sleep(10 * time.Millisecond) // job 1 is running
	fmt.Println("started:", worker.Status())

	worker.Pause()
	fmt.Println("Pause:", worker.Status())

	worker.WaitUntilPaused()
	fmt.Println("paused:", worker.Status(), "- pending:", worker.NumPending())

	worker.Resume()
	worker.WaitUntilIdle()
	fmt.Println("drained:", worker.Status())
}

The worker is Initiated until Start. Start makes it Idle, and it is Running a moment later, because jobs were already waiting. Pause returns while job 1 still runs, so the worker is Pausing. Once job 1 is done it is Paused, and jobs 2 and 3 wait until Resume. When they are done the worker is Idle again.

A status is a snapshot. Between reading it and acting on it the worker may have moved on. To know where a worker has landed after Pause or Stop, use the waiting helpers below.

Moving between states

From any other state, each call returns an error. The errors below are in the same order as the states.

The errors are values in the package, listed under Variables. Two details:

• Pause, Resume and Stop return immediately. Restart does not: it waits for running jobs to finish before it starts the worker again. If another goroutine restarts the worker first, it returns ErrRunningWorker.
• A stopped worker cannot be resumed. Resume returns ErrWorkerStopped. Use Restart, or Start, which also accepts a stopped worker.

The program below stops a worker, tries to resume it, adds a job while it is stopped, then restarts it.

package main

import (
	"fmt"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(func(j varmq.Job[int]) {
		fmt.Println("job", j.Data(), "done")
	})
	queue := worker.BindQueue()

	fmt.Println("Start:", worker.Start())

	worker.StopAndWait()
	fmt.Println("stopped:", worker.Status())
	fmt.Println("Resume:", worker.Resume())

	queue.Add(1)
	fmt.Println("added while stopped - pending:", worker.NumPending())

	worker.Restart()
	worker.WaitUntilIdle()
	fmt.Println("restarted:", worker.Status())
}

Start fails on a worker that is already running, which a worker is by default (see WithAutoRun). Resume refuses the stopped worker. The job added while it was stopped waits in the queue, and runs after Restart.

What happens to jobs

Pause and stop never interrupt a running job. Each running job finishes first.

Jobs that have not started stay queued and are not lost. A worker may have taken several jobs from the queue together (see how it works). On pause or stop it finishes the job it is running and parks the rest of the batch. Parked jobs are still Queued, and the worker’s NumPending counts them, but the queue’s Len() does not. They run first, in their original order, after Resume or Restart. Two more cases make NumPending read high. A job closed while it waits still counts, in Len() too, until the worker reaches it. While a goroutine works through a batch, the batch’s finished jobs count until the whole batch ends. Purge closes them.

A worker pause is not a queue pause. A paused queue stops every worker that takes from it, in every process. Pausing a queue covers it.

A stop also cancels the worker’s Context(). A restart gives the worker a fresh one.

Waiting for a state

Pause, Resume and Stop only start the change. To block until it is done, call PauseAndWait or StopAndWait. Waiting covers these and the other wait calls.

Settings

WithAutoRun

WithAutoRun decides whether the worker starts when you create it. The default is true: NewWorker returns a worker that is already Idle, and Start on it returns ErrRunningWorker.

Pass WithAutoRun(false) to get an Initiated worker instead. You can bind queues and add jobs to it right away. The jobs wait until you call Start. Use it when you want to finish setting things up before any job runs.

WithStoppedRetention

This setting matters only for a worker with a name (WithName or a string argument to NewWorker). Start adds a named worker to WorkerRegistry, the map that the REST Handler reads. A worker without a name is never registered.

When a registered worker reaches Stopped, it stays in the registry for a while, so it still appears as Stopped. WithStoppedRetention sets how long. When that time has passed and the worker is still stopped, it is removed from the registry.

• The default is one minute. DefaultStoppedRetention changes the default for workers created after the call.
• 0 turns the cleanup off: a stopped worker stays in the registry for good.
• Starting the worker again cancels the timer, and re-registers the worker if it had already been removed.

worker := varmq.NewWorker(process, 2,
	varmq.WithName("emails"),
	varmq.WithStoppedRetention(10*time.Minute),
)

Next: How it works.