api_test.go

package spool

import (
	"errors"
	"os"
	"testing"
)

// TestDefaultMaxBytesIsBounded pins the defaulting decision that matters most
// operationally: an unconfigured spool must not be allowed to grow until the
// volume dies. Options{} is what every casual caller and every benchmark uses.
func TestDefaultMaxBytesIsBounded(t *testing.T) {
	s := mustOpen(t, t.TempDir(), Options{})
	if got := s.Cap(); got != DefaultMaxBytes {
		t.Fatalf("zero-value Options gave Cap()=%d, want DefaultMaxBytes=%d", got, DefaultMaxBytes)
	}
	if got := s.Stats().MaxBytes; got != DefaultMaxBytes {
		t.Errorf("Stats().MaxBytes=%d, want %d", got, DefaultMaxBytes)
	}
}

// TestUnboundedRequiresOptIn: uncapped is still available, but only by asking.
func TestUnboundedRequiresOptIn(t *testing.T) {
	s := mustOpen(t, t.TempDir(), Options{MaxBytes: Unbounded})
	if got := s.Cap(); got != 0 {
		t.Fatalf("Cap()=%d after MaxBytes:Unbounded, want 0 (the documented uncapped value)", got)
	}
}

// TestErrExceedsCapIsPermanent separates the two conditions that used to share
// ErrFull. A record larger than the whole cap fails on an EMPTY queue and will
// fail identically forever, so a caller retrying on ErrFull would spin on it.
func TestErrExceedsCapIsPermanent(t *testing.T) {
	const cap = 64
	s := mustOpen(t, t.TempDir(), Options{MaxBytes: cap})
	oversized := make([]byte, cap) // cap + FrameOverhead > cap, on any queue state

	err := s.Append(oversized)
	if !errors.Is(err, ErrExceedsCap) {
		t.Fatalf("Append(oversized) on an empty queue = %v, want ErrExceedsCap", err)
	}
	if !errors.Is(err, ErrFull) {
		t.Error("ErrExceedsCap must wrap ErrFull so existing errors.Is(err, ErrFull) callers keep working")
	}
	if _, _, ok, _ := s.Pop(); ok {
		t.Fatal("the refused record was enqueued anyway")
	}
	// The documented threshold: a record fits iff len <= MaxBytes-FrameOverhead.
	if err := s.Append(make([]byte, cap-FrameOverhead)); err != nil {
		t.Fatalf("Append at exactly MaxBytes-FrameOverhead = %v, want success", err)
	}
	// And a merely-too-big-right-now record is plain ErrFull, which clears.
	err = s.Append([]byte("x"))
	if !errors.Is(err, ErrFull) || errors.Is(err, ErrExceedsCap) {
		t.Fatalf("Append on a full queue = %v, want plain ErrFull", err)
	}
	if n := s.Stats().Full; n != 2 {
		t.Errorf("Stats().Full = %d, want 2", n)
	}
}

// TestRecordTooLarge covers the frame format's own ceiling. Constructing the
// input needs a >4 GiB allocation, so it runs only when explicitly asked for.
func TestRecordTooLarge(t *testing.T) {
	if os.Getenv("SPOOL_HUGE_ALLOC_TEST") == "" {
		t.Skip("needs a >4 GiB allocation; set SPOOL_HUGE_ALLOC_TEST=1 to run")
	}
	s := mustOpen(t, t.TempDir(), Options{MaxBytes: Unbounded})
	err := s.Append(make([]byte, MaxRecordBytes+1))
	if !errors.Is(err, ErrRecordTooLarge) {
		t.Fatalf("Append(MaxRecordBytes+1) = %v, want ErrRecordTooLarge", err)
	}
	if errors.Is(err, ErrFull) {
		t.Error("ErrRecordTooLarge must not wrap ErrFull: no cap setting makes it succeed")
	}
}

// TestPopAfterCloseIsRejected turns what used to be a logged observation into
// a contract. An unguarded Pop after Close reopened the head segment into
// s.readF (which nothing would ever close again) and could run
// retireConsumedLocked, unlinking segment files while persistCursor silently
// no-opped — deletion without acknowledgement.
func TestPopAfterCloseIsRejected(t *testing.T) {
	dir := t.TempDir()
	s := mustOpen(t, dir, Options{MaxBytes: 1 << 20})
	mustAppend(t, s, "one")
	mustAppend(t, s, "two")

	if err := s.Close(); err != nil {
		t.Fatalf("Close: %v", err)
	}
	for _, op := range []struct {
		name string
		err  error
	}{
		{"Append", s.Append([]byte("x"))},
		{"AppendNoSync", s.AppendNoSync([]byte("x"))},
		{"Sync", s.Sync()},
	} {
		if !errors.Is(op.err, ErrClosed) {
			t.Errorf("%s after Close = %v, want ErrClosed", op.name, op.err)
		}
	}
	data, commit, ok, err := s.Pop()
	if ok || !errors.Is(err, ErrClosed) {
		t.Fatalf("Pop after Close = (%q, ok=%v, err=%v), want ErrClosed", data, ok, err)
	}
	if commit != nil {
		t.Error("Pop returned a commit function alongside ErrClosed")
	}
	// Close is idempotent.
	if err := s.Close(); err != nil {
		t.Errorf("second Close = %v, want nil", err)
	}
	// Nothing was retired: both records survive.
	s2 := mustOpen(t, dir, Options{MaxBytes: 1 << 20})
	if got := drain(t, s2); len(got) != 2 {
		t.Fatalf("after restart drained %q, want both records", got)
	}
}

// TestCommitAfterCloseIsNoop: a commit captured before Close must not mutate
// state or unlink files afterwards.
func TestCommitAfterCloseIsNoop(t *testing.T) {
	dir := t.TempDir()
	s := mustOpen(t, dir, Options{MaxBytes: 1 << 20})
	mustAppend(t, s, "one")
	mustAppend(t, s, "two")

	_, commit, ok, err := s.Pop()
	if !ok || err != nil {
		t.Fatalf("Pop: ok=%v err=%v", ok, err)
	}
	if err := s.Close(); err != nil {
		t.Fatal(err)
	}
	commit() // must not advance readOff or retire anything

	s2 := mustOpen(t, dir, Options{MaxBytes: 1 << 20})
	if got := drain(t, s2); len(got) != 2 {
		t.Fatalf("after restart drained %q, want both records — a commit after Close must not persist", got)
	}
}

// TestRequeueRotatesHead replaces the old AppendForce footgun: it moves an
// undeliverable head record to the back so everything behind it can drain,
// ignoring MaxBytes because the rotation is size-neutral.
func TestRequeueRotatesHead(t *testing.T) {
	s := mustOpen(t, t.TempDir(), Options{MaxBytes: 128})
	for _, r := range []string{"poison", "good-1", "good-2"} {
		mustAppend(t, s, r)
	}
	data, commit, ok, err := s.Pop()
	if !ok || err != nil || string(data) != "poison" {
		t.Fatalf("Pop = (%q, ok=%v, err=%v)", data, ok, err)
	}
	rotated, err := s.Requeue(data, commit)
	if err != nil || !rotated {
		t.Fatalf("Requeue = (%v, %v), want (true, nil)", rotated, err)
	}
	if got := drain(t, s); len(got) != 3 || got[0] != "good-1" || got[2] != "poison" {
		t.Fatalf("drained %q, want the poison record moved to the back", got)
	}
}

// TestRequeueAloneIsNoop: rewriting the only queued record is pointless churn,
// and a Requeue that silently no-opped would be indistinguishable from one that
// rotated — hence the rotated return.
func TestRequeueAloneIsNoop(t *testing.T) {
	s := mustOpen(t, t.TempDir(), Options{MaxBytes: 1 << 20})
	mustAppend(t, s, "only")
	data, commit, ok, err := s.Pop()
	if !ok || err != nil {
		t.Fatalf("Pop: ok=%v err=%v", ok, err)
	}
	rotated, err := s.Requeue(data, commit)
	if err != nil {
		t.Fatalf("Requeue: %v", err)
	}
	if rotated {
		t.Error("Requeue rotated the only queued record")
	}
	if got := drain(t, s); len(got) != 1 || got[0] != "only" {
		t.Fatalf("drained %q, want the record still queued exactly once", got)
	}
}

// TestSignalFiresWhenPopAdvancesPastDamage: a consumer that treats every
// ok==false the same way must not block forever after Pop skips damaged data.
// With an idle producer nothing else would ever wake it.
func TestSignalFiresWhenPopAdvancesPastDamage(t *testing.T) {
	dir := t.TempDir()
	s := mustOpen(t, dir, Options{MaxBytes: 1 << 20})
	mustAppend(t, s, "damaged-record")
	mustAppend(t, s, "intact")
	path := s.segs[0].path
	if err := s.Close(); err != nil {
		t.Fatal(err)
	}

	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	raw[segHeaderLen+frameHeaderLen] ^= 0xff // flip a payload byte of record 1
	if err := os.WriteFile(path, raw, 0o644); err != nil {
		t.Fatal(err)
	}

	s2 := mustOpen(t, dir, Options{MaxBytes: 1 << 20})
	// Drain any signal left from load.
	select {
	case <-s2.Signal():
	default:
	}
	if _, _, ok, err := s2.Pop(); ok || !errors.Is(err, ErrCorrupt) {
		t.Fatalf("Pop = ok:%v err:%v, want ErrCorrupt", ok, err)
	}
	select {
	case <-s2.Signal():
	default:
		t.Fatal("Signal did not fire after Pop advanced past a damaged frame: a waiting consumer would block forever")
	}
	if n := s2.Stats().LostBytes; n == 0 {
		t.Error("Stats().LostBytes is 0 after a frame was destroyed")
	}
}

func TestStatsCounters(t *testing.T) {
	s := mustOpen(t, t.TempDir(), Options{MaxBytes: 1 << 20})
	for i := 0; i < 5; i++ {
		mustAppend(t, s, "record")
	}
	// One popped-but-uncommitted, one popped twice (redelivery), three left.
	_, _, _, _ = s.Pop()
	_, commit, _, _ := s.Pop()
	commit()

	st := s.Stats()
	if st.Appended != 5 {
		t.Errorf("Appended = %d, want 5", st.Appended)
	}
	if st.Popped != 2 {
		t.Errorf("Popped = %d, want 2 (redeliveries count)", st.Popped)
	}
	if st.Committed != 1 {
		t.Errorf("Committed = %d, want 1", st.Committed)
	}
	if st.DiskBytes < st.BacklogBytes {
		t.Errorf("DiskBytes %d < BacklogBytes %d: the footprint can never be under the backlog", st.DiskBytes, st.BacklogBytes)
	}
	if st.BacklogBytes != s.Bytes() {
		t.Errorf("Stats().BacklogBytes = %d but Bytes() = %d", st.BacklogBytes, s.Bytes())
	}
	if st.Segments != s.Segments() {
		t.Errorf("Stats().Segments = %d but Segments() = %d", st.Segments, s.Segments())
	}
}

// TestStatsUnsyncedBytes is the crash-loss window: what AppendNoSync has
// accepted that an fsync has not yet covered.
func TestStatsUnsyncedBytes(t *testing.T) {
	s := mustOpen(t, t.TempDir(), Options{MaxBytes: 1 << 20})
	if n := s.Stats().UnsyncedBytes; n != 0 {
		t.Fatalf("UnsyncedBytes = %d on a fresh spool, want 0", n)
	}
	if err := s.AppendNoSync([]byte("not-durable")); err != nil {
		t.Fatal(err)
	}
	want := int64(frameHeaderLen + len("not-durable"))
	if n := s.Stats().UnsyncedBytes; n != want {
		t.Errorf("UnsyncedBytes = %d after AppendNoSync, want %d", n, want)
	}
	if err := s.Sync(); err != nil {
		t.Fatal(err)
	}
	if n := s.Stats().UnsyncedBytes; n != 0 {
		t.Errorf("UnsyncedBytes = %d after Sync, want 0", n)
	}
}