Persistence and distribution

An in-memory queue is gone when the process exits. A persistent or distributed queue keeps its jobs in storage 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 the two kinds of queue do 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.

Persistent or distributed

You bind both kinds to a worker the way you bind an in-memory queue. Only a plain worker from NewWorker can bind them. An error worker or a result worker binds in-memory queues only.

	Persistent	Distributed
Bind with	WithPersistentQueue, WithPersistentPriorityQueue	WithDistributedQueue, WithDistributedPriorityQueue
The storage must implement	IPersistentQueue, IPersistentPriorityQueue	IDistributedQueue, IDistributedPriorityQueue: the persistent ones plus ISubscribable
You get back	PersistentQueue[T] or PersistentPriorityQueue[T], which has Worker()	DistributedQueue[T] or DistributedPriorityQueue[T], which has no Worker()
Jobs survive a restart	as long as the storage keeps them	as long as the storage keeps them
The worker learns of new jobs	when something in its own process signals it: an Add, a start or resume, or one of its jobs finishing. With redisq or sqliteq as its only source, it also checks the storage once a second while it last found it empty	when something in its own process signals it, and when the storage notifies it
A process that only adds jobs	not provided	NewDistributedQueue and NewDistributedPriorityQueue wrap the storage with no worker

The row about learning of new jobs is the one that decides which to use. A persistent worker pulls when something in its own process signals it: an Add, a start or resume, or one of its own jobs finishing. When redisq or sqliteq is the worker’s only source, the worker also checks the storage once a second while it is active and last found the storage empty, so it finds a job another process added within about a second. With another source bound as well, nothing signals it when another process adds a job, so it may not look until one of those happens. Adapters has the details. Use a persistent queue when one process wants its jobs to survive a restart.

A distributed queue adds Subscribe. The worker subscribes when you bind the queue, and the adapter calls the worker back with the action "enqueued" when a job is added. That wakes the worker to pull. It is how several processes can consume one queue, with each job going to one of them. Adapters has the rest of the contract.

What is stored

varmq stores each job as JSON, in the shape {"id":"","status":"Created","data":"job-1"}. data is what you passed to Add. That has two consequences:

• The payload type has to survive encoding/json. Add returns false when JSON encoding fails, for example for a function, so function helpers cannot be used 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 what happens to it next is up to the adapter.

Add on these queues returns only a bool. It is false when the encoding fails or when the storage refuses the job, for example because the queue was closed. There is no job handle, so there is nothing to Wait on and no Err or Result to read, and there is no AddAll. Worker types has the table.

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. When the queue is the worker’s only source and implements IBatchAcknowledgeable, as redisq and sqliteq do, it takes a job for each free slot in one DequeueBatchWithAckIds call instead. The storage hands each job over together with an acknowledgement ID, and 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, and closing it calls Acknowledge with that ID. The storage can now forget the job.

Three details about step 3:

• A failed or panicking job is acknowledged too. varmq recovers a panic, counts the job as failed and reports the panic. Then it acknowledges the job like any other. Nothing in varmq retries a job.
• A job whose acknowledgement fails is reported. If Acknowledge 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, because it is unacknowledged and a long wait could let a visibility timeout hand it to another consumer. Call StopAndWait before you close the adapter’s queue, so that running jobs can acknowledge.

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, and what redisq, sqliteq or an adapter of your own does about a job that was taken and not acknowledged.

Crash test · simulation

runs of job 7

0

acknowledged

no

A scene with one job, number 7, and three waiting jobs. The job goes through its steps: waiting, taken, running, returned and acknowledged. In memory the last step is closed, because there is nothing to acknowledge. You pick where the queue lives: in memory, in redisq on Redis, in sqliteq, or in an adapter of your own. Use Step to move the job on, and Kill the process at any step. The scene then shows what the storage holds and what that adapter does with a job that was taken and not acknowledged. In memory, the job and the waiting jobs are lost. With redisq and sqliteq the job comes back; if its function had started, it runs a second time. With your own adapter, the adapter decides.

In memoryredisqsqliteqYour adapter

Redis (redisq)outside the process

queue.Add(job)

StepKill the processReset

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 whose process died mid-run is handed out again, and so is a job whose acknowledgement failed after it ran. Write the job function so that running it twice is safe.
• A job can be lost with some adapters. An adapter that takes a job and records it as in flight in two separate steps can lose it if the process dies between them. redisq and sqliteq do both in one step.
• varmq does not deduplicate for you. WithJobId puts an ID in the stored JSON. Storage that implements IDedupQueue can use it to store a retried Add once, which stops a double add and says nothing about a double run. redisq and sqliteq both implement it.

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.WithPersistentQueue(pq)
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. Unless the adapter signals on bind, as below, nothing signals it again until something in this process does, as listed above, so the old jobs can sit in storage until then. 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.

Metrics

Metrics describes the four counters for in-memory queues. For these queues, keep three things in mind.

• Submitted on a persistent queue counts this process’s accepted Add calls. A job that was already in storage, or that another process added, is not counted there. It still counts in Completed when this worker runs it, so Completed can be higher than Submitted.
• The counters of a worker on a distributed queue may differ from what you expect. Submitted in particular. These docs do not cover how a distributed queue’s counters count yet.
• NumPending asks the storage. It is the number of jobs waiting there, plus jobs the worker has taken and not yet started. The queue handle’s Len leaves out the second kind. Each call is a call to the storage.

Persistent topics are not released

Topics are in-memory only. A persistent topic was built and then moved out of the main branch before release, in #110. It lives on the exp/persistent-topic branch, it is not part of v1.10.0, and these pages do not describe it. To share jobs between processes, use a distributed queue.