go-kv/wal/block_writer_test.go

package wal

import (
	"bytes"
	"encoding/binary"
	"hash/crc32"
	"testing"
)

func TestBlockWriterSingleRecord(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	payload := []byte("hello world")
	if err := bw.WriteRecord(RecFull, payload, &buf); err != nil {
		t.Fatalf("WriteRecord: %v", err)
	}

	// Record should still be buffered (block not full).
	if buf.Len() != 0 {
		t.Fatalf("expected no flush yet, got %d bytes", buf.Len())
	}

	// Flush to get the data.
	if err := bw.Flush(&buf); err != nil {
		t.Fatalf("Flush: %v", err)
	}

	written := buf.Bytes()

	// Verify the record is at the start of a full block.
	if len(written) != WalBlockSize {
		t.Fatalf("expected full block %d bytes, got %d", WalBlockSize, len(written))
	}

	// Decode and verify the physical record.
	rec, consumed, err := DecodePhysicalRecord(written)
	if err != nil {
		t.Fatalf("DecodePhysicalRecord: %v", err)
	}

	if rec.Type != RecFull {
		t.Errorf("type = %d, want RecFull(%d)", rec.Type, RecFull)
	}
	if string(rec.Payload) != "hello world" {
		t.Errorf("payload = %q, want %q", rec.Payload, "hello world")
	}

	// Remaining bytes after the record should be zero padding.
	recEnd := consumed
	for i := recEnd; i < WalBlockSize; i++ {
		if written[i] != 0 {
			t.Errorf("padding byte [%d] = %d, want 0", i, written[i])
		}
	}
}

func TestBlockWriterPadding(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	payloadLen := WalBlockSize - PhysicalRecordHeaderSize
	payload := make([]byte, payloadLen)
	for i := range payload {
		payload[i] = byte(i % 256)
	}

	if err := bw.WriteRecord(RecFull, payload, &buf); err != nil {
		t.Fatalf("WriteRecord: %v", err)
	}

	if buf.Len() != WalBlockSize {
		t.Fatalf("expected auto-flush of full block (%d bytes), got %d", WalBlockSize, buf.Len())
	}

	if bw.BlockOffset() != 0 {
		t.Errorf("BlockOffset = %d, want 0 after full block write", bw.BlockOffset())
	}

	buf.Reset()
	smallPayload := []byte("next")
	if err := bw.WriteRecord(RecFull, smallPayload, &buf); err != nil {
		t.Fatalf("WriteRecord after full block: %v", err)
	}
	if buf.Len() != 0 {
		t.Fatalf("expected no flush for partial block, got %d bytes", buf.Len())
	}
}

func TestBlockWriterPaddingNeeded(t *testing.T) {
	tests := []struct {
		name   string
		offset uint32
		want   int
	}{
		{"beginning", 0, 0},
		{"mid_block", 100, 0},
		{"7_remaining", WalBlockSize - 7, 7},
		{"6_remaining", WalBlockSize - 6, 6},
		{"1_remaining", WalBlockSize - 1, 1},
		{"full_block", WalBlockSize, 0}, // would be reset
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			bw := &BlockWriter{offset: tt.offset}
			got := bw.paddingNeeded()
			if got != tt.want {
				t.Errorf("paddingNeeded() = %d, want %d", got, tt.want)
			}
		})
	}
}

func TestBlockWriterCrossBlock(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	payloadLen := WalBlockSize - PhysicalRecordHeaderSize - 5
	payload1 := make([]byte, payloadLen)
	for i := range payload1 {
		payload1[i] = byte('A' + i%26)
	}

	if err := bw.WriteRecord(RecFull, payload1, &buf); err != nil {
		t.Fatalf("WriteRecord payload1: %v", err)
	}

	if buf.Len() != 0 {
		t.Fatalf("expected no flush after first record, got %d bytes", buf.Len())
	}

	payload2 := []byte("second")
	if err := bw.WriteRecord(RecFull, payload2, &buf); err != nil {
		t.Fatalf("WriteRecord payload2: %v", err)
	}

	if buf.Len() != WalBlockSize {
		t.Fatalf("expected %d bytes flushed, got %d", WalBlockSize, buf.Len())
	}

	firstBlock := buf.Bytes()[:WalBlockSize]
	rec, _, err := DecodePhysicalRecord(firstBlock)
	if err != nil {
		t.Fatalf("DecodePhysicalRecord block 1: %v", err)
	}
	if rec.Type != RecFull {
		t.Errorf("type = %d, want RecFull", rec.Type)
	}
	if len(rec.Payload) != payloadLen {
		t.Errorf("payload len = %d, want %d", len(rec.Payload), payloadLen)
	}
	for i := WalBlockSize - 5; i < WalBlockSize; i++ {
		if firstBlock[i] != 0 {
			t.Errorf("padding byte [%d] = %d, want 0", i, firstBlock[i])
		}
	}

	if err := bw.Flush(&buf); err != nil {
		t.Fatalf("Flush: %v", err)
	}

	secondBlock := buf.Bytes()[WalBlockSize:]
	if len(secondBlock) != WalBlockSize {
		t.Fatalf("second block: expected %d bytes, got %d", WalBlockSize, len(secondBlock))
	}

	rec2, _, err := DecodePhysicalRecord(secondBlock)
	if err != nil {
		t.Fatalf("DecodePhysicalRecord block 2: %v", err)
	}
	if string(rec2.Payload) != "second" {
		t.Errorf("payload2 = %q, want %q", rec2.Payload, "second")
	}
}

func TestBlockWriterRecordTooLarge(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	// Payload that exceeds block capacity.
	payload := make([]byte, WalBlockSize)
	err := bw.WriteRecord(RecFull, payload, &buf)
	if err == nil {
		t.Fatal("expected error for oversized record")
	}
}

func TestBlockWriterMultipleRecords(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	// Write several small records.
	records := []struct {
		recType uint8
		payload []byte
	}{
		{RecFirst, []byte("part1")},
		{RecMiddle, []byte("part2")},
		{RecLast, []byte("part3")},
	}

	for _, r := range records {
		if err := bw.WriteRecord(r.recType, r.payload, &buf); err != nil {
			t.Fatalf("WriteRecord(%d, %q): %v", r.recType, r.payload, err)
		}
	}

	if err := bw.Flush(&buf); err != nil {
		t.Fatalf("Flush: %v", err)
	}

	data := buf.Bytes()

	// Decode all three records from the block.
	offset := 0
	for i, expected := range records {
		rec, consumed, err := DecodePhysicalRecord(data[offset:])
		if err != nil {
			t.Fatalf("record %d: DecodePhysicalRecord at offset %d: %v", i, offset, err)
		}
		if rec.Type != expected.recType {
			t.Errorf("record %d: type = %d, want %d", i, rec.Type, expected.recType)
		}
		if string(rec.Payload) != string(expected.payload) {
			t.Errorf("record %d: payload = %q, want %q", i, rec.Payload, expected.payload)
		}
		offset += consumed
	}
}

func TestBlockWriterReset(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	if err := bw.WriteRecord(RecFull, []byte("data"), &buf); err != nil {
		t.Fatalf("WriteRecord: %v", err)
	}

	if bw.BlockOffset() == 0 {
		t.Fatal("expected non-zero offset after write")
	}

	bw.Reset()
	if bw.BlockOffset() != 0 {
		t.Errorf("BlockOffset after Reset = %d, want 0", bw.BlockOffset())
	}
}

func TestBlockWriterFlushEmptyBlock(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	// Flushing an empty block should be a no-op.
	if err := bw.Flush(&buf); err != nil {
		t.Fatalf("Flush empty: %v", err)
	}
	if buf.Len() != 0 {
		t.Errorf("expected 0 bytes, got %d", buf.Len())
	}
}

func TestBlockWriterOffsetTracking(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	// Write a small record and verify offset.
	payload := []byte("track-me")
	if err := bw.WriteRecord(RecFull, payload, &buf); err != nil {
		t.Fatalf("WriteRecord: %v", err)
	}

	expectedOffset := uint32(PhysicalRecordHeaderSize + len(payload))
	if bw.BlockOffset() != expectedOffset {
		t.Errorf("BlockOffset = %d, want %d", bw.BlockOffset(), expectedOffset)
	}

	// Flush should write exactly one full block.
	if err := bw.Flush(&buf); err != nil {
		t.Fatalf("Flush: %v", err)
	}
	if buf.Len() != WalBlockSize {
		t.Errorf("flushed %d bytes, want %d", buf.Len(), WalBlockSize)
	}
}

func TestBlockWriterAutoFlushFullBlock(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	// Fill the block exactly.
	payloadLen := WalBlockSize - PhysicalRecordHeaderSize
	payload := make([]byte, payloadLen)
	for i := range payload {
		payload[i] = byte(i)
	}

	if err := bw.WriteRecord(RecFull, payload, &buf); err != nil {
		t.Fatalf("WriteRecord exact fill: %v", err)
	}

	// Block should have been auto-flushed.
	if buf.Len() != WalBlockSize {
		t.Errorf("expected auto-flush of %d bytes, got %d", WalBlockSize, buf.Len())
	}
	if bw.BlockOffset() != 0 {
		t.Errorf("BlockOffset after auto-flush = %d, want 0", bw.BlockOffset())
	}

	// Verify CRC is correct by decoding.
	data := buf.Bytes()
	rec, _, err := DecodePhysicalRecord(data)
	if err != nil {
		t.Fatalf("DecodePhysicalRecord: %v", err)
	}
	if len(rec.Payload) != payloadLen {
		t.Errorf("payload len = %d, want %d", len(rec.Payload), payloadLen)
	}
}

func TestBlockWriterPhysicalRecordCRC(t *testing.T) {
	bw := NewBlockWriter()
	var buf bytes.Buffer

	payload := []byte("crc-check")
	if err := bw.WriteRecord(RecFull, payload, &buf); err != nil {
		t.Fatalf("WriteRecord: %v", err)
	}
	if err := bw.Flush(&buf); err != nil {
		t.Fatalf("Flush: %v", err)
	}

	data := buf.Bytes()

	// Manually verify CRC: covers length + type + payload.
	storedCRC := binary.LittleEndian.Uint32(data[0:4])
	length := binary.LittleEndian.Uint16(data[4:6])
	recType := data[6]

	if recType != RecFull {
		t.Errorf("type = %d, want RecFull", recType)
	}
	if int(length) != len(payload) {
		t.Errorf("length = %d, want %d", length, len(payload))
	}

	// Verify CRC over [length, type, payload].
	crcData := data[4 : 7+length]
	computedCRC := crc32.Checksum(crcData, crc32cTable)
	if storedCRC != computedCRC {
		t.Errorf("CRC mismatch: stored %d, computed %d", storedCRC, computedCRC)
	}
}