Errors
A failure in VarMQ appears in one of three places. A few values are returned directly from calls you make. This page covers:
- where a failed job is reported;
- how to read the worker’s error channel and set an error handler;
- every exported
Errvalue, with the call that returns it.
Add has no error result. It returns a bool, and false means the queue did not take the job. See queues.
Where a failure is reported
| Where | What it gets | Which workers |
|---|---|---|
The job’s handle: Err() or Result() | what the function returned, or the panic | NewErrWorker and NewResultWorker |
The worker’s Errs() channel | every failure the worker sees | all three |
The error handler, set with WithErrHandler | every failure the worker sees | all three |
A job’s function can fail in two ways, and both are reported the same way:
- It returns an error. An error worker or result worker sends that error to the job’s handle, to
Errs()and to the handler. The error is the same value you returned, soerrors.Isworks on it. A plainNewWorkerfunction returns nothing, so it cannot fail this way. - It panics. The worker recovers the panic and counts the job as failed. The panic becomes an error, which goes where a returned error goes. A plain worker has no handle to carry it, so it goes to
Errs()and the handler only.
The error for a panic reads panic recovered inside worker: <value>. For an error worker it is err-worker instead of worker, and for a result worker result-worker. The value is formatted into the text with %v, so errors.Is cannot find the panic value inside it. Worker types shows the three handles side by side.
The worker itself can also report an error that belongs to no job function: a storage problem or a failed acknowledgement. Those go to Errs() and the handler only.
For a group of jobs added with AddAll, an error group handle has its own Errs(). That one is a different channel, and it is closed once the group has ended. The worker’s Errs() below is not. See adding many jobs. A failure in a group job also reaches the worker’s Errs() and the handler.
The Errs channel
The worker’s Errs() returns a read-only channel of error. Three things to know about it:
- It holds one error. The channel has room for a single value. When a job fails and the channel is full, the error is not put on the channel, and the worker does not wait for you. A reader that is slow, or absent, loses errors. The handler does not.
- It is never closed.
for err := range worker.Errs()does not end when the worker stops. Run it in a goroutine of its own, or read withselectand adefault. - An error does not say which job it came from. Put the job’s ID, or something else that identifies the work, in the message you return.
go func() {
for err := range worker.Errs() {
log.Println("worker error:", err)
}
}()A reader like this one drains the channel as errors arrive. If you need every error, set a handler as well. The channel can still drop an error when two errors arrive faster than the reader takes them.
In the program below every job fails and nothing reads Errs() while the jobs run.
8 lines: package and imports
package main
import (
"fmt"
"github.com/goptics/varmq"
)
func main() {
worker := varmq.NewErrWorker(func(j varmq.Job[int]) error {
return fmt.Errorf("job %d failed", j.Data())
})
queue := worker.BindQueue()
for i := 1; i <= 3; i++ {
queue.Add(i)
}
worker.WaitUntilIdle() // nothing read Errs() while the jobs ran
fmt.Println("waiting on Errs():", len(worker.Errs()))
fmt.Println("first:", <-worker.Errs())
select {
case err := <-worker.Errs():
fmt.Println("second:", err)
default:
fmt.Println("no second error")
}
}
Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.
Output
Three jobs failed, but one error waits on the channel: job 1 failed, the first. The other two found the channel full and were dropped, so the select takes its default. The message names the job because the program put the number in it.
The error handler
WithErrHandler sets a function that the worker calls for every error it reports. The function receives the Worker and the error, so one handler can serve several workers and log the worker’s Name(). The program below names its error worker mailer with a string after the function and sets a handler. Job 2 returns an error and job 3 panics.
9 lines: package and imports
package main
import (
"errors"
"fmt"
"github.com/goptics/varmq"
)
func main() {
worker := varmq.NewErrWorker(func(j varmq.Job[int]) error {
switch j.Data() {
case 2:
return errors.New("bad input")
case 3:
panic("boom")
}
return nil
}, "mailer", varmq.WithErrHandler(func(w varmq.Worker, err error) {
fmt.Printf("handler (%s): %v\n", w.Name(), err)
}))
queue := worker.BindQueue()
var jobs []varmq.EnqueuedErrJob
for i := 1; i <= 3; i++ {
job, _ := queue.Add(i)
jobs = append(jobs, job)
}
worker.WaitUntilIdle()
for i, job := range jobs {
fmt.Printf("job %d: %v\n", i+1, job.Err())
}
}
Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.
Output
The handler runs once for each failure and prints mailer. Its lines come first, because the program waits for the worker to go idle before it reads the handles. Each handle then has its own outcome: nil, the returned error, and the panic as panic recovered inside err-worker: boom.
- The handler runs for every error, including those the full
Errs()channel dropped. - It runs synchronously on the goroutine that reports the error, and it must not block. For an error from a job, or from a failed acknowledgement, that is the goroutine that ran the job. A slow handler then delays that goroutine’s next job. For an error from pulling a job off its storage, it is the worker’s event loop. The event loop hands out every job, so a slow handler delays all of them.
- Passing
nilis ignored. The worker keeps the handler it had. - The default handler does nothing.
DefaultErrHandlerchanges it for the workers you create after the call. Configuration says how the defaults work.
Matching an error
The package exports its sentinel errors as variables, so match them with errors.Is, not by comparing text:
9 lines: package and imports
package main
import (
"errors"
"fmt"
"github.com/goptics/varmq"
)
func main() {
worker := varmq.NewWorker(func(j varmq.Job[int]) {
fmt.Println("job", j.Data())
})
worker.BindQueue()
err := worker.TunePool(1)
fmt.Println("TunePool(1):", err)
fmt.Println("is ErrSameConcurrency:", errors.Is(err, varmq.ErrSameConcurrency))
worker.StopAndWait()
err = worker.Resume()
fmt.Println("Resume:", err)
fmt.Println("is ErrWorkerStopped:", errors.Is(err, varmq.ErrWorkerStopped))
}
Tab inserts a tab. Esc leaves the editor. Ctrl+Enter runs.
Output
TunePool with the limit the worker already has returns ErrSameConcurrency. After StopAndWait, Resume returns ErrWorkerStopped. errors.Is matches both.
The next two sections list every exported error, grouped by how you receive them. The messages are the text each one carries. Four of them are not returned from a call: the worker reports them on Errs() and to the handler.
Returned by calls you make
| Error | Message | Returned when |
|---|---|---|
ErrRunningWorker | worker is already running | Start on a worker that is Idle or Running, such as one just created with auto-run on. Restart when another goroutine restarted the worker first. |
ErrNotRunningWorker | worker is not running | Pause, Resume, Stop, Restart or TunePool on a worker that is Initiated: created with WithAutoRun(false) and not started. The combined calls that end in a pause or stop return it too. |
ErrWorkerPaused | worker is paused | Start or TunePool on a Paused worker. |
ErrWorkerPausing | worker is pausing | Start, Resume or TunePool while the worker is Pausing. |
ErrWorkerStopping | worker is stopping | Start, Pause, Resume or TunePool while the worker is Stopping. |
ErrWorkerStopped | worker is stopped | Pause, Resume or TunePool on a Stopped worker. |
ErrSameConcurrency | worker already has the same concurrency | TunePool with the limit the worker already has. A value below 1 is turned into the CPU count first, and that is what is compared. |
ErrJobProcessing | job is processing, you can't close processing job | Close on a job handle while the job’s function is running. |
ErrJobAlreadyClosed | job is already closed | Close on a job that is already closed, which includes one that has run, since the worker closes those. |
ErrJobCancelled | varmq: job cancelled | Err() or Result() of a job cancelled before it ran, or one cancelled while it ran that then returned an error or panicked. |
ErrCancelUnsupported | varmq: storage can't cancel jobs | Cancel on a job whose storage has no JobCanceller. |
ErrWorkflowExists | varmq: workflow already exists | Submit of a workflow whose ID the store holds with a different plan. The same plan again succeeds. |
ErrWorkflowNotFound | varmq: workflow not found | Cancel, Status or Wait on a workflow the store does not hold, or an invalid ID. |
ErrNoResult | varmq: task has no result | TaskContext.Result for a name that is not a dependency, or a dependency with no result. |
ErrDepFailed | varmq: dependency failed | TaskContext.Result for a dependency that failed or was cancelled. |
ErrResultTooLarge | varmq: task result too large | A task’s result is bigger than the store’s limit (WithResultLimit): the task fails. |
ErrScopeMismatch | varmq: queues are in different storage scopes | wf.Add or Submit with a queue that can’t take workflow tasks, or one in another storage than the workflow store. |
The first six rows are errors that come from the worker’s states. Lifecycle lists, call by call, which states return which error. Concurrency covers TunePool, and jobs covers closing a job.
Sent to Errs and the handler
These four are not returned from any call. The worker reports them when it pulls a job from its storage or closes a finished job. They go to Errs() and to the handler, and the worker continues.
| Error | Message | Reported when |
|---|---|---|
ErrFailedToDequeue | failed to dequeue job | A bound queue’s Dequeue returned no job at the moment the worker pulled from it. The worker checks that a queue is not empty before it pulls, so the built-in queues rarely give it. |
ErrFailedToCastJob | failed to cast job | The storage returned something that is not a job of the worker’s type. The built-in queues hold jobs, so this comes from custom, persistent or distributed storage. |
ErrParseJob | failed to parse job, followed by the decoder’s message | A persistent or distributed queue returned bytes that do not decode as a job. The error wraps ErrParseJob, so match it with errors.Is. |
ErrAcknowledgeJob | queue failed to acknowledge job, followed by the job and acknowledgement IDs | A persistent or distributed queue refused the acknowledgement when the worker closed a job it had taken from it. The error wraps ErrAcknowledgeJob. |
Only persistent and distributed storage hands out bytes to decode or jobs to acknowledge. So ErrParseJob and ErrAcknowledgeJob do not come from the built-in in-memory queues. Persistence covers those queues.
One more error has no variable. The helpers Func, ErrFunc and ResultFunc handle a nil function with the error provided function is nil. ErrFunc and ResultFunc return it; Func panics with it, so the text arrives inside the panic message. It is not exported, so match it with strings.Contains. See function helpers.