go-kv/wal/record.go

package wal

import (
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
)

// PhysicalRecord represents a single physical record in the WAL.
type PhysicalRecord struct {
	CRC     uint32
	Length  uint16
	Type    uint8
	Payload []byte
}

// EncodePhysicalRecord encodes a physical record with the given type and payload.
// Format: [crc32 u32 LE][length u16 LE][type u8][payload bytes]
// CRC covers length + type + payload.
func EncodePhysicalRecord(recType uint8, payload []byte) []byte {
	length := uint16(len(payload))
	buf := make([]byte, PhysicalRecordHeaderSize+len(payload))

	// Write length and type first so we can compute CRC.
	binary.LittleEndian.PutUint16(buf[4:6], length)
	buf[6] = recType
	copy(buf[7:], payload)

	// CRC covers bytes [4:] = length + type + payload. Castagnoli polynomial
	// per design §3.2 line 359.
	crc := crc32.Checksum(buf[4:], crc32cTable)
	binary.LittleEndian.PutUint32(buf[0:4], crc)

	return buf
}

// DecodePhysicalRecord decodes a physical record from data.
// Returns the record, number of bytes consumed, and any error.
func DecodePhysicalRecord(data []byte) (rec *PhysicalRecord, consumed int, err error) {
	if len(data) < PhysicalRecordHeaderSize {
		return nil, 0, errors.New("record: data too short for header")
	}

	crc := binary.LittleEndian.Uint32(data[0:4])
	length := binary.LittleEndian.Uint16(data[4:6])
	recType := data[6]

	// H1: length must be > 0 per design §3.2 line 389.
	if length == 0 {
		return nil, 0, errors.New("record: length must be > 0")
	}
	// H2: type must be valid per design §3.2 line 366-372.
	if recType < RecFull || recType > RecLast {
		return nil, 0, fmt.Errorf("record: invalid type %d", recType)
	}

	if int(length) > len(data)-PhysicalRecordHeaderSize {
		return nil, 0, errors.New("record: data too short for payload")
	}

	payload := make([]byte, length)
	copy(payload, data[7:7+length])

	// Verify CRC: covers length + type + payload.
	expectedCRC := crc32.Checksum(data[4 : 7+length], crc32cTable)
	if crc != expectedCRC {
		return nil, 0, errors.New("record: CRC mismatch")
	}

	consumed = PhysicalRecordHeaderSize + int(length)
	return &PhysicalRecord{
		CRC:     crc,
		Length:  length,
		Type:    recType,
		Payload: payload,
	}, consumed, nil
}

// PaddingNeeded returns the number of padding bytes needed at blockOffset.
// If the remaining space in the current block is <= PhysicalRecordHeaderSize (7),
// that remaining space must be zero-padded.
func PaddingNeeded(blockOffset uint32) int {
	remaining := WalBlockSize - (blockOffset % WalBlockSize)
	if remaining <= PhysicalRecordHeaderSize {
		return int(remaining)
	}
	return 0
}

// CanFitRecord reports whether a physical record with the given payload length
// can fit in the current block starting at blockOffset.
func CanFitRecord(blockOffset uint32, payloadLen uint32) bool {
	remaining := WalBlockSize - (blockOffset % WalBlockSize)
	return int(remaining) >= PhysicalRecordHeaderSize+int(payloadLen)
}

// SplitIntoRecords splits an encoded WAL batch into physical record payloads
// respecting 32 KB block boundaries.
// Each returned byte slice is the full encoded physical record (header + payload).
func SplitIntoRecords(encodedBatch []byte) [][]byte {
	maxPayload := WalBlockSize - PhysicalRecordHeaderSize
	total := len(encodedBatch)

	if total == 0 {
		return nil
	}

	// Single record fits entirely.
	if total <= maxPayload {
		return [][]byte{EncodePhysicalRecord(RecFull, encodedBatch)}
	}

	var records [][]byte
	offset := 0

	for offset < total {
		chunkLen := min(total-offset, maxPayload)

		var recType uint8
		switch {
		case offset == 0 && offset+chunkLen == total:
			recType = RecFull
		case offset == 0:
			recType = RecFirst
		case offset+chunkLen == total:
			recType = RecLast
		default:
			recType = RecMiddle
		}

		records = append(records, EncodePhysicalRecord(recType, encodedBatch[offset:offset+chunkLen]))
		offset += chunkLen
	}

	return records
}