LedgerQ

A disk-backed message queue for single-node applications - written in pure Go with zero dependencies.

Feature-complete as of v1.4.0 - LedgerQ focuses on being excellent at local message queuing rather than trying to replace distributed systems like Kafka.

Features

• Append-only log that reopens at its last complete entry after a crash
• Zero dependencies beyond the Go standard library
• High throughput with batch operations
• Priority queue support (v1.1.0+) with starvation prevention
• Dead Letter Queue (DLQ) support (v1.2.0+) for failed message handling
• Payload compression (v1.3.0+) with GZIP to reduce disk usage
• Message deduplication (v1.4.0+) for exactly-once semantics
• Replay from message ID or timestamp
• Message TTL and headers
• Metrics and pluggable logging
• Automatic compaction with retention policies
• Thread-safe concurrent access
• CLI tool for management

Why LedgerQ?

Single-node message queue on disk. No network, no clustering, no external dependencies. If you're building CLI tools, desktop apps, or edge devices that need task queues on disk, LedgerQ stores messages on disk using an append-only log.

When to Use LedgerQ

Perfect for:

• Offline-first applications - Mobile sync, desktop apps that work without network
• Local task processing - CI/CD pipelines, backup jobs, batch processing
• Edge/IoT devices - Limited connectivity, single-node operation
• Event sourcing - Local audit trails and event logs
• Development/testing - Simpler than running Kafka/RabbitMQ locally
• Embedded systems - Zero external dependencies, small footprint

When NOT to Use LedgerQ

Use these instead:

• Need multiple consumers on different machines? → Use Kafka (distributed streaming)
• Need pub/sub fan-out patterns? → Use NATS or RabbitMQ (message broker)
• Need distributed consensus? → Use Kafka or Pulsar (distributed log)
• Need cross-language support via HTTP? → Use RabbitMQ or ActiveMQ (AMQP/STOMP)
• Already using a database? → Use PostgreSQL LISTEN/NOTIFY or Redis Streams

Comparison

Feature	LedgerQ	Kafka	NATS	RabbitMQ
Deployment	Single binary	Cluster (ZooKeeper/KRaft)	Single/Cluster	Single/Cluster
Dependencies	None	Java, ZooKeeper	None	Erlang
Consumer Model	Single reader	Consumer groups	Subjects/Queues	Queues/Exchanges
Use Case	Local queuing	Distributed streaming	Lightweight messaging	Enterprise messaging
Setup Time	Instant	Hours	Minutes	Minutes

LedgerQ's sweet spot: You need reliability without the operational overhead of distributed systems.

Trade-offs

• ✅ Zero config, just a directory path
• ✅ Reopens after a crash at the last complete entry
• ✅ Fast sequential I/O (append-only)
• ✅ No network protocols or ports
• ✅ Embeddable in any Go application
• ❌ Single process only (file locking)
• ❌ Buffered by default: a crash can lose acknowledged messages. AutoSync prevents that, at about 27 writes a second
• ❌ Not distributed (single node)
• ❌ Single consumer (FIFO order)

Quick Start

Installation

go get github.com/1mb-dev/ledgerq/pkg/ledgerq@latest

Basic Usage

package main

import (
    "fmt"
    "log"

    "github.com/1mb-dev/ledgerq/pkg/ledgerq"
)

func main() {
    q, err := ledgerq.Open("/tmp/myqueue", nil)
    if err != nil {
        log.Fatal(err)
    }
    defer q.Close()

    offset, _ := q.Enqueue([]byte("Hello, World!"))
    fmt.Printf("Enqueued at offset: %d\n", offset)

    msg, _ := q.Dequeue()
    fmt.Printf("Dequeued [ID:%d]: %s\n", msg.ID, msg.Payload)
}

Configuration

opts := ledgerq.DefaultOptions("/tmp/myqueue")
opts.AutoSync = true
opts.MaxSegmentSize = 100 * 1024 * 1024
opts.CompactionInterval = 5 * time.Minute

q, _ := ledgerq.Open("/tmp/myqueue", opts)

Core Operations

Batch operations (10-100x faster):

payloads := [][]byte{[]byte("msg1"), []byte("msg2"), []byte("msg3")}
offsets, _ := q.EnqueueBatch(payloads)

messages, _ := q.DequeueBatch(10)

Message TTL:

q.EnqueueWithTTL([]byte("temporary"), 5*time.Second)

Message headers:

headers := map[string]string{"content-type": "application/json"}
q.EnqueueWithHeaders(payload, headers)

Payload compression (v1.3.0+):

// Enable compression by default
opts := ledgerq.DefaultOptions("/tmp/myqueue")
opts.DefaultCompression = ledgerq.CompressionGzip
opts.CompressionLevel = 6  // 1 (fastest) to 9 (best compression)
opts.MinCompressionSize = 1024  // Only compress >= 1KB
q, _ := ledgerq.Open("/tmp/myqueue", opts)

// Messages are automatically compressed/decompressed
q.Enqueue(largePayload)  // Compressed if >= 1KB

// Or control compression per-message
q.EnqueueWithCompression(payload, ledgerq.CompressionGzip)

Message deduplication (v1.4.0+):

// Enable deduplication with a 5-minute window
opts := ledgerq.DefaultOptions("/tmp/myqueue")
opts.DefaultDeduplicationWindow = 5 * time.Minute
opts.MaxDeduplicationEntries = 100000  // Max 100K unique messages tracked
q, _ := ledgerq.Open("/tmp/myqueue", opts)

// Enqueue with deduplication
offset, isDup, _ := q.EnqueueWithDedup(payload, "order-12345", 0)
if isDup {
    fmt.Printf("Duplicate detected! Original at offset %d\n", offset)
} else {
    fmt.Printf("New message enqueued at offset %d\n", offset)
}

// Custom per-message window
q.EnqueueWithDedup(payload, "request-789", 10*time.Minute)

// View deduplication statistics
stats := q.Stats()
fmt.Printf("Tracked entries: %d\n", stats.DedupTrackedEntries)

Priority queue (v1.1.0+):

// Enable priority mode (set at queue creation, cannot be changed later)
opts := ledgerq.DefaultOptions("/tmp/myqueue")
opts.EnablePriorities = true
q, _ := ledgerq.Open("/tmp/myqueue", opts)

// Enqueue with priority
q.EnqueueWithPriority([]byte("urgent"), ledgerq.PriorityHigh)
q.EnqueueWithPriority([]byte("normal"), ledgerq.PriorityMedium)
q.EnqueueWithPriority([]byte("background"), ledgerq.PriorityLow)

// Dequeue in priority order (High → Medium → Low)
msg, _ := q.Dequeue() // Returns "urgent" first

Batch operations with priorities (v1.1.0+):

// Batch enqueue with per-message options
messages := []ledgerq.BatchEnqueueOptions{
    {Payload: []byte("urgent task"), Priority: ledgerq.PriorityHigh},
    {Payload: []byte("normal task"), Priority: ledgerq.PriorityMedium},
    {Payload: []byte("background task"), Priority: ledgerq.PriorityLow, TTL: time.Hour},
}
offsets, _ := q.EnqueueBatchWithOptions(messages)

Dead Letter Queue (v1.2.0+):

// Enable DLQ with max retries
opts := ledgerq.DefaultOptions("/tmp/myqueue")
opts.DLQPath = "/tmp/myqueue/dlq"
opts.MaxRetries = 3
q, _ := ledgerq.Open("/tmp/myqueue", opts)

// Process messages with Ack/Nack
msg, _ := q.Dequeue()

// Check retry info for custom backoff logic
if info := q.GetRetryInfo(msg.ID); info != nil {
    backoff := time.Duration(1<<uint(info.RetryCount)) * time.Second
    time.Sleep(backoff) // Exponential backoff
}

if processSuccessfully(msg.Payload) {
    q.Ack(msg.ID) // Mark as successfully processed
} else {
    q.Nack(msg.ID, "processing failed") // Record failure (moves to DLQ after MaxRetries)
}

// Inspect DLQ messages
dlq := q.GetDLQ()
dlqMsg, _ := dlq.Dequeue()
fmt.Printf("Failed: %s, Reason: %s\n",
    string(dlqMsg.Payload),
    dlqMsg.Headers["dlq.failure_reason"])

// Requeue from DLQ after fixing issues
q.RequeueFromDLQ(dlqMsg.ID)

Note: Nack() tracks failures but does not auto-retry messages. After MaxRetries failures, messages move to DLQ. For retry patterns with backoff, see USAGE.md.

Streaming:

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

q.Stream(ctx, func(msg *ledgerq.Message) error {
    fmt.Printf("Received: %s\n", msg.Payload)
    return nil
})

Replay:

q.SeekToMessageID(100)
q.SeekToTimestamp(time.Now().Add(-1 * time.Hour).UnixNano())

Performance

Measured through the public API with 200-byte payloads (Go 1.27, macOS, Intel i5-8257U). Dequeue with 5,000 queued. Dequeue syncs before it reads.

Operation	Throughput	Latency
Enqueue (buffered)	~230K ops/sec	~4 μs
Enqueue (AutoSync)	~27 ops/sec	~37 ms
EnqueueBatch (100 msgs, buffered)	~2M msgs/sec	~500 ns/msg
EnqueueBatch (100 msgs, AutoSync)	~2.6K msgs/sec	~385 μs/msg
Dequeue	~80 ops/sec	~12 ms

Use batch operations for high throughput.

Examples

See examples/ for runnable code with comprehensive READMEs:

Getting Started:

• simple - Basic operations ⭐ Start here!
• producer-consumer - Concurrent access with multiple goroutines

Core Features:

• ttl - Message expiration (time-to-live)
• headers - Message metadata and routing
• replay - Seeking and time-travel through message history
• streaming - Real-time event streaming
• metrics - Monitoring and observability

Advanced Features (v1.1.0+):

• priority - Priority ordering (v1.1.0+)
• dlq - Dead Letter Queue for failed messages (v1.2.0+)
• compression - GZIP payload compression (v1.3.0+)
• deduplication - Exactly-once semantics (v1.4.0+)

CLI Tool

go install github.com/1mb-dev/ledgerq/cmd/ledgerq@latest

View queue statistics:

$ ledgerq stats /path/to/queue

Queue Statistics
================
Directory:         /path/to/queue
Total Messages:    1500
Pending Messages:  342
Next Message ID:   1501
Read Message ID:   1159
Segment Count:     3

Inspect queue metadata:

$ ledgerq inspect /path/to/queue

Compact queue (remove processed messages):

$ ledgerq compact /path/to/queue

Peek at messages (without dequeuing):

$ ledgerq peek /path/to/queue 5

Documentation

• Usage Guide - Complete reference
• Architecture - Internal design
• API Reference - GoDoc
• Contributing - Development guide

Limitations

Batch Operations and Priority Queues

Important: DequeueBatch() always returns messages in FIFO order (by message ID), even when EnablePriorities=true. This is a performance trade-off: batch dequeue optimizes for sequential I/O rather than priority ordering.

For priority-aware consumption, use Dequeue() in a loop or the Stream() API.

// FIFO batch dequeue (fast, sequential I/O)
messages, _ := q.DequeueBatch(100)

// Priority-aware dequeue (respects priority order)
for i := 0; i < 100; i++ {
    msg, err := q.Dequeue()
    // ...
}

Priority Queue Memory Usage

When EnablePriorities=true, the queue maintains an in-memory index of all unprocessed message IDs, organized by priority level. Memory usage scales linearly with queue depth (measured with runtime.MemStats):

• 1M pending messages: ~24 MB
• 10M pending messages: ~240 MB
• 100M pending messages: ~2.4 GB

For very deep queues (>10M messages), consider:

• Using FIFO mode (EnablePriorities=false)
• Implementing application-level batching to keep queue depth low
• Running compaction regularly to remove processed messages

Testing

go test ./...
go test -race ./...
go test -bench=. ./internal/queue
go test -fuzz=FuzzEnqueueDequeue -fuzztime=30s ./internal/queue

License

MIT License - see LICENSE

Credits

Inspired by Apache Kafka, NATS Streaming, and Python's persist-queue.

Built as a learning project for message queue internals and Go systems programming.