go-kv/wal/validate.go

package wal

import (
	"fmt"
	"math"

	"github.com/dailz/go-kv/config"
)

// ValidateBatchLimits checks that a batch of WAL entries satisfies all resource
// limits from the supplied configuration before sequence allocation or WAL append.
// Returns a descriptive error on violation, nil on success.
func ValidateBatchLimits(entries []*WalEntry, cfg *config.WalConfig) error {
	entryCount := uint64(len(entries))
	if entryCount == 0 {
		return fmt.Errorf("wal: batch entry count must be > 0")
	}
	if entryCount > uint64(cfg.MaxBatchEntries) {
		return fmt.Errorf("wal: entry count %d exceeds limit %d", entryCount, cfg.MaxBatchEntries)
	}

	// Per-entry validation and total encoded size calculation.
	// Each entry encoded size = 1(opType) + 1(valueKind) + varint(keyLen) + varint(valLen) + keyLen + valLen
	// We use MaxWalVarintBytes (5) as worst-case varint size.
	var totalEncodedSize uint64
	for i, e := range entries {
		keyLen := uint64(len(e.Key))
		if keyLen == 0 || keyLen > uint64(cfg.MaxKeyBytes) {
			return fmt.Errorf("wal: entry %d: key length %d out of range [1, %d]", i, keyLen, cfg.MaxKeyBytes)
		}
		valLen := uint64(len(e.Value))

		if e.OpType == OpPut && e.ValueKind == VKInline {
			if valLen > uint64(cfg.MaxInlineValue) {
				return fmt.Errorf("wal: entry %d: inline value length %d exceeds limit %d", i, valLen, cfg.MaxInlineValue)
			}
		}

		// entrySize = 2 + varint(keyLen) + varint(valLen) + keyLen + valLen
		// Use worst-case varint size for safety.
		entrySize, err := safeAddChecked(2+MaxWalVarintBytes+MaxWalVarintBytes, keyLen)
		if err != nil {
			return fmt.Errorf("wal: entry %d: size overflow: %w", i, err)
		}
		entrySize, err = safeAddChecked(entrySize, valLen)
		if err != nil {
			return fmt.Errorf("wal: entry %d: size overflow: %w", i, err)
		}
		totalEncodedSize, err = safeAddChecked(totalEncodedSize, entrySize)
		if err != nil {
			return fmt.Errorf("wal: total encoded size overflow: %w", err)
		}
	}

	// totalWithBatchHeader = WalBatchHeaderSize + totalEncodedSize
	totalWithBatchHeader, err := safeAddChecked(WalBatchHeaderSize, totalEncodedSize)
	if err != nil {
		return fmt.Errorf("wal: batch size overflow: %w", err)
	}

	if totalWithBatchHeader > uint64(cfg.MaxBatchSize) {
		return fmt.Errorf("wal: total batch size %d exceeds limit %d", totalWithBatchHeader, cfg.MaxBatchSize)
	}

	// Check that the batch fits in a WAL segment with physical record overhead.
	blockSize := uint64(cfg.BlockSize)
	prHeaderSize := uint64(PhysicalRecordHeaderSize)

	if blockSize <= prHeaderSize {
		return fmt.Errorf("wal: block size %d must be > physical record header size %d", blockSize, prHeaderSize)
	}
	maxPayload := blockSize - prHeaderSize

	numRecords := divCeilChecked(totalWithBatchHeader, maxPayload)
	overhead, err := safeMulChecked(numRecords, prHeaderSize)
	if err != nil {
		return fmt.Errorf("wal: physical record overhead overflow: %w", err)
	}

	totalWithOverhead, err := safeAddChecked(totalWithBatchHeader, overhead)
	if err != nil {
		return fmt.Errorf("wal: total with overhead overflow: %w", err)
	}

	maxSegmentPayload := cfg.MaxSegmentSize - WalFileHeaderSize
	if totalWithOverhead > maxSegmentPayload {
		return fmt.Errorf("wal: batch with overhead %d exceeds segment payload %d", totalWithOverhead, maxSegmentPayload)
	}

	return nil
}

// safeAddChecked returns a + b or an error if the result overflows uint64.
func safeAddChecked(a, b uint64) (uint64, error) {
	if a > math.MaxUint64-b {
		return 0, fmt.Errorf("uint64 overflow: %d + %d", a, b)
	}
	return a + b, nil
}

// safeMulChecked returns a * b or an error if the result overflows uint64.
func safeMulChecked(a, b uint64) (uint64, error) {
	if a != 0 && b > math.MaxUint64/a {
		return 0, fmt.Errorf("uint64 overflow: %d * %d", a, b)
	}
	return a * b, nil
}

// divCeilChecked returns ceil(a / b) for b > 0.
func divCeilChecked(a, b uint64) uint64 {
	return (a + b - 1) / b
}