fathom/internal/store/shard_test.go

package store

import (
	"os"
	"path/filepath"
	"testing"

	"git.z.gd/zzg/fathom/internal/model"
)

func writeChunks(t *testing.T, dir string, batches ...[]model.Span) *Shard {
	t.Helper()
	sh, err := OpenShard(dir, model.TableSpans, "2026-09-15")
	if err != nil {
		t.Fatal(err)
	}
	for _, b := range batches {
		c, err := buildSpanChunk(b, sh.Blob())
		if err != nil {
			t.Fatal(err)
		}
		if err := sh.AppendChunk(c); err != nil {
			t.Fatal(err)
		}
	}
	return sh
}

func countRows(t *testing.T, sh *Shard) int {
	t.Helper()
	n := 0
	if err := sh.Chunks(func(c *Chunk) error { n += c.Rows; return nil }); err != nil {
		t.Fatal(err)
	}
	return n
}

func TestShardAppendAndIterate(t *testing.T) {
	dir := t.TempDir()
	spans := sampleSpans()
	sh := writeChunks(t, dir, spans[:2], spans[2:])
	defer sh.Close()

	var got []model.Span
	err := sh.Chunks(func(c *Chunk) error {
		for i := 0; i < c.Rows; i++ {
			got = append(got, readSpan(c, i))
		}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 3 {
		t.Fatalf("read %d spans across chunks, wrote 3", len(got))
	}
	// Chunks must come back in write order, so the second batch's span is last.
	if got[2].SpanID != "span-3" {
		t.Errorf("chunks came back out of order: last span is %q", got[2].SpanID)
	}
}

// TestTornTailRecovery is the crash-safety claim the format rests on: a chunk
// left half-written must cost exactly that chunk and nothing earlier.
func TestTornTailRecovery(t *testing.T) {
	spans := sampleSpans()

	truncations := []struct {
		name string
		cut  func(size int64) int64
	}{
		{"mid body", func(s int64) int64 { return s - 20 }},
		{"one byte short", func(s int64) int64 { return s - 1 }},
		{"mid frame header", func(s int64) int64 { return s - 1 }},
	}

	for _, tc := range truncations {
		t.Run(tc.name, func(t *testing.T) {
			dir := t.TempDir()
			sh := writeChunks(t, dir, spans[:2])
			goodSize := func() int64 {
				st, err := os.Stat(filepath.Join(dir, model.TableSpans, "2026-09-15.fathom"))
				if err != nil {
					t.Fatal(err)
				}
				return st.Size()
			}()
			// Append a second chunk, then damage only that one.
			c, err := buildSpanChunk(spans[2:], sh.Blob())
			if err != nil {
				t.Fatal(err)
			}
			if err := sh.AppendChunk(c); err != nil {
				t.Fatal(err)
			}
			sh.Close()

			path := filepath.Join(dir, model.TableSpans, "2026-09-15.fathom")
			st, _ := os.Stat(path)
			if err := os.Truncate(path, tc.cut(st.Size())); err != nil {
				t.Fatal(err)
			}

			// Reopening must discard the torn chunk and keep the first.
			sh2, err := OpenShard(dir, model.TableSpans, "2026-09-15")
			if err != nil {
				t.Fatalf("reopen after torn write: %v", err)
			}
			defer sh2.Close()

			if got := countRows(t, sh2); got != 2 {
				t.Errorf("after recovery got %d rows, want the 2 from the intact chunk", got)
			}
			st2, _ := os.Stat(path)
			if st2.Size() != goodSize {
				t.Errorf("recovered file is %d bytes, want %d (the intact prefix)", st2.Size(), goodSize)
			}

			// And the shard must still be writable afterwards.
			c2, err := buildSpanChunk(spans[:1], sh2.Blob())
			if err != nil {
				t.Fatal(err)
			}
			if err := sh2.AppendChunk(c2); err != nil {
				t.Fatalf("append after recovery: %v", err)
			}
			if got := countRows(t, sh2); got != 3 {
				t.Errorf("after appending post-recovery got %d rows, want 3", got)
			}
		})
	}
}

// TestCorruptedBodyRecovery covers the case truncation does not: bytes that are
// all present but wrong. Only the CRC catches this.
func TestCorruptedBodyRecovery(t *testing.T) {
	dir := t.TempDir()
	spans := sampleSpans()
	sh := writeChunks(t, dir, spans[:2], spans[2:])
	sh.Close()

	path := filepath.Join(dir, model.TableSpans, "2026-09-15.fathom")
	b, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	// Flip a bit deep in the final chunk's body.
	b[len(b)-5] ^= 0xFF
	if err := os.WriteFile(path, b, 0o644); err != nil {
		t.Fatal(err)
	}

	sh2, err := OpenShard(dir, model.TableSpans, "2026-09-15")
	if err != nil {
		t.Fatal(err)
	}
	defer sh2.Close()
	if got := countRows(t, sh2); got != 2 {
		t.Errorf("got %d rows after corrupting the last chunk, want 2", got)
	}
}

func TestEmptyAndMissingShard(t *testing.T) {
	dir := t.TempDir()
	sh, err := OpenShard(dir, model.TableSpans, "2026-09-15")
	if err != nil {
		t.Fatal(err)
	}
	defer sh.Close()
	if got := countRows(t, sh); got != 0 {
		t.Errorf("fresh shard reported %d rows", got)
	}
}

func TestBlobRefOutOfRange(t *testing.T) {
	blob, err := OpenBlobFile(t.TempDir() + "/x.blob")
	if err != nil {
		t.Fatal(err)
	}
	defer blob.Close()
	if _, err := blob.Append("hello"); err != nil {
		t.Fatal(err)
	}
	// A damaged chunk can produce a reference past the end of the blob file.
	if _, err := blob.Read(BlobRef{Off: 0, Len: 9999}); err == nil {
		t.Error("reading past the end of the blob file succeeded")
	}
	if _, err := blob.Read(BlobRef{Off: -1, Len: 1}); err == nil {
		t.Error("reading at a negative offset succeeded")
	}
	// The zero reference is normal and means absent.
	if s, err := blob.Read(BlobRef{}); err != nil || s != "" {
		t.Errorf("zero blob ref gave (%q, %v), want empty and no error", s, err)
	}
}