example_test.go

package spool_test

import (
	"errors"
	"fmt"
	"os"

	"github.com/JohanLindvall/spool"
)

func Example() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64 << 20})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()

	// Append is durable: once it returns, the record survives a crash.
	if err := q.Append([]byte("batch-1")); err != nil {
		panic(err)
	}

	// Pop hands back the record with a commit function; commit only after the
	// record is truly handled (e.g. the collector acked it) — a crash before
	// commit re-delivers it.
	data, commit, ok, err := q.Pop()
	fmt.Printf("%q ok=%v err=%v\n", data, ok, err)
	commit()
	// Output:
	// "batch-1" ok=true err=<nil>
}

// Group commit trades one fsync per record for one per group. The checkpoint
// rule moves with it: advance only after Sync returns, never after
// AppendNoSync. This is the API whose misuse loses data.
func Example_groupCommit() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64 << 20})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()

	batch := [][]byte{[]byte("a"), []byte("b"), []byte("c")}
	for _, rec := range batch {
		// NOT durable yet. A crash here loses the whole group.
		if err := q.AppendNoSync(rec); err != nil {
			panic(err)
		}
	}
	if err := q.Sync(); err != nil {
		// The group was rolled back and must be re-appended. Do NOT advance
		// any upstream checkpoint.
		if errors.Is(err, spool.ErrGroupLost) {
			fmt.Println("group lost; re-append")
		}
		return
	}
	// Only now may the producer forget its copy of a, b and c.
	fmt.Println("durable:", q.Stats().Appended, "records")
	// Output:
	// durable: 3 records
}

// A consumer loop that blocks instead of polling, stays cancellable, and does
// not spin on the error returns. Pop's errors are advancing — the queue has
// already moved past the damage — so they must not be retried in a tight loop
// without also waiting.
func Example_consumerLoop() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64 << 20})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()
	_ = q.Append([]byte("only"))

	done := make(chan struct{})
	close(done) // stand-in for ctx.Done(); this example stops after one record

	for {
		data, commit, ok, err := q.Pop()
		switch {
		case errors.Is(err, spool.ErrClosed):
			return
		case err != nil:
			// Corruption was dropped and the queue advanced. Count it and go
			// round again; Signal fires for these too, so the wait below is
			// still correct if this was the last thing in the queue.
			fmt.Println("read error:", err)
		case !ok:
			select {
			case <-q.Signal():
			case <-done:
				fmt.Println("stopped with", q.Bytes(), "bytes queued")
				return
			}
		default:
			fmt.Printf("send %q\n", data)
			commit() // only after the send succeeded
		}
	}
	// Output:
	// send "only"
	// stopped with 0 bytes queued
}

// ErrFull is backpressure and clears as the consumer drains. ErrExceedsCap
// wraps it but never clears: the record cannot fit under this cap at all.
func Example_errFull() {
	dir, _ := os.MkdirTemp("", "spool")
	defer func() { _ = os.RemoveAll(dir) }()

	q, err := spool.Open(dir, spool.Options{MaxBytes: 64})
	if err != nil {
		panic(err)
	}
	defer func() { _ = q.Close() }()

	oversized := q.Append(make([]byte, 64))
	fmt.Println("oversized permanent:", errors.Is(oversized, spool.ErrExceedsCap))

	// A record that fits: at most MaxBytes-FrameOverhead bytes.
	fmt.Println("largest that fits:", q.Cap()-spool.FrameOverhead)
	// Output:
	// oversized permanent: true
	// largest that fits: 52
}