Jobs

A job is one unit of work: your data, an optional ID, and a status that changes as the worker runs it. You never build one yourself. Add creates it, puts it in the queue and gives you a handle.

Two views of a job

The worker function and the code that called Add see different things.

The worker function gets a Job[T]. It has three methods. Data() returns what you passed to Add, and ID() returns the job’s ID. Context() ends when someone cancels the job while it runs. The interface has no status and no Close.

The caller gets a handle. Add returns the handle and a bool:

job, ok := queue.Add("resize photo")

ok is false when the queue did not take the job. This happens when the queue is full or closed, or when a job with the same ID is still in the queue. In that case job is nil, so check ok before you use it.

With a plain worker the handle is an EnqueuedJob. The other two worker types add one method that returns the outcome:

Every handle has ID(), Status(), IsClosed(), Wait(), WaitContext(), Close() and Cancel(). Err and Result block until the job is closed, then return what the worker function sent. For a job that was closed before it ran, Err returns nil and Result returns the zero value and a nil error. See worker types for the three workers side by side.

Job IDs

A job has no ID unless you give it one. ID() returns an empty string by default.

• WithJobId sets the ID of one job: queue.Add(data, varmq.WithJobId("job-1")). An empty string is ignored.
• WithJobIdGenerator sets a function that makes IDs. Add calls it, on the caller’s goroutine, for every job; its value is used when the job has no ID of its own.

An in-memory queue refuses a second Add with an ID that a job still in the queue holds. Job IDs covers when an ID is free again, generated IDs and the g: prefix of AddAll items.

Job status

Status() on a handle returns one of five strings:

Finished is short-lived. As soon as your function returns, the worker closes the job to release anyone waiting on it. By the time Wait returns, the status is Closed, and IsClosed() is true. So Closed does not tell you whether the job ran. Record that in your own worker function if it matters.

Move one job through its statuses below. Pick the worker type to see what the handle returns, and try Close() at each status.

The worker below runs one job at a time, and each job takes 500 ms. The program reads both statuses while job-1 runs.

package main

import (
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(func(j varmq.Job[string]) {
		fmt.Println("running", j.ID(), j.Data())
		time.Sleep(500 * time.Millisecond)
	})
	queue := worker.BindQueue()

	first, _ := queue.Add("resize photo", varmq.WithJobId("job-1"))
	second, _ := queue.Add("send email", varmq.WithJobId("job-2"))
	time.Sleep(100 * time.Millisecond) // job-1 has started by now

	fmt.Println(first.ID(), first.Status())
	fmt.Println(second.ID(), second.Status())

	first.Wait()
	second.Wait()
	fmt.Println(first.ID(), first.Status(), first.IsClosed())
	fmt.Println(second.ID(), second.Status(), second.IsClosed())
}

job-1 is Processing and job-2 is Queued. After Wait, both read Closed and IsClosed() is true, though both ran.

Closing a queued job

Close() cancels a job that has not started. It only marks the job closed; it does not take it out of the queue. When the worker reaches a closed job it skips it, and your function never sees it. Closing also releases Wait immediately.

It succeeds while the job waits. It fails in two cases:

• A running job returns ErrJobProcessing. To stop a running job, cancel it.
• A closed job returns ErrJobAlreadyClosed. That includes a job that already ran, since the worker closes those too.

On shared storage, Close only stops this handle from waiting. It returns nil, Err then returns ErrDetached, and the job still runs.

Until the worker reaches it, a job closed while it waited still counts in the queue’s Len() and in the worker’s NumPending(). Because it never ran, it never counts as completed.

Here job-1 is running and job-2 and job-3 wait behind it.

package main

import (
	"fmt"
	"time"

	"github.com/goptics/varmq"
)

func main() {
	worker := varmq.NewWorker(func(j varmq.Job[string]) {
		fmt.Println("running", j.ID(), j.Data())
		time.Sleep(500 * time.Millisecond)
	})
	queue := worker.BindQueue()

	first, _ := queue.Add("resize photo", varmq.WithJobId("job-1"))
	second, _ := queue.Add("send email", varmq.WithJobId("job-2"))
	third, _ := queue.Add("build report", varmq.WithJobId("job-3"))
	time.Sleep(100 * time.Millisecond) // job-1 is running, the others wait

	fmt.Println("close job-1:", first.Close())
	fmt.Println("close job-2:", second.Close())
	second.Wait() // returns at once: job-2 is closed
	fmt.Println("job-2 is", second.Status())
	fmt.Println("close job-2 again:", second.Close())

	third.Wait() // the worker skips job-2 and runs job-3
}

Closing job-1 fails with ErrJobProcessing. Closing job-2 works, its Wait returns immediately while job-1 is still running, and a second Close fails with ErrJobAlreadyClosed. The worker then runs job-3 and never prints a line for job-2.

Cancelling a job

Cancel() stops a job, whether it waits or runs.

• A job that has not started is closed and never runs. On an error or result worker, Err or Result returns ErrJobCancelled.
• A running job has its Context() cancelled. VarMQ cannot stop your function, so the function must watch j.Context() and return.
• A running job that then returns an error, or panics, ends cancelled, and Err returns ErrJobCancelled. If it returns without an error, it ends as usual.
• A job that has ended returns ErrJobAlreadyClosed.

worker := varmq.NewErrWorker(func(j varmq.Job[string]) error {
	select {
	case <-time.After(time.Minute):
		return nil
	case <-j.Context().Done():
		return j.Context().Err()
	}
})
queue := worker.BindQueue()

job, _ := queue.Add("report")
job.Cancel()
fmt.Println(job.Err()) // varmq: job cancelled

On shared storage, Cancel reaches the job in whichever process holds it. A waiting workflow task is the exception: cancel its workflow. The storage must support it. Without that support, Cancel returns ErrCancelUnsupported. The REST API can cancel a job by its ID too.

Jobs in other processes

On shared storage such as Redis or SQLite, another process can run the job. The handle that Add returns still learns how the job ended, if the storage reports outcomes. A producer that runs no worker gets the same handles from NewQueue, NewErrQueue or NewResultQueue.

• Err returns nil for a completed job. For a failed job it returns a *RemoteError. It holds only the error text, so errors.Is does not match your own error values.
• Result decodes the result from JSON. It returns ErrNoResult when the storage kept no result, for example a result over 64 KiB.
• ErrDetached means this handle cannot learn the outcome. The storage does not report outcomes, or you called Close on the handle. The job itself still runs.

A job on storage that reports outcomes or cancels jobs always has an ID: when you give none, VarMQ makes a random one. To wait on a job from a process that did not add it, use OpenJob, OpenErrJob or OpenResultJob with that ID. The storage finds a job that already ended while it keeps the job’s record. Persistence says which storage supports this.

Waiting on a job

Wait() blocks until the job is closed, whether it ran or you closed it. If that has already happened, it returns immediately. WaitContext does the same, but gives up when its context ends. To wait for everything a worker has queued instead of one job, see waiting.

Next: Worker lifecycle.