Replace boltdb with modernc.org/sqlite

This commit replaces the storage backend of lasr. Now, sqlite is used.

It also removes support for dead-lettering, putting the responsibility
for that in the hands of the user.

The new storage backend comes with a significant performance penalty.
The library is now around 10 times slower. More work will be needed to
achieve parity with the boltdb-backed queue.

Signed-off-by: Eric Chlebek <eric@sensu.io>

Author: Eric Chlebek

README.md

@@ -2,7 +2,7 @@
 [![GoDoc](https://img.shields.io/badge/godoc-reference-blue.svg?style=flat-square)](https://godoc.org/github.com/echlebek/lasr)
 
 # lasr
-A persistent message queue backed by BoltDB. This queue is useful when the producers and consumers can live in the same process.
+A persistent work queue backed by sqlite3. This queue is useful when the producers and consumers can live in the same process.
 
 Project goals
 -------------

acknack.go

@@ -2,77 +2,40 @@ package lasr
 
 import (
 	"sync/atomic"
-
-	"github.com/boltdb/bolt"
 )
 
-func (q *Q) ack(id []byte) error {
-	q.mu.RLock()
-	defer q.mu.RUnlock()
-	var wake bool
-	err := q.db.Update(func(tx *bolt.Tx) error {
-		var err error
-		wake, err = q.stopWaitingOn(tx, id)
-		if err != nil {
-			return err
-		}
-		bucket, err := q.bucket(tx, q.keys.unacked)
-		if err != nil {
-			return err
-		}
-		return bucket.Delete(id)
-	})
-	if err == nil {
-		q.inFlight.Done()
-	}
-	if wake && !q.isClosed() {
-		q.waker.Wake()
-	}
-	return err
-}
+func (q *Q) ack(id ID) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
 
-func (q *Q) nack(id []byte, retry bool) error {
-	q.mu.RLock()
-	defer q.mu.RUnlock()
-	var wake bool
-	err := q.db.Update(func(tx *bolt.Tx) (rerr error) {
-		bucket, err := q.bucket(tx, q.keys.unacked)
-		if err != nil {
-			return err
-		}
-		if retry {
-			wake = true
-			val := bucket.Get(id)
-			ready, err := q.bucket(tx, q.keys.ready)
-			if err != nil {
-				return err
-			}
-			return ready.Put(id, val)
-		}
-		wake, err = q.stopWaitingOn(tx, id)
-		if err != nil {
-			return err
-		}
-		if len(q.keys.returned) > 0 {
-			val := bucket.Get(id)
-			returned, err := q.bucket(tx, q.keys.returned)
-			if err != nil {
-				return err
-			}
-			return returned.Put(id, val)
-		}
-		return bucket.Delete(id)
-	})
+	_, err := q.db.Exec(ackSQL, id)
 	if err != nil {
 		return err
 	}
+
 	q.inFlight.Done()
-	if wake && !q.isClosed() {
+
+	if !q.isClosed() {
 		q.waker.Wake()
 	}
+
 	return nil
 }
 
+func (q *Q) nack(id ID, retry bool) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if retry {
+		_, err := q.db.Exec(nackRetrySQL, id)
+		if err == nil {
+			q.inFlight.Done()
+		}
+		q.waker.Wake()
+		return err
+	}
+	return q.ack(id)
+}
+
 // Ack acknowledges successful receipt and processing of the Message.
 func (m *Message) Ack() (err error) {
 	if !atomic.CompareAndSwapInt32(&m.once, 0, 1) {
@@ -90,61 +53,3 @@ func (m *Message) Nack(retry bool) (err error) {
 	}
 	return m.q.nack(m.ID, retry)
 }
-
-// stopWaitingOn causes all messages waiting on id to not wait on id.
-// If stopWaitingOn finds any messages that were waiting on id that are not
-// waiting on any other messages, it will move them to the Ready state.
-func (q *Q) stopWaitingOn(tx *bolt.Tx, id []byte) (bool, error) {
-	// blocking -> x blocking y
-	// blockedOn -> x blocked on y
-	wake := false
-	blocking, err := q.bucket(tx, q.keys.blocking)
-	if err != nil {
-		return wake, err
-	}
-	blocker := blocking.Bucket(id)
-	if blocker == nil {
-		return wake, nil
-	}
-	blockedOn, err := q.bucket(tx, q.keys.blockedOn)
-	if err != nil {
-		return wake, err
-	}
-	c := blocker.Cursor()
-	for k, _ := c.First(); k != nil; k, _ = c.Next() {
-		blockedMsg := blockedOn.Bucket(k)
-		if blockedMsg == nil {
-			continue
-		}
-		if err := blockedMsg.Delete(id); err != nil {
-			return wake, err
-		}
-		c = blockedMsg.Cursor()
-		if k, _ := c.First(); k != nil {
-			// There are still messages blocking the release of blockedMsg
-			continue
-		}
-		// at this point, nothing is causing 'blockedMsg' to wait anymore.
-		// a message will be placed into the ready state
-		if err := blockedOn.DeleteBucket(k); err != nil {
-			return wake, err
-		}
-		waiting, err := q.bucket(tx, q.keys.waiting)
-		if err != nil {
-			return wake, err
-		}
-		v := waiting.Get(k)
-		ready, err := q.bucket(tx, q.keys.ready)
-		if err != nil {
-			return wake, err
-		}
-		if err := ready.Put(k, v); err != nil {
-			return wake, err
-		}
-		wake = true
-		if err := waiting.Delete(k); err != nil {
-			return wake, err
-		}
-	}
-	return wake, blocking.DeleteBucket(id)
-}

acknack_test.go

@@ -4,13 +4,10 @@ import (
 	"context"
 	"sync"
 	"testing"
-
-	"github.com/boltdb/bolt"
 )
 
 func TestAckConcurrent(t *testing.T) {
-	q, cleanup := newQ(t, WithDeadLetters())
-	defer cleanup()
+	q := newQ(t)
 
 	msg := []byte("foo")
 	if _, err := q.Send(msg); err != nil {
@@ -58,8 +55,7 @@ func TestAckConcurrent(t *testing.T) {
 }
 
 func TestNackConcurrent(t *testing.T) {
-	q, cleanup := newQ(t, WithDeadLetters())
-	defer cleanup()
+	q := newQ(t)
 
 	msg := []byte("foo")
 	if _, err := q.Send(msg); err != nil {
@@ -107,8 +103,7 @@ func TestNackConcurrent(t *testing.T) {
 }
 
 func TestNackDeletesMessage_GH5(t *testing.T) {
-	q, cleanup := newQ(t)
-	defer cleanup()
+	q := newQ(t)
 
 	_, err := q.Send([]byte("foo"))
 	if err != nil {
@@ -124,19 +119,12 @@ func TestNackDeletesMessage_GH5(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = q.db.Update(func(tx *bolt.Tx) error {
-		bucket, err := q.bucket(tx, q.keys.unacked)
-		if err != nil {
-			return err
-		}
-		k, _ := bucket.Cursor().First()
-		if k != nil {
-			t.Error("message remains unacked")
-		}
-		return nil
-	})
-
-	if err != nil {
+	row := q.db.QueryRow("SELECT count(*) FROM queue WHERE id = ?;", msg.ID)
+	var count int
+	if err := row.Scan(&count); err != nil {
 		t.Fatal(err)
 	}
+	if count > 0 {
+		t.Fatalf("bad count: got %d, want 0", count)
+	}
 }