Persistence and distribution

An in-memory queue is gone when the process exits. A queue in shared storage keeps its jobs outside the process, behind a storage adapter. VarMQ has no adapter of its own. The two that exist are separate modules: redisq for Redis and sqliteq for SQLite.

This page covers what such a queue does in VarMQ. Redis and SQLite show the setup and what each adapter does with a job that was not finished. Adapters lists the interfaces, for reading an adapter or writing one.

Pick an adapter

• Redis (redisq): use it when several processes share one queue. It needs a Redis server.
• SQLite (sqliteq): use it for one process with no server to run.

Installation says which adapter versions have these features. Adapters has the full list of interfaces.

Guarantees at a glance

What this guarantees explains each answer.

Persistent or distributed

You bind shared storage the way you bind your own queue: with WithQueue or WithPriorityQueue. A plain worker, an error worker and a result worker can all bind it. VarMQ sees that the storage implements IAcknowledgeable, and treats it as shared storage.

queue := worker.WithQueue(rq) // rq from redisq or sqliteq
job, ok := queue.Add("job 1")

A process that only adds jobs uses NewQueue or NewPriorityQueue. It needs no worker. For the jobs of an error or result worker, use NewErrQueue or NewResultQueue, so the handle has Err or Result. Earlier releases had other binding methods for this storage. They are deprecated, and their API reference names the call to use.

The worker learns of new jobs in one of two ways:

• Storage with Subscribe, such as redisq, calls the worker back when any process adds a job. That wakes the worker to pull. It is how several processes can consume one queue, with each job going to one of them. Such storage is also called a distributed queue.
• Storage without it, such as sqliteq, is a persistent queue. The worker pulls when something in its own process signals it: an Add, a start or resume, or one of its own jobs finishing.

Sometimes storage without Subscribe is the worker’s only source. Then the worker also checks the storage once a second, while it is active and last found the storage empty. So it finds a job that another process added within about a second. With another source bound as well, it may not look until one of the signals above happens. Adapters has the details.

What is stored

VarMQ stores each job as JSON, in the shape {"id":"…","status":"Created","data":"job-1","enqueued_at":1794229533324}. data is what you passed to Add. enqueued_at is the time of the Add, in Unix milliseconds. That has two consequences:

• The payload type has to work with encoding/json. Add returns false when JSON encoding fails, for example for a function. So you cannot use function helpers here.
• Producers and consumers have to agree on the type. A consumer decodes the stored JSON into its own T. Bytes that do not decode are reported as ErrParseJob on Errs() and to the error handler. The worker does not acknowledge such a job, so the adapter decides what happens to it next.

Add returns a job handle and a bool. The bool is false when the encoding fails, or when the storage refuses the job, for example because the queue was closed. AddAll works too.

The job handle learns how the job ended when the storage reports outcomes. redisq and sqliteq do. Jobs in other processes covers Wait, Err, Result and the errors they return. When you give a job no ID, VarMQ makes a random one, so the outcome and a cancel can find the job. It does this when the storage reports outcomes or cancels jobs.

The handle’s Len asks the storage, so it counts the jobs waiting there, including jobs another process added. WithQueueCapacity reaches the storage only if the adapter implements CapacitySetter. Neither redisq nor sqliteq does. With them Add is not refused, and IsFull only compares Len with the number you passed.

Acknowledgement

An in-memory job is done when its function returns. A job in storage needs one more step, so that storage knows the job was not lost on the way.

1. When the worker has a free slot, it takes a job with DequeueWithAckId. Some queues are the worker’s only source and implement IBatchAcknowledgeable, as redisq and sqliteq do. For such a queue, the worker takes a job for each free slot in one DequeueBatchWithAckIds call instead. The storage gives each job together with an acknowledgement ID. It keeps its own record that the job is out.
2. The job’s function runs.
3. When the function returns, the worker closes the job. Closing it reports the outcome with that ID: Acknowledge when the job succeeded, Reject with the error text when it failed. A cancelled job is reported as cancelled, when the storage supports it. The storage can now forget the job, or keep a record of it.

Four details about step 3:

• A failed or panicking job is rejected. VarMQ recovers a panic, counts the job as failed and reports the panic. Then it calls Reject, which removes the job from storage like Acknowledge does. The job does not run again, unless Reject fails: then the storage can give it out again (see what this guarantees). Nothing in VarMQ retries a job, so every failure is final.
• A cancelled job is never retried either. It ends as cancelled. Cancelling a job covers it.
• A job whose acknowledgement fails is reported. If Acknowledge or Reject returns false, the worker sends ErrAcknowledgeJob to Errs() and the handler. The job’s function has already run.
• Taken jobs run. A worker with these queues takes one job per free slot and starts it. A Pause or Stop does not park a job the worker already took. The job is unacknowledged, and after a long wait a visibility timeout could give it to another consumer. A visibility timeout is the time some storage, such as redisq, waits for an acknowledgement. After that time, the storage gives the job out again. See shutting down.

The scene below runs one job through these steps. Pick where the queue lives, then press Kill the process at any step. It shows what the storage holds. It also shows what redisq, sqliteq or your own adapter does with a job that was taken and not acknowledged.

What this guarantees

VarMQ acknowledges a job it took from storage after the job’s function has returned, and not before. So if the process dies while a job runs, the job was not acknowledged.

What the storage then does with it is the adapter’s decision. VarMQ cannot make it come back. Redis and SQLite say what redisq and sqliteq do. With either adapter the best case is at-least-once delivery:

• A job can run twice. With these adapters, a job is given out again if its process died while it ran. A job is also given out again if its acknowledgement failed after it ran. Write the job function so that running it twice is safe.
• A job can be lost with some adapters. Some adapters take a job and record it as in progress in two separate steps. Such an adapter can lose the job if the process dies between the steps. redisq and sqliteq do both in one step.
• VarMQ does not deduplicate for you. WithJobId puts an ID in the stored JSON. Storage can use it in EnqueueOnce to store a retried Add once. This stops a double add. It says nothing about a double run. redisq and sqliteq both do.

Starting a worker on a non-empty queue

After a restart, jobs from the earlier run are still in storage. Bind the queue first and start the worker afterwards:

worker := varmq.NewWorker(process, varmq.WithAutoRun(false))
queue := worker.WithQueue(rq)
worker.Start()

A worker that starts on creation, which is the default, signals its event loop to look for work before you have bound the queue. If the loop reads that signal first, it finds no queue. Some adapters signal on bind, as below. Without that, nothing signals the loop again until something in this process does, as listed above. Until then, the old jobs can stay in storage. With WithAutoRun(false), Start runs after the bind and signals the loop itself. Lifecycle covers the option.

Adapters that implement IBatchAcknowledgeable signal the worker when you bind them. redisq and sqliteq both do, so with them the order does not matter.

Shutting down

Close things in this order, so that every running job can acknowledge:

1. Stop adding jobs.
2. Call StopAndWait on the worker. Running jobs finish and acknowledge. To run every waiting job first, call WaitAndStop, then WaitUntilStopped.
3. Call Close on the adapter’s queue. With redisq, Close puts back every job that queue object took and did not acknowledge, so a job that still runs would run again.
4. Call Close on the adapter’s Queues value.

With defer, the order is the reverse of the calls. Defer the Queues close first, then the queue close. The programs on Redis and SQLite do this.

Pausing

A queue pause on shared storage is kept in the storage. Every process that takes from the queue stops, and the pause survives a restart. Storage that cannot pause returns ErrPauseUnsupported.

Metrics

Metrics describes the three counters of a worker. For these queues, note three things.

• Each worker counts the jobs it ran. A job that another process ran counts in that process. So the counters of one process do not add up to the queue’s work.
• The queue counts the jobs it received. Read totals.submitted from GET /queues. It counts jobs that every process added.
• NumPending asks the storage. It is the number of jobs waiting there, plus jobs the worker has taken and not yet started. Each call is a call to the storage. A paused queue’s jobs do not count.

Topics

Topics are in-memory only. To share jobs between processes, use shared storage such as redisq.

Next: Redis.