cascadeq

README

cascadeq

cascadeq is a persistent, disk-backed FIFO queue for []byte items. It runs entirely in memory under normal load and spills to disk only when the configured memory limits are reached — so it is fast when the queue is shallow and durable when it grows deep. Data in the queue is persisted to disk when the queue is closed and is available for reading when the queue is opened again.

Features

• Memory-first: items stay in memory until configurable byte or item-count limits are hit
• Disk overflow: tailQ is flushed to sequentially numbered .dat files; files are reloaded into headQ as items are consumed
• Persisted: in-memory state is persisted on Close; existing files are picked up automatically on restart
• Optional compression: pass WithGzip(true) to compress overflow files
• Idle snapshots: pass WithSnapshotInterval to periodically persist in-memory data when the queue is idle, to increase crash safety with minmal throughput cost.
• Channel-based API: consume items with a plain <-q.Out() in a select

Installation

go get github.com/gammazero/cascadeq

Usage

package main

import (
    "fmt"
    "log"

    "github.com/gammazero/cascadeq"
)

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

    // Write items.
    for i := range 5 {
        if err := q.Put([]byte(fmt.Sprintf("item-%d", i))); err != nil {
            log.Fatal(err)
        }
    }

    // Read items until the queue is empty.
    for {
        select {
        case item := <-q.Out():
            fmt.Println(string(item))
        case <-q.Empty():
            return
        }
    }
}

Options

Option	Default	Description
WithMaxMemory(n)	1 MiB	Maximum bytes held across both in-memory queues
WithMaxMemItems(n)	4096	Maximum item count held across both in-memory queues
WithMaxItemSize(n)	64 KiB	Reject items larger than this
WithMinItemSize(n)	0	Reject items smaller than this
WithGzip(true)	disabled	Compress overflow files with gzip
WithSnapshotInterval(d)	disabled	Write in-memory state to disk after this much idle time

Design

Dual-queue memory model

The queue maintains two in-memory deques at all times:

• headQ — the read-from queue; holds the oldest unconsumed items
• tailQ — the write-to queue; holds the newest items

The configured memory/item limits are divided equally between headQ and tailQ. Stats.MaxQBytes and Stats.MaxQLen reflect this per-queue half.

Write path

New items go to headQ when both queues are empty and headQ has space. Otherwise they go to tailQ. When tailQ is full:

• If no overflow files exist and headQ has room, shift items from tailQ into headQ.
• Otherwise, flush tailQ to the next numbered .dat file and clear tailQ.

Read path

Items are consumed from the front of headQ. When headQ empties:

1. Load the next numbered file into headQ, then delete the file.
2. If no files exist, swap tailQ ↔ headQ (O(1)).
3. If neither, signal Empty() and stop sending on the output channel.

File format and naming

Files are stored in the directory passed to New and named {name}-{hexnum}.dat (or .dat.gz when compression is enabled). Each file is a sequence of big-endian int32 length-prefixed byte records. File number 0 is reserved for the headQ snapshot written on Close or on an idle snapshot tick; higher numbers are tailQ overflow files written in sequence. Corrupt files are renamed with a .bad extension rather than deleted.

Single-goroutine event loop

All state mutation happens inside one goroutine via a select over the input channel (Put), output channel (Out), clear and stats request channels, the empty signal channel, an optional snapshot ticker, and the close signal. A single sync.RWMutex only protects the closed flag, gating Put/Clear/Stats from racing with Close. No other synchronization is needed.

Snapshot feature

When WithSnapshotInterval is set, a ticker fires at half the configured interval. If no items have been written or read since the previous tick (idle), the in-memory queues are written to disk. On Close, a synchronous snapshot is always written for any non-empty queue.

Documentation

Overview

Package cascadeq implements a persistent, disk-backed FIFO queue for []byte items. It operates primarily in memory and spills to disk only when the configured memory limits are reached, making it efficient for workloads that fit in memory while remaining durable under backpressure.

Memory model

The queue maintains two in-memory deques at all times:

• headQ: the read-from queue, holding the oldest unconsumed items
• tailQ: the write-to queue, holding the newest items

The configured memory/item limits are divided equally between headQ and tailQ. When tailQ is full, its contents are flushed to a numbered .dat file on disk unless headQ has room to absorb them directly. Items are read back by loading the next numbered file into headQ; when no files remain, tailQ and headQ are swapped in O(1).

Usage

Create a Queue with New, write items with Put, and consume them by ranging over the channel returned by Out:

q, err := cascadeq.New("myqueue", "/var/data/queues")
if err != nil {
    log.Fatal(err)
}
defer q.Close()

// Producer
if err := q.Put([]byte("hello")); err != nil {
    log.Println(err)
}

// Consumer
for {
    select {
    case item := <-q.Out():
        process(item)
    case <-q.Empty():
        return // no more items right now
    case <-q.Done():
        return // queue closed
    }
}

Options

Behaviour is configured via functional options passed to New:

• WithMaxMemory: cap total in-memory bytes (default 1 MiB)
• WithMaxMemItems: cap total in-memory item count (default 4096)
• WithMaxItemSize: reject items above this byte size (default 64 KiB)
• WithMinItemSize: reject items below this byte size (default 0)
• WithGzip: enable gzip compression of disk files
• WithSnapshotInterval: periodically persist in-memory state when idle

Persistence and file format

Overflow files are written to the directory supplied to New and named {name}-{hexnum}.dat (or .dat.gz when compression is enabled). Each file is a sequence of big-endian int32 length-prefixed byte records. File number 0 is reserved for the headQ snapshot written on close or idle snapshot; higher numbers are sequential tailQ overflow files. On restart, New re-discovers these files and resumes from where the queue left off. Corrupt files are renamed with a .bad extension rather than deleted.

Concurrency

All state mutations run inside a single goroutine. Put, Clear, and Stats are safe to call from multiple goroutines concurrently. Close is idempotent and safe to call from any goroutine.

Example (Basic)

Example_basic demonstrates creating a queue, writing items, and reading them back until the queue is empty.

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/gammazero/cascadeq"
)

func main() {
	dir, err := os.MkdirTemp("", "cascadeq-example-*")
	if err != nil {
		log.Fatal(err)
	}
	defer os.RemoveAll(dir)

	q, err := cascadeq.New("example", dir)
	if err != nil {
		log.Fatal(err)
	}
	defer q.Close()

	for i := range 3 {
		if err := q.Put(fmt.Appendf(nil, "item-%d", i)); err != nil {
			log.Fatal(err)
		}
	}

	for {
		select {
		case item := <-q.Out():
			fmt.Println(string(item))
		case <-q.Empty():
			return
		}
	}
}

Output:
item-0
item-1
item-2

Example (Options)

Example_options shows configuring memory limits and gzip compression.

package main

import (
	"fmt"
	"log"
	"os"

	"github.com/gammazero/cascadeq"
)

func main() {
	dir, err := os.MkdirTemp("", "cascadeq-example-*")
	if err != nil {
		log.Fatal(err)
	}
	defer os.RemoveAll(dir)

	q, err := cascadeq.New("example", dir,
		cascadeq.WithMaxMemory(64*1024),  // 64 KiB in-memory budget
		cascadeq.WithMaxMemItems(128),    // at most 128 items in memory
		cascadeq.WithMaxItemSize(4*1024), // reject items larger than 4 KiB
		cascadeq.WithGzip(true),          // compress overflow files
	)
	if err != nil {
		log.Fatal(err)
	}
	defer q.Close()

	if err := q.Put([]byte("hello")); err != nil {
		log.Fatal(err)
	}

	item := <-q.Out()
	fmt.Println(string(item))
}

Output:
hello

Index

• Constants
• Variables
• func WithGzip(enable bool) func(*Queue)
• func WithMaxItemSize(maxSize int) func(*Queue)
• func WithMaxMemItems(maxItems int) func(*Queue)
• func WithMaxMemory(maxBytes int) func(*Queue)
• func WithMinItemSize(minSize int) func(*Queue)
• func WithSnapshotInterval(d time.Duration) func(*Queue)
• type Queue
  • func New(name, dir string, options ...func(*Queue)) (*Queue, error)
  • func (q *Queue) Clear() error
  • func (q *Queue) Close() error
  • func (q *Queue) Dir() string
  • func (q *Queue) Done() <-chan struct{}
  • func (q *Queue) Empty() <-chan struct{}
  • func (q *Queue) Name() string
  • func (q *Queue) Out() <-chan []byte
  • func (q *Queue) Put(item []byte) (err error)
  • func (q *Queue) Stats() Stats
• type Stats

Examples

• Package (Basic)
• Package (Options)

Constants

const (
	DefaultMaxMemory   = 1024 * 1024
	DefaultMaxMemItems = 4096
	DefaultMaxItemSize = 65536
)

const (
	BadFileExt = ".bad"
)

Variables

var ErrClosed = errors.New("closed")

ErrClosed is returned when I/O is attempted on a closed Queue.

Functions

func WithGzip

func WithGzip(enable bool) func(*Queue)

WithGzip enables, if passed true, gzip compression of buffer files.

func WithMaxItemSize

func WithMaxItemSize(maxSize int) func(*Queue)

WithMaxItemSize specifies the maximum allowed size of a single []byte item in the queue.

func WithMaxMemItems

func WithMaxMemItems(maxItems int) func(*Queue)

WithMaxMemItems sets the maximum number of items that the queue keeps in memory before items are written to disk.

func WithMaxMemory

func WithMaxMemory(maxBytes int) func(*Queue)

WithMaxMemory sets the maximum amount of memory used by all items in the queue before items are written to disk.

func WithMinItemSize

func WithMinItemSize(minSize int) func(*Queue)

WithMinItemSize specifies the minimum allowed size of a single []byte item in the queue.

func WithSnapshotInterval

func WithSnapshotInterval(d time.Duration) func(*Queue)

WithSnapshotInterval enables snapshots and sets the amount of time that the queue must be idle before saving a snapshot of the items stored in memory. The queue must be idle for at least half of the specified time and at most the entire specified time. Idle snapshots are disabled by default and are enabled when a positive value is specified for this option.

Types

type Queue

type Queue struct {
	// contains filtered or unexported fields
}

Queue implements a filesystem backed FIFO queue.

func New

func New(name, dir string, options ...func(*Queue)) (*Queue, error)

New creates a new file-backed FIFO queue instance.

func (*Queue) Clear

func (q *Queue) Clear() error

Clear removes all items from the queue.

func (*Queue) Close

func (q *Queue) Close() error

Close stops the queue's internal goroutine and prevents any more input or output with the queue. After calling Close, any attempted input or output results in an error.

func (*Queue) Dir

func (q *Queue) Dir() string

Dir returns the directory where queued data files are stored.

func (*Queue) Done

func (q *Queue) Done() <-chan struct{}

Done returns a channel that is closed when the Queue is closed.

func (*Queue) Empty

func (q *Queue) Empty() <-chan struct{}

Empty returns a channel that is signaled when the queue is empty. This is useful for exiting a select when there are currently no more queued items to read.

func (*Queue) Name

func (q *Queue) Name() string

Name returns the name of the Queue instance.

func (*Queue) Out

func (q *Queue) Out() <-chan []byte

Out returns the receive-only []byte channel for reading data.

func (*Queue) Put

func (q *Queue) Put(item []byte) (err error)

Put writes a []byte to the queue.

func (*Queue) Stats

func (q *Queue) Stats() Stats

Stats retrieves information about queue internal data.

type Stats

type Stats struct {
	Closed     bool
	MaxQBytes  int
	MaxQLen    int
	HeadQBytes int
	HeadQLen   int
	TailQBytes int
	TailQLen   int
	Files      []string
}

Stats holds information about the Queue internal state.