go-kv/wal/recover_offset_test.go

package wal

import (
	"errors"
	"os"
	"path/filepath"
	"strings"
	"testing"
)

// writeRawSegmentHeader writes a 32-byte WAL header to filePath. Used by
// findLastCompleteBatchEnd tests to construct minimal segment files.
func writeRawSegmentHeader(t *testing.T, filePath string) {
	t.Helper()
	hdr := &WalFileHeader{
		BlockSize:     32 * 1024,
		SegmentID:     0,
		StartSequence: 0,
	}
	encoded := EncodeWalHeader(hdr)
	if err := os.WriteFile(filePath, encoded[:], 0o644); err != nil {
		t.Fatalf("WriteFile header: %v", err)
	}
}

// appendRawBytes appends arbitrary bytes to filePath.
func appendRawBytes(t *testing.T, filePath string, data []byte) {
	t.Helper()
	f, err := os.OpenFile(filePath, os.O_WRONLY|os.O_APPEND, 0o644)
	if err != nil {
		t.Fatalf("OpenFile append: %v", err)
	}
	defer f.Close()
	if _, err := f.Write(data); err != nil {
		t.Fatalf("Write: %v", err)
	}
}

// appendFullRecord encodes and appends a complete WAL batch as a Full record.
func appendFullRecord(t *testing.T, filePath string, batch []byte) {
	t.Helper()
	rec := EncodePhysicalRecord(RecFull, batch)
	appendRawBytes(t, filePath, rec)
}

// appendFragment encodes and appends a single fragment record.
func appendFragment(t *testing.T, filePath string, recType uint8, payload []byte) {
	t.Helper()
	rec := EncodePhysicalRecord(recType, payload)
	appendRawBytes(t, filePath, rec)
}

func TestFindLastCompleteBatchEnd_CleanSegment(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	batchA := []byte("batch-A-content")
	batchB := []byte("batch-B-content")
	appendFullRecord(t, filePath, batchA)
	endOfA, _ := fileSize(filePath)
	appendFullRecord(t, filePath, batchB)
	endOfB, _ := fileSize(filePath)

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != endOfB {
		t.Errorf("got %d, want %d (end of Batch B)", got, endOfB)
	}
	if got == endOfA {
		t.Errorf("got end of Batch A, should be end of Batch B")
	}
}

// Regression guard for H8: partial fragment tail must return end of last
// COMPLETE batch, not end of last physical record.
func TestFindLastCompleteBatchEnd_PartialTailFragment(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	batchA := []byte("batch-A-content")
	appendFullRecord(t, filePath, batchA)
	endOfA, _ := fileSize(filePath)

	// Append First + Middle fragments (no Last) — H8 case.
	appendFragment(t, filePath, RecFirst, []byte("first-fragment-data"))
	appendFragment(t, filePath, RecMiddle, []byte("middle-fragment-data"))

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != endOfA {
		t.Errorf("got %d, want %d (end of Batch A, NOT end of Middle fragment)", got, endOfA)
	}
}

func TestFindLastCompleteBatchEnd_NoBatches(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != int64(WalFileHeaderSize) {
		t.Errorf("got %d, want %d (WalFileHeaderSize)", got, WalFileHeaderSize)
	}
}

func TestFindLastCompleteBatchEnd_PhysicalCorruption(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	batchA := []byte("batch-A-content")
	appendFullRecord(t, filePath, batchA)
	endOfA, _ := fileSize(filePath)

	// Append corrupted bytes (will fail CRC).
	appendRawBytes(t, filePath, []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF})

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != endOfA {
		t.Errorf("got %d, want %d (end of Batch A, before corruption)", got, endOfA)
	}
}

func TestFindLastCompleteBatchEnd_PartialBatchOnly(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	// Only First + Middle fragments, no complete batch ever.
	appendFragment(t, filePath, RecFirst, []byte("first-data"))
	appendFragment(t, filePath, RecMiddle, []byte("middle-data"))

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != int64(WalFileHeaderSize) {
		t.Errorf("got %d, want %d (no complete batch, stay at header)", got, WalFileHeaderSize)
	}
}

// Oracle BLOCKING test: must continue past padding in a full block to read
// the next block. Original code returned on first padding, truncating all
// later batches.
func TestFindLastCompleteBatchEnd_BlockBoundaryPadding(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	// Batch A: large enough to nearly fill block 1.
	// Block 1 layout: [32-byte header][Batch A (Full record)] [padding to end]
	// We need: 32 + len(Full record of A) + padding == 32 + WalBlockSize
	// Full record = 7 (header) + len(payload). So:
	//   len(A) such that 7 + len(A) leaves < 7 bytes before block end
	// Then writer pads to end of block and Batch B goes in block 2.

	// Available in block 1 after file header = WalBlockSize - 32 = 32736
	// We want Batch A record size to be 32736 - 6 = 32730 (leaving 6 bytes, < 7, padding)
	// So payload size = 32730 - 7 = 32723
	// Batch A = batch header (18) + entry bytes. Entry: Put with key + value.
	// Simplest: use raw bytes (we're testing physical layout, not batch validity).
	bigPayload := make([]byte, 32723)
	for i := range bigPayload {
		bigPayload[i] = byte('A')
	}
	recA := EncodePhysicalRecord(RecFull, bigPayload)
	appendRawBytes(t, filePath, recA)

	endOfBlock1 := int64(WalFileHeaderSize + WalBlockSize) // 32 + 32768
	currentSize, _ := fileSize(filePath)
	// Pad to end of block 1 with zeros.
	padLen := int(endOfBlock1 - currentSize)
	if padLen > 0 {
		appendRawBytes(t, filePath, make([]byte, padLen))
	}

	// Batch B in block 2 (normal small batch after padding boundary).
	batchB := []byte("batch-B")
	recB := EncodePhysicalRecord(RecFull, batchB)
	appendRawBytes(t, filePath, recB)
	endOfB, _ := fileSize(filePath)

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != endOfB {
		t.Errorf("got %d, want %d (end of Batch B in block 2, after padding)", got, endOfB)
	}
	// Specifically: must NOT be at end of Batch A (would be ~32755, before padding).
	if got < endOfBlock1 {
		t.Errorf("got %d < %d (returned at end of Batch A, missed block 2)", got, endOfBlock1)
	}
}

func TestFindLastCompleteBatchEnd_NonZeroTailPadding(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	batchA := []byte("batch-A")
	appendFullRecord(t, filePath, batchA)
	endOfA, _ := fileSize(filePath)

	// Append 3 non-zero bytes (< PhysicalRecordHeaderSize=7). This is
	// technically invalid padding (per design: padding must be zeros), but
	// findLastCompleteBatchEnd should still return end of last complete
	// batch — this is "tail corruption" classification territory.
	appendRawBytes(t, filePath, []byte{0xFF, 0xFF, 0xFF})

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != endOfA {
		t.Errorf("got %d, want %d (end of Batch A)", got, endOfA)
	}
}

func TestFindLastCompleteBatchEnd_ZeroTailPadding(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)

	batchA := []byte("batch-A")
	appendFullRecord(t, filePath, batchA)
	endOfA, _ := fileSize(filePath)

	// Append 3 zero bytes (valid tail padding in a short final block).
	appendRawBytes(t, filePath, []byte{0, 0, 0})

	got, err := findLastCompleteBatchEnd(filePath)
	if err != nil {
		t.Fatalf("findLastCompleteBatchEnd: %v", err)
	}
	if got != endOfA {
		t.Errorf("got %d, want %d (end of Batch A)", got, endOfA)
	}
}

func fileSize(filePath string) (int64, error) {
	fi, err := os.Stat(filePath)
	if err != nil {
		return 0, err
	}
	return fi.Size(), nil
}

// -------- truncateAndPersist tests --------

func TestTruncateAndPersist_Success(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)
	appendRawBytes(t, filePath, []byte("batch-A"))
	appendRawBytes(t, filePath, []byte("extra-bytes-to-be-truncated"))

	endOfA, _ := fileSize(filePath)
	truncateOffset := int64(WalFileHeaderSize) + 7 // just past header, before "batch-A"

	if err := truncateAndPersist(filePath, truncateOffset, dir, nil); err != nil {
		t.Fatalf("truncateAndPersist: %v", err)
	}

	gotSize, _ := fileSize(filePath)
	if gotSize != truncateOffset {
		t.Errorf("file size = %d, want %d (truncated)", gotSize, truncateOffset)
	}
	_ = endOfA // not used; verify only that size matches truncate point
}

func TestTruncateAndPersist_FtruncateFailure(t *testing.T) {
	dir := t.TempDir()
	nonExistent := filepath.Join(dir, "no-such-file.wal")

	err := truncateAndPersist(nonExistent, 100, dir, nil)
	if err == nil {
		t.Fatal("expected error on non-existent file")
	}
	if !strings.Contains(err.Error(), "ftruncate") {
		t.Errorf("error should mention 'ftruncate', got: %v", err)
	}
}

func TestTruncateAndPersist_DirFsyncFailure(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)
	appendRawBytes(t, filePath, []byte("extra"))

	orig := dirFsyncFn
	dirFsyncFn = func(string) error { return errors.New("simulated dir fsync failure") }
	t.Cleanup(func() { dirFsyncFn = orig })

	err := truncateAndPersist(filePath, int64(WalFileHeaderSize), dir, nil)
	if err == nil {
		t.Fatal("expected error on dir fsync failure")
	}
	if !strings.Contains(err.Error(), "fsync WAL dir") {
		t.Errorf("error should mention 'fsync WAL dir', got: %v", err)
	}
}

func TestTruncateAndPersist_SegmentFsyncFailure(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)
	appendRawBytes(t, filePath, []byte("extra"))

	orig := segmentFsyncFn
	segmentFsyncFn = func(string) error { return errors.New("simulated segment fsync failure") }
	t.Cleanup(func() { segmentFsyncFn = orig })

	err := truncateAndPersist(filePath, int64(WalFileHeaderSize), dir, nil)
	if err == nil {
		t.Fatal("expected error on segment fsync failure")
	}
	if !strings.Contains(err.Error(), "fsync truncated segment") {
		t.Errorf("error should mention 'fsync truncated segment', got: %v", err)
	}
}

// Oracle nice-to-have: verify retry after dir-fsync failure. The first
// call fails after ftruncate succeeded; the second call (with fsync
// restored) must succeed and the file must end up correctly truncated.
func TestTruncateAndPersist_RetryAfterDirFsyncFailure(t *testing.T) {
	dir := t.TempDir()
	filePath := filepath.Join(dir, "segment-0.wal")
	writeRawSegmentHeader(t, filePath)
	appendRawBytes(t, filePath, []byte("extra-bytes"))

	truncateOffset := int64(WalFileHeaderSize)

	// First call: inject dir fsync failure.
	orig := dirFsyncFn
	dirFsyncFn = func(string) error { return errors.New("simulated") }
	if err := truncateAndPersist(filePath, truncateOffset, dir, nil); err == nil {
		t.Fatal("first call should fail")
	}
	dirFsyncFn = orig

	// Second call: must succeed (ftruncate is idempotent).
	if err := truncateAndPersist(filePath, truncateOffset, dir, nil); err != nil {
		t.Fatalf("retry truncateAndPersist: %v", err)
	}

	gotSize, _ := fileSize(filePath)
	if gotSize != truncateOffset {
		t.Errorf("after retry, file size = %d, want %d", gotSize, truncateOffset)
	}
}