feat(wal): implement segment rotation, recovery scanner, and record parser

- wal/segment_manager.go: segment lifecycle with rotation at batch boundaries
- wal/scanner.go: segment discovery, ordering, and continuity validation
- wal/record_parser.go: block-level physical record parsing with tail corruption detection
- Comprehensive tests for all modules, all pass with -race

wal/record_parser.go

@@ -0,0 +1,132 @@
+package wal
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"os"
+)
+
+// TailCorruptionError indicates that the WAL tail contains corrupt data
+// (bad CRC, unexpected non-zero padding bytes, etc.). Recovery may safely
+// truncate at the last valid record.
+type TailCorruptionError struct {
+	Offset int
+	Err    error
+}
+
+func (e *TailCorruptionError) Error() string {
+	return fmt.Sprintf("wal: tail corruption at offset %d: %v", e.Offset, e.Err)
+}
+
+func (e *TailCorruptionError) Unwrap() error {
+	return e.Err
+}
+
+// IsTailCorruption reports whether err indicates tail corruption in a WAL
+// segment. Callers may safely truncate the segment at the last valid record.
+func IsTailCorruption(err error) bool {
+	var tce *TailCorruptionError
+	return errors.As(err, &tce)
+}
+
+// ParseBlock parses physical records from a single block of raw bytes.
+// The block is typically WalBlockSize (32 KB) bytes, but the last block of a
+// segment may be shorter. Trailing bytes after the last record must be all
+// zeros (padding); non-zero trailing bytes produce a TailCorruptionError.
+func ParseBlock(data []byte) ([]*PhysicalRecord, error) {
+	var records []*PhysicalRecord
+	pos := 0
+
+	for pos < len(data) {
+		remaining := len(data) - pos
+
+		// If fewer than PhysicalRecordHeaderSize bytes remain, they must be
+		// zero-padding.
+		if remaining < PhysicalRecordHeaderSize {
+			tail := data[pos:]
+			if !isAllZeros(tail) {
+				return records[:len(records):len(records)], &TailCorruptionError{
+					Offset: pos,
+					Err:    fmt.Errorf("non-zero padding bytes in tail (%d bytes)", len(tail)),
+				}
+			}
+			break
+		}
+
+		// Check for zero-filled header (preallocated / unwritten space).
+		if isAllZeros(data[pos : pos+PhysicalRecordHeaderSize]) {
+			// Verify rest of block is also zeros.
+			if !isAllZeros(data[pos:]) {
+				return records[:len(records):len(records)], &TailCorruptionError{
+					Offset: pos,
+					Err:    errors.New("zero header but non-zero bytes follow"),
+				}
+			}
+			break
+		}
+
+		rec, consumed, err := DecodePhysicalRecord(data[pos:])
+		if err != nil {
+			return records[:len(records):len(records)], &TailCorruptionError{
+				Offset: pos,
+				Err:    err,
+			}
+		}
+		records = append(records, rec)
+		pos += consumed
+	}
+
+	return records, nil
+}
+
+// ParseRecordsFromFile opens a WAL segment file, skips the file header, reads
+// blocks sequentially, and returns all physical records in order. Short final
+// blocks are handled correctly.
+func ParseRecordsFromFile(filePath string) ([]*PhysicalRecord, error) {
+	f, err := os.Open(filePath)
+	if err != nil {
+		return nil, fmt.Errorf("wal: parse records: %w", err)
+	}
+	defer f.Close()
+
+	// Skip file header.
+	if _, err := f.Seek(WalFileHeaderSize, 0); err != nil {
+		return nil, fmt.Errorf("wal: seek past header: %w", err)
+	}
+
+	var allRecords []*PhysicalRecord
+	buf := make([]byte, WalBlockSize)
+
+	for {
+		n, readErr := f.Read(buf)
+		if readErr != nil {
+			if errors.Is(readErr, os.ErrClosed) {
+				return nil, fmt.Errorf("wal: file closed during read: %w", readErr)
+			}
+			break
+		}
+		if n == 0 {
+			break
+		}
+
+		blockData := buf[:n]
+		recs, err := ParseBlock(blockData)
+		if err != nil {
+			// Return records collected so far along with the error.
+			return allRecords, err
+		}
+		allRecords = append(allRecords, recs...)
+
+		// If we got a short block, this was the last one.
+		if n < WalBlockSize {
+			break
+		}
+	}
+
+	return allRecords, nil
+}
+
+func isAllZeros(data []byte) bool {
+	return bytes.Count(data, []byte{0}) == len(data)
+}