SQLite

sqliteq is the SQLite adapter for varmq. It keeps jobs in SQLite database files, so jobs survive a restart of the process with no server to run. It provides persistent queues only: it has no distributed queue, so WithDistributedQueue does not take it. Read persistence and distribution first for what a persistent queue does.

Setup

Three steps set it up:

1. 1

   Add the two modules

   sqliteq uses mattn/go-sqlite3, which is built with cgo, so building needs CGO_ENABLED=1 and a C compiler.

   terminalCopy

   go get github.com/goptics/varmq github.com/goptics/sqliteq

2. 2

   Open the store

   sqliteq.New takes the path of a folder and returns a Queues value and an error. The folder is created if it does not exist.

   main.goCopy

   sdb, err := sqliteq.New("data")
   if err != nil {
   	panic(err)
   }
   defer sdb.Close()

3. 3

   Create a queue and bind it

   Each queue gets its own database file in the folder, named after the queue in lower case: the queue jobs lives in jobs.db. Both constructors take the name of the queue and return an error as well as the queue.

   First in, first outLowest number first

   main.goCopy

   pq, err := sdb.NewQueue("jobs")
   if err != nil {
   	panic(err)
   }
   defer pq.Close()
   
   queue := worker.WithPersistentQueue(pq)

   main.goCopy

   pq, err := sdb.NewPriorityQueue("priority_jobs")
   if err != nil {
   	panic(err)
   }
   defer pq.Close()
   
   queue := worker.WithPersistentPriorityQueue(pq)
   queue.Add("alert", 1) // a lower number runs first

A queue name may only hold letters, digits, ., _ and -. The files use write-ahead logging.

To keep every queue in one file instead, pass WithSingleFile(): the path is then the database file, with one table for each queue. That is also how to open a file written by an older sqliteq, which kept every queue in one file. New returns an error when the path does not fit the layout, such as a file path without WithSingleFile().

Constructor	Order	Bind it with
NewQueue	first in, first out	WithPersistentQueue
NewPriorityQueue	lowest number first	WithPersistentPriorityQueue

Both take the name of the queue and return an error as well as the queue. Within one process, asking for the same name again returns the same queue. WithRemoveOnComplete(false) is an option for either. It keeps an acknowledged job in the table, marked as completed, where the default deletes it.

A persistent queue

The program below binds a persistent queue to a worker with concurrency 1, adds ten jobs, and waits for them.

package main

import (
  "fmt"
  "time"

  "github.com/goptics/sqliteq"
  "github.com/goptics/varmq"
)

func main() {
  sdb, err := sqliteq.New("data")
  if err != nil {
    panic(err)
  }
  defer sdb.Close()

  pq, err := sdb.NewQueue("jobs")
  if err != nil {
    panic(err)
  }
  defer pq.Close()

  worker := varmq.NewWorker(
    func(j varmq.Job[string]) {
      fmt.Println("processing", j.Data())
      time.Sleep(time.Second)
      fmt.Println("processed", j.Data())
    },
    varmq.WithAutoRun(false),
  )

  queue := worker.WithPersistentQueue(pq)
  fmt.Println("left in the file by an earlier run:", queue.Len())

  // Start after the bind, so the worker looks at jobs already in the file.
  worker.Start()

  for i := range 10 {
    queue.Add(fmt.Sprintf("task %d", i))
  }

  worker.WaitAndStop()
}

Three choices in it matter:

• The worker is created with WithAutoRun(false) and started after the bind. Binding an sqliteq queue already makes the worker look for jobs a previous run left in the file, as the overview says. Starting it later only holds it back while the program counts them.
• queue.Len() is read before Start. It counts the jobs in the table that are waiting, not the ones a worker has taken.
• Every job finishes before the deferred closes run. WaitAndStop lets every job run, then stops the worker.

Run locally

1. Make a module, save the program as main.go and add the dependencies:

   mkdir sqlite-persistent && cd sqlite-persistent
   go mod init sqlite-persistent
   go get github.com/goptics/varmq github.com/goptics/sqliteq

2. Run it with go run .. It creates the folder data, with jobs.db in it. The first run prints left in the file by an earlier run: 0, then a processing line and a processed line for each job, one second apart. Concurrency is 1, so they run in the order they were added.

3. Run it again, and press Ctrl+C a few seconds in.

4. Run it a third time. It prints how many jobs are left, and the worker runs those first, before the ten new ones. The job that was running when you pressed Ctrl+C is one of them, which the next section explains.

Delete the data folder to start from an empty queue. SQLite also keeps jobs.db-wal and jobs.db-shm next to jobs.db while the file is open.

A priority queue

NewPriorityQueue gives a queue that hands out the lowest number first. WithPersistentPriorityQueue binds it, and Add on the handle takes the priority after the data.

package main

import (
  "fmt"

  "github.com/goptics/sqliteq"
  "github.com/goptics/varmq"
)

func main() {
  sdb, err := sqliteq.New("data")
  if err != nil {
    panic(err)
  }
  defer sdb.Close()

  pq, err := sdb.NewPriorityQueue("priority_jobs")
  if err != nil {
    panic(err)
  }
  defer pq.Close()

  worker := varmq.NewWorker(
    func(j varmq.Job[string]) {
      fmt.Println("processing", j.Data())
    },
    varmq.WithAutoRun(false),
  )

  queue := worker.WithPersistentPriorityQueue(pq)

  // A lower number runs first.
  queue.Add("report", 10)
  queue.Add("alert", 1)
  queue.Add("newsletter", 5)

  worker.Start()
  worker.WaitAndStop()
}

Run locally

Make a module the way the persistent queue’s steps do, with this program as main.go and sqlite-priority as the module name, and run it. It creates data/priority_jobs.db and prints:

processing alert
processing newsletter
processing report

Run it again and it reopens the same file, adds three more jobs and prints the same three lines. Jobs with the same priority come out in the order they were added.

Jobs that were not acknowledged

The overview describes when varmq acknowledges a job. This is what sqliteq does with a job that was taken and not acknowledged, for example because the process died.

The scene below shows the table of a queue named jobs. Take a job, then kill the process and restart the program. Or open the file from a second process while the job runs.

Job through SQLite · simulation

A scene that shows the table of a sqliteq queue named jobs, one row per job, with a status of pending, processing or completed. Jobs 7, 8, 9 and 10 are pending. Take a job turns its row into processing in one statement. Acknowledge deletes the row, or marks it completed when you keep completed rows. Kill the process leaves the row processing. Restart the program opens the queue again, and sqliteq sets every processing row back to pending, so the job runs again. A second process that opens the same file does the same while the first process still runs the job, so the job can run twice, and the first process’s acknowledgement then fails.

data/jobs.dbtable jobs

Rows of the jobs table

id	status	data

pq, err := sdb.NewQueue("jobs")

Take a jobAcknowledgeKill the processOpen from a second process WithRemoveOnComplete(false)Reset

• A job has a status in its row. A new row is pending. When the worker takes the job, one statement turns the row into processing. Acknowledge deletes the row, or marks it completed with WithRemoveOnComplete(false). It only matches a row that is processing.
• Taken jobs return to pending when a queue is first opened. The first time a process opens a queue, with NewQueue or NewPriorityQueue, sqliteq sets every processing row that was not acknowledged back to pending. It does this once per queue in each process, and at no other time.
• So a job from a crashed run comes back on the next start. The job that was running when the process died is processing in the file. The next program that creates the queue puts it back, and it runs again. There is no timeout. In the run above, that is the job you interrupted.
• It also puts back jobs that are still running. sqliteq does not know which process took a row. Another process that opens a queue on the same file returns the first process’s running jobs to pending, and they run twice. To avoid that, use one process for each database file.
• Len counts pending rows only. A job that is processing is counted by the worker’s NumProcessing, not by Len.

Other details

• A worker takes several jobs in one statement. sqliteq implements IBatchAcknowledgeable, so a worker with only this queue claims as many jobs as it has free slots in one statement, and each one starts on its own goroutine at once.
• Job IDs are checked. sqliteq implements IDedupQueue and IDedupPriorityQueue. An Add with WithJobId stores nothing while a row with that ID exists: pending, processing, or completed with WithRemoveOnComplete(false). Once the job is acknowledged and its row deleted, the ID can be added again.
• Purge deletes every row. That includes the processing row of a job that is running. When the job finishes, its Acknowledge finds no row and returns false, and the worker reports ErrAcknowledgeJob to Errs() and the handler. The job’s function has run, so the error is about the bookkeeping.
• There is no capacity. sqliteq does not implement CapacitySetter, so WithQueueCapacity does not make Add refuse jobs.
• Close only stops the queue. It makes later Enqueue and Dequeue calls on that queue fail, for every holder of it in the process. It does not close the database. Close on the Queues value does.