BlockQueue

A durable, embeddable message queue for Go, backed by SQLite or PostgreSQL.

BlockQueue is an embeddable, at-least-once message queue for Go backed by SQLite or PostgreSQL. It stores one canonical message and one delivery row per subscriber, so fan-out state, retries, leases, DLQ transitions, and schedules remain transactional.

Turso/libSQL support is experimental.

[!IMPORTANT] v0.2.0 is a clean schema break and requires a new, empty database. It does not perform an in-place database upgrade from v0.1.

What v0.2.0 provides

• Durable publish is the default in the Go API; explicit async publish is available when admission latency matters more than crash durability.
• A weighted writer budget limits both pending message count and bytes. A reservation is released only after the database transaction finishes.
• Claims use database locking, a new receipt token for every delivery lease, idempotent ACK, fenced stale receipts, delayed NACK, snooze, cancellation, and lease extension.
• PublishTx, AckDeliveryTx, and the other *Tx methods can commit queue state atomically with application tables in the same SQLite or PostgreSQL database.
• Delivery claims and processing failures are counted separately. Subscriber retry policy supports exponential backoff with bounded deterministic jitter, and every failure is retained in a paginated error history.
• Priority, delayed delivery, absolute RFC3339 scheduling, recurring five-field cron schedules, IANA timezones, run history, and overlap protection.
• Transactional, checksummed embedded schema migrations.
• /livez, /readyz, bounded retention, adaptive SQLite checkpoints, and
• /livez, /readyz, an embedded OpenAPI 3.1 document, RFC 9457 problem responses, bounded retention, adaptive SQLite checkpoints, and optional Prometheus metrics.

There is one queue engine and one current HTTP contract at /v1; the project does not maintain parallel v1/v2 engines or schemas. See the v0.2 migration guide for source-level changes and fresh-database rollout instructions.

Install

go get github.com/yudhasubki/blockqueue

The server binary is optional:

go build -o blockqueue ./cmd/blockqueue

Embed in a Go application

Only the root package and the selected storage driver are required:

package main

import (
	"context"
	"log"
	"time"

	"github.com/yudhasubki/blockqueue"
	"github.com/yudhasubki/blockqueue/store/sqlite"
)

func main() {
	driver, err := sqlite.Open("blockqueue.db", sqlite.Config{})
	if err != nil {
		log.Fatal(err)
	}

	queue := blockqueue.New(driver, blockqueue.Options{})
	if err := queue.Run(context.Background()); err != nil {
		log.Fatal(err)
	}
	defer queue.Close()

	topic := blockqueue.NewTopic("orders")
	worker := blockqueue.NewSubscriber(topic, "fulfillment", blockqueue.SubscriberOptions{
		MaxAttempts:        5,
		VisibilityDuration: "30s",
		DequeueBatchSize:   10,
	})
	if err := queue.CreateTopic(context.Background(), topic, blockqueue.Subscribers{worker}); err != nil {
		log.Fatal(err)
	}

	receipt, err := queue.Publish(context.Background(), topic, blockqueue.Message{
		Message:        `{"order_id":"1022"}`,
		IdempotencyKey: "order-1022",
		Priority:       10,
	})
	if err != nil {
		log.Fatal(err)
	}
	log.Printf("persisted %s", receipt.MessageID)

	deliveries, err := queue.ClaimWait(context.Background(), topic, worker.Name, 10, time.Minute)
	if err != nil {
		log.Fatal(err)
	}
	for _, delivery := range deliveries {
		// Process the message before acknowledging this exact lease.
		if err := queue.AckDelivery(context.Background(), topic, worker.Name,
			delivery.ID, delivery.ReceiptToken); err != nil {
			log.Fatal(err)
		}
	}
}

Publish and BatchPublish wait for commit. PublishAsync and BatchPublishAsync return after bounded in-memory admission. If a durable caller's context expires after admission, it receives *CommitUnknownError with stable message IDs; the writer continues owning the admitted messages.

Transactional enqueue and completion

Use WithTx when application tables and BlockQueue use the same database. The producer can atomically create application data and enqueue later work:

err := queue.WithTx(ctx, nil, func(tx *sql.Tx) error {
	if err := insertOrder(ctx, tx, orderID, "pending"); err != nil {
		return err
	}
	_, err := queue.PublishTx(ctx, tx, topic, blockqueue.Message{
		Message:        `{"order_id":"1022","action":"fulfill"}`,
		IdempotencyKey: "fulfill-order-1022",
	})
	return err
})

That commit does not include consumer execution. The consumer claims the committed delivery later, then can atomically store its result and complete the delivery:

err := queue.WithTx(ctx, nil, func(tx *sql.Tx) error {
	if err := markOrderFulfilled(ctx, tx, orderID); err != nil {
		return err
	}
	return queue.AckDeliveryTx(
		ctx, tx, topic, worker.Name, delivery.ID, delivery.ReceiptToken,
	)
})

PublishTx returns state: staged; the rows become visible only if the transaction commits. AckDeliveryTx, NackDeliveryTx, SnoozeDeliveryTx, CancelDeliveryTx, and CancelMessageTx provide the same atomic boundary for consumer side effects. Keep callbacks short and free of network calls. SQLite has one writer, so an open caller transaction intentionally blocks queue writes until commit or rollback. PostgreSQL uses a shared topology fence, so publishers can proceed concurrently while destructive subscriber mutations wait for the transaction. Shutdown drains transactions created by WithTx; callers that begin a raw transaction themselves must coordinate its lifetime with shutdown.

The runnable transactional example shows both commits against one SQLite database, including receipt-fenced consumer completion.

Run the HTTP server

Copy config.yaml.example, then run:

./blockqueue migrate -config config.yaml
./blockqueue http -config config.yaml

Queue.Run also applies migrations, so the explicit migration command is optional for embedded deployments.

Configuration decoding rejects unknown YAML fields and expands ${NAME} from the process environment. This keeps PostgreSQL passwords out of committed configuration files; unset variables expand to an empty value and normal connection validation still applies.

Create a topic:

curl -X POST http://127.0.0.1:8080/v1/topics \
  -H 'Content-Type: application/json' \
  -d '{
    "name":"orders",
    "subscribers":[{
      "name":"fulfillment",
      "option":{
        "max_attempts":5,
        "visibility_duration":"30s",
        "dequeue_batch_size":10
      }
    }]
  }'

Async publish is the HTTP default and returns 202 with state: admitted:

curl -X POST http://127.0.0.1:8080/v1/topics/orders/messages \
  -H 'Content-Type: application/json' \
  -d '{"message":"order-1022","idempotency_key":"order-1022","priority":10}'

Wait for commit with ?wait_for=commit; this returns a definitive duplicate result:

curl -X POST 'http://127.0.0.1:8080/v1/topics/orders/messages?wait_for=commit' \
  -H 'Content-Type: application/json' \
  -d '{"message":"order-1022","idempotency_key":"order-1022"}'

Claim and ACK the returned receipt token:

curl -X POST 'http://127.0.0.1:8080/v1/topics/orders/subscribers/fulfillment/claim?timeout=30s&limit=10'

curl -X POST http://127.0.0.1:8080/v1/topics/orders/subscribers/fulfillment/messages/MESSAGE_ID/ack \
  -H 'Content-Type: application/json' \
  -d '{"receipt_token":"RECEIPT_TOKEN"}'

JSON decoding is strict. The HTTP limits are 1 MiB per message, 1,000 messages per batch, 16 MiB per request body, 16 KiB of headers, and 128 bytes per idempotency key.

The complete OpenAPI 3.1 contract is served at /openapi.json. Errors use application/problem+json with stable code values. The HTTP surface exposes message status, delivery/message cancellation, snooze, and delivery error history. Database transaction methods are intentionally Go-only: a remote HTTP request cannot join the caller's local transaction; cross-database systems should publish through an application outbox.

Delivery contract

• Delivery is at-least-once. Consumers must make side effects idempotent.
• A claim owns a delivery only for its current receipt token and lease. ACK, NACK, and lease extension reject stale receipts.
• delivery_count increases when a lease is claimed; failure_count increases only on NACK or lease expiry. Dead-lettering is based on failures, with three failures and exponential retry delay as the default.
• Snooze returns a claimed delivery to pending without consuming a failure. Cancellation is terminal and idempotent, and failure records remain queryable while the delivery is retained.
• Publish waits for the canonical message and all subscriber delivery rows to commit. PublishAsync guarantees bounded process-local admission, not crash durability.
• The database is authoritative. PostgreSQL notifications and in-memory wakeups reduce latency but are never required for correctness.
• Built-in authentication and exactly-once execution are outside the project scope. Protect the HTTP server with a private network or reverse proxy.

v0.2 is the durable embedded/HTTP fan-out core. A typed Go worker runtime and multi-node maintenance leader election are intentionally roadmap items rather than hidden behavior in this release.

Storage and durability

Set store.DurabilityBalanced explicitly when lower latency is more important than the strict default. Async admission has no local disk spool; use durable publish when a successful response must imply a committed transaction. Processed deliveries are retained for seven days and schedule-run history for 30 days by default. Dead letters are retained indefinitely unless Options.DeadLetterRetention is set explicitly.

Development

go test ./...
go test -race ./...
go vet ./...
go test -run '^$' -bench . -benchmem ./...

The SQLite and PostgreSQL contract suite is shared. PostgreSQL tests create a random schema per run, remove it during cleanup, and refuse a database whose name does not end in _test:

createdb blockqueue_test
BLOCKQUEUE_TEST_POSTGRES_URL='postgres://postgres:postgres@127.0.0.1:5432/blockqueue_test?sslmode=disable' \
  go test -count=1 ./...

CI runs the complete suite and race detector against both storage backends, plus vet and a guarded PostgreSQL benchmark smoke. Benchmark scenarios and exact persisted-row checks are documented in benchmark/README.md.

For component boundaries and lock ownership, see docs/architecture.md.

Roadmap

• v0.2.x: optional in-process event subscription, tracing hooks, and focused test helpers without changing the schema.
• v0.3: typed Go workers and PostgreSQL maintenance leader election.
• v0.4: versioned workflow/DAG orchestration as a separate layer over the queue.

License

Apache License 2.0.