safequeue

A durable, idempotent job queue for Go with rate limiting, priority, backpressure, and fair scheduling built in — a small library plus an optional HTTP service. Standard library only; no third-party dependencies.

The problem it solves: a naive in-memory channel loses work on crash, redelivers nothing, has no retry policy, and lets one noisy producer starve everyone else. safequeue gives you at-least-once durability and the admission-control machinery you normally have to bolt on afterwards — in one dependency-free package.

• At-least-once, durable. Every state change is appended to a log and fsync'd; reopening the data directory replays the log to rebuild state. A torn final record (partial write before a crash) is tolerated.
• Idempotent enqueue. Enqueuing the same idempotency key twice returns the original message instead of creating a duplicate.
• Priority delivery. Enqueue with a priority; higher priority is dequeued first, FIFO within a band. Default priority 0 reduces to plain FIFO.
• Backpressure. Bounded capacity with pluggable policy: fail fast (ErrBackpressure / HTTP 429), shed the oldest, or block with a context.
• Rate limiting. Thread-safe token-bucket and leaky-bucket limiters with Allow / Wait(ctx) / Reserve semantics and an injectable clock.
• Fair scheduling. Deficit Round Robin across per-tenant keys so a flood from one producer cannot starve others; weights control relative throughput.
• Observability. A tiny metrics interface with an expvar adapter, plus Depth() and backpressure counters in Stats().

Maintainer: Cognis Digital · License: COCL 1.0

Install

go get github.com/cognis-digital/safequeue/queue

Module: github.com/cognis-digital/safequeue (Go 1.22+). Cross-platform (Windows, macOS, Linux). Build the service binary with the Makefile, install.sh (macOS/Linux), or install.ps1 (Windows); a Dockerfile is provided for containers.

Packages

Import path	What it provides
.../safequeue/queue	Durable, idempotent at-least-once queue + priority + backpressure
.../safequeue/ratelimit	Token-bucket and leaky-bucket limiters
.../safequeue/scheduler	Deficit Round Robin fair scheduler
.../safequeue/metrics	Minimal metrics interface + expvar adapter
.../safequeue/config	JSON config loader + validation (schema in config/)
.../safequeue/httpapi	JSON/HTTP service

Library usage

package main

import (
	"fmt"
	"log"

	"github.com/cognis-digital/safequeue/queue"
)

func main() {
	q, err := queue.Open("./data", queue.Options{
		Capacity:           10000,             // backpressure at 10k live messages
		BackpressurePolicy: queue.PolicyReject, // fail fast when full
	})
	if err != nil {
		log.Fatal(err)
	}
	defer q.Close()

	// Producer: enqueue with an idempotency key and a priority.
	msg, created, err := q.EnqueueWith("send-welcome:user-42", "user-42",
		queue.EnqueueOptions{Priority: 5, FairnessKey: "tenant-a"})
	if err == queue.ErrBackpressure {
		return // queue full: slow down / retry
	}
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println(msg.ID, "created:", created)

	// Consumer: lease a message, process it, then ack or nack.
	m, err := q.Dequeue()
	if err == queue.ErrEmpty {
		return
	}
	if processOK(m) {
		q.Ack(m.ID)
	} else {
		q.Nack(m.ID) // retry with backoff, or dead-letter if exhausted
	}
}

func processOK(*queue.Message) bool { return true }

Rate limiting

import "github.com/cognis-digital/safequeue/ratelimit"

// 500 events/sec sustained, bursts up to 100.
tb := ratelimit.NewTokenBucket(500, 100, ratelimit.RealClock{})

if tb.Allow() { /* proceed now */ }
_ = tb.Wait(ctx)          // block until a token is available or ctx is done
r := tb.Reserve()         // commit a slot; r.Delay() says how long to wait

LeakyBucket has the same shape but enforces a strictly even output rate instead of permitting bursts.

Fair scheduling

import "github.com/cognis-digital/safequeue/scheduler"

d := scheduler.New(1)     // quantum 1 => per-item interleaving
d.SetWeight("premium", 3) // 3x throughput of a weight-1 tenant when both busy
d.Add("premium", job)
d.Add("free", job)
item, _ := d.Next()       // returns the next item under DRR fairness

Options (durable core)

Field	Default	Meaning
VisibilityTimeout	30s	How long a leased message stays invisible before redelivery.
MaxAttempts	5	Delivery attempts before a message is dead-lettered.
BaseBackoff	1s	First retry delay; doubles each attempt up to MaxBackoff.
MaxBackoff	5m	Upper bound on the backoff delay.
Capacity	0	Bounded live (ready+leased) messages; 0 = unbounded.
BackpressurePolicy	Reject	At capacity: Reject (429), DropOldest (shed), Block.
Metrics	no-op	A metrics.Recorder sink; injectable for expvar/OTel.
Clock	real	Time source; injectable for deterministic tests.

Benchmarks

Real numbers from go test -bench on this machine — AMD Ryzen 9 5950X, Go 1.26.4, windows/amd64. Reproduce with make bench. Your hardware and disk will differ; the durable enqueue path is dominated by fsync, so it reflects your storage, not just CPU.

Benchmark	ns/op	Notes
TokenBucket.Allow	~13 ns/op	0 allocs; ~76M ops/sec single-threaded
TokenBucket.Allow (parallel)	~45 ns/op	under contention across 32 goroutines
LeakyBucket.Allow	~13 ns/op	0 allocs
scheduler.DRR Add+Next	~190 ns/op	1 alloc; 8 keys
queue.Enqueue (durable, fsync)	~1.45 ms/op	fsync-bound (~690 durable enqueues/sec on this SSD)

The limiter and scheduler are CPU-bound and fast. The durable enqueue is deliberately fsync-per-op for crash safety; batch/group-commit is a documented roadmap item (see issues) for callers who want higher durable throughput.

HTTP service

go run ./cmd/safequeue -addr :8080 -data ./data -capacity 10000 -backpressure reject
# or from a config file:
go run ./cmd/safequeue -config ./config/config.example.json -metrics

Flags: -addr, -data, -visibility, -max-attempts, -base-backoff, -max-backoff, -capacity, -backpressure (reject|drop_oldest|block), -config, -metrics.

Endpoints

Method & path	Request body	Response
POST /enqueue	{"payload":"...","idempotency_key":"...","priority":N,"fairness_key":"..."}	{"message":{...},"created":bool} (429 if at capacity)
POST /dequeue	(none)	200 leased message, or 404 if empty
POST /ack	{"id":"..."}	{"ok":true} (404/409 on misuse)
POST /nack	{"id":"..."}	{"dead":bool} (404/409 on misuse)
GET /stats	(none)	counts by state + backpressure counters
GET /depth	(none)	{"current":N,"high_water":N}
GET /healthz	(none)	{"ok":true}
GET /debug/vars	(none, only with -metrics)	expvar dump incl. safequeue metrics

Status codes: 400 invalid/missing fields, 404 unknown id or empty queue, 405 wrong method, 409 message not leased, 429 backpressure (at capacity).

Configuration file

config/config.schema.json is a JSON Schema (Draft 2020-12) for the config file; config/config.example.json is a working example. The loader rejects unknown fields and validates value ranges and cross-field invariants (e.g. base_backoff must not exceed max_backoff).

Delivery semantics

1. Enqueue adds a ready message. A non-empty idempotency key that already maps to a live message returns that message (no duplicate). If Capacity is set and reached, BackpressurePolicy decides: reject, drop-oldest, or block.
2. Dequeue picks the highest-priority ready message whose visibility time has passed (FIFO within a priority band), leases it for VisibilityTimeout, increments Attempts, and returns a copy.
3. Ack deletes the leased message; its idempotency key is freed.
4. Nack schedules a retry after exponential backoff, or dead-letters the message once Attempts reaches MaxAttempts.
5. Visibility timeout. A leased message not acked/nacked before its lease expires is reclaimed to ready (at-least-once — make consumers idempotent).

Design docs

See docs/ARCHITECTURE.md for the layered design (durable core + rate/priority/backpressure/fairness), the fairness algorithm, and the storage/wire format.

Examples

Runnable demos under examples/ (each exits 0):

go run ./examples/priority   # priority-ordered delivery
go run ./examples/ratelimit  # token-bucket smoothing a burst
go run ./examples/fairness   # DRR fair scheduling / no starvation

Development

make build   # go build ./...
make test    # go test ./... -race -count=1   (requires cgo for -race)
make bench   # go test -bench . -benchmem
make lint    # gofmt -l . (must be empty) + go vet ./...

Tests inject a fake clock and use millisecond backoffs, so the suite never sleeps on the real wall clock (the limiter/scheduler tests are fully deterministic).

License

License: COCL 1.0. See the canonical COCL text distributed with Cognis Digital projects.