diff --git a/.omo/boulder.json b/.omo/boulder.json index 1e701d1..3d6dda3 100644 --- a/.omo/boulder.json +++ b/.omo/boulder.json @@ -8,7 +8,7 @@ "plan_name": "phase1-wal", "status": "active", "started_at": "2026-06-12T05:09:32.588Z", - "updated_at": "2026-06-12T05:22:47.301Z", + "updated_at": "2026-06-12T05:26:58.955Z", "session_ids": [ "opencode:ses_145c3bae9ffeTB2zbsTym0Cev8" ], @@ -42,6 +42,19 @@ "status": "completed", "ended_at": "2026-06-12T05:22:47.301Z", "elapsed_ms": 173304 + }, + "todo:6": { + "task_key": "todo:6", + "task_label": "6", + "task_title": "WAL Batch encode/decode + fragment collector", + "session_id": "opencode:ses_145b65d08ffebmSv4sk5I69JXZ", + "agent": "Sisyphus-Junior", + "category": "unspecified-high", + "updated_at": "2026-06-12T05:26:58.955Z", + "started_at": "2026-06-12T05:26:43.370Z", + "status": "completed", + "ended_at": "2026-06-12T05:26:58.955Z", + "elapsed_ms": 15585 } } } @@ -49,7 +62,7 @@ "active_plan": "/home/dailz/workspace/src/go-kv/.omo/plans/phase1-wal.md", "started_at": "2026-06-12T05:09:32.588Z", "status": "active", - "updated_at": "2026-06-12T05:22:47.301Z", + "updated_at": "2026-06-12T05:26:58.955Z", "session_ids": [ "opencode:ses_145c3bae9ffeTB2zbsTym0Cev8" ], @@ -83,6 +96,19 @@ "status": "completed", "ended_at": "2026-06-12T05:22:47.301Z", "elapsed_ms": 173304 + }, + "todo:6": { + "task_key": "todo:6", + "task_label": "6", + "task_title": "WAL Batch encode/decode + fragment collector", + "session_id": "opencode:ses_145b65d08ffebmSv4sk5I69JXZ", + "agent": "Sisyphus-Junior", + "category": "unspecified-high", + "updated_at": "2026-06-12T05:26:58.955Z", + "started_at": "2026-06-12T05:26:43.370Z", + "status": "completed", + "ended_at": "2026-06-12T05:26:58.955Z", + "elapsed_ms": 15585 } }, "agent": "atlas" diff --git a/.omo/plans/phase1-wal.md b/.omo/plans/phase1-wal.md index bb4958b..d3ffbb0 100644 --- a/.omo/plans/phase1-wal.md +++ b/.omo/plans/phase1-wal.md @@ -555,7 +555,7 @@ Max Concurrent: 6 (Wave 1b) **Commit**: YES (group with Tasks 3, 4) -- [ ] 6. WAL Batch encode/decode + fragment collector +- [x] 6. WAL Batch encode/decode + fragment collector **What to do**: - 创建 `wal/batch.go`：`WalBatch` 结构体（Flags uint16, BaseSequence uint64, EntryCount uint32, EntriesSize uint32, Entries []byte） @@ -633,7 +633,7 @@ Max Concurrent: 6 (Wave 1b) **Commit**: YES (group with Task 9) - Message: `feat(wal): implement WAL batch codec and resource validation` -- [ ] 7. Arena allocator +- [x] 7. Arena allocator **What to do**: - 创建 `memtable/arena.go`：`Arena` 结构体 @@ -759,7 +759,7 @@ Max Concurrent: 6 (Wave 1b) **Commit**: YES (with Task 7) -- [ ] 9. WAL Batch resource validation +- [x] 9. WAL Batch resource validation **What to do**: - 创建 `wal/validate.go`：`ValidateBatchLimits(entries []*WalEntry, cfg *config.WalConfig) error` diff --git a/memtable/arena.go b/memtable/arena.go new file mode 100644 index 0000000..8783081 --- /dev/null +++ b/memtable/arena.go @@ -0,0 +1,84 @@ +package memtable + +import ( + "errors" + "sync" + "sync/atomic" +) + +// ErrArenaFull is returned when the arena cannot satisfy an allocation request. +var ErrArenaFull = errors.New("arena: insufficient capacity") + +// Arena is a bump allocator backed by a fixed-size byte slice. +// It provides thread-safe allocation with 8-byte alignment. +// +// The arena is used by the memtable to store keys, values, and skip-list +// node structures. Once allocated, bytes are never freed — the entire +// arena is discarded when the memtable is flushed. +type Arena struct { + buf []byte + offset uint32 // next allocation offset (atomic for reads) + capacity uint32 // total capacity (immutable) + mu sync.Mutex +} + +// NewArena creates a new Arena with the given capacity in bytes. +// The entire buffer is allocated upfront. +func NewArena(capacity uint32) *Arena { + return &Arena{ + buf: make([]byte, capacity), + capacity: capacity, + } +} + +// Allocate reserves alignedSize bytes in the arena and returns the offset +// at which the caller can write. The size is aligned up to 8 bytes. +func (a *Arena) Allocate(size uint32) (uint32, error) { + alignedSize := (size + 7) &^ uint32(7) + + a.mu.Lock() + rem := a.capacity - a.offset + if rem < alignedSize { + a.mu.Unlock() + return 0, ErrArenaFull + } + off := a.offset + a.offset += alignedSize + a.mu.Unlock() + + return off, nil +} + +// GetBytes returns a slice of the arena buffer at [offset, offset+size). +// It panics if the range is out of bounds. +func (a *Arena) GetBytes(offset, size uint32) []byte { + end := offset + size + if end > a.capacity { + panic("arena: GetBytes out of bounds") + } + return a.buf[offset:end] +} + +// Remaining returns the number of bytes still available for allocation. +func (a *Arena) Remaining() uint32 { + return a.capacity - atomic.LoadUint32(&a.offset) +} + +// Capacity returns the total capacity of the arena. +func (a *Arena) Capacity() uint32 { + return a.capacity +} + +// Reserve checks whether the arena has at least totalSize bytes remaining +// without actually allocating. It is used by the WAL writer to verify +// capacity before appending entries. +func (a *Arena) Reserve(totalSize uint32) error { + a.mu.Lock() + rem := a.capacity - a.offset + a.mu.Unlock() + + if rem < totalSize { + return ErrArenaFull + } + return nil +} diff --git a/memtable/arena_test.go b/memtable/arena_test.go new file mode 100644 index 0000000..3e8ac85 --- /dev/null +++ b/memtable/arena_test.go @@ -0,0 +1,115 @@ +package memtable + +import ( + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestArenaAllocate(t *testing.T) { + a := NewArena(1024) + + off, err := a.Allocate(9) + require.NoError(t, err) + require.Equal(t, uint32(0), off) + + copy(a.GetBytes(off, 9), "test data") + require.Equal(t, []byte("test data"), a.GetBytes(off, 9)) + + off2, err := a.Allocate(200) + require.NoError(t, err) + // 9 bytes aligned to 16, so second allocation starts at 16 + require.Equal(t, uint32(16), off2) +} + +func TestArenaFull(t *testing.T) { + a := NewArena(64) + + off, err := a.Allocate(60) + require.NoError(t, err) + require.Equal(t, uint32(0), off) // 60 aligned to 64 + + _, err = a.Allocate(10) + require.ErrorIs(t, err, ErrArenaFull) +} + +func TestArenaAlignment(t *testing.T) { + a := NewArena(1024) + + sizes := []uint32{1, 3, 5, 7, 8, 9, 13, 16, 100} + for _, s := range sizes { + off, err := a.Allocate(s) + require.NoError(t, err, "size=%d", s) + require.Equal(t, uint32(0), off%8, "offset %d not 8-byte aligned for size %d", off, s) + } +} + +func TestArenaConcurrent(t *testing.T) { + a := NewArena(65536) + + const goroutines = 64 + const perGoroutine = 32 + + var wg sync.WaitGroup + wg.Add(goroutines) + + offsets := make([][]uint32, goroutines) + + for i := 0; i < goroutines; i++ { + i := i + go func() { + defer wg.Done() + locals := make([]uint32, 0, perGoroutine) + for j := 0; j < perGoroutine; j++ { + off, err := a.Allocate(8) + if err == nil { + // Write unique data to detect corruption + copy(a.GetBytes(off, 8), []byte{byte(i), byte(j), 0, 0, 0, 0, 0, 0}) + locals = append(locals, off) + } + } + offsets[i] = locals + }() + } + + wg.Wait() + + // Verify data integrity — each byte pair must match its goroutine/j index + for i, locals := range offsets { + for j, off := range locals { + data := a.GetBytes(off, 8) + require.Equal(t, byte(i), data[0], "goroutine %d offset %d", i, off) + require.Equal(t, byte(j), data[1], "iteration %d offset %d", j, off) + } + } +} + +func TestArenaReserve(t *testing.T) { + a := NewArena(128) + + require.NoError(t, a.Reserve(64)) + require.NoError(t, a.Reserve(128)) + require.ErrorIs(t, a.Reserve(129), ErrArenaFull) + + // Allocate some space + _, err := a.Allocate(32) + require.NoError(t, err) // 32 aligned = 32 + + // 128 - 32 = 96 remaining + require.NoError(t, a.Reserve(96)) + require.ErrorIs(t, a.Reserve(97), ErrArenaFull) +} + +func TestArenaRemainingAndCapacity(t *testing.T) { + cap := uint32(256) + a := NewArena(cap) + + require.Equal(t, cap, a.Capacity()) + require.Equal(t, cap, a.Remaining()) + + _, err := a.Allocate(10) + require.NoError(t, err) // 10 aligned to 16 + + require.Equal(t, cap-16, a.Remaining()) +} diff --git a/wal/batch.go b/wal/batch.go new file mode 100644 index 0000000..c74c80c --- /dev/null +++ b/wal/batch.go @@ -0,0 +1,216 @@ +package wal + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" +) + +// WalBatch represents a single WAL batch — the physical persistence unit. +// A batch consists of an 18-byte header followed by a contiguous Entries region. +type WalBatch struct { + Flags uint16 + BaseSequence uint64 + EntryCount uint32 + EntriesSize uint32 + Entries []byte +} + +// EncodeWalBatch encodes a WAL batch from the given base sequence and entries. +// The returned byte slice is: BatchHeader(18) + encoded entries. +func EncodeWalBatch(baseSequence uint64, entries []*WalEntry) ([]byte, error) { + if len(entries) == 0 { + return nil, errors.New("wal: batch requires at least one entry") + } + if uint32(len(entries)) > MaxWalBatchEntryCount { + return nil, fmt.Errorf("wal: entry count %d exceeds maximum %d", len(entries), MaxWalBatchEntryCount) + } + + // Encode all entries. + var entriesBuf bytes.Buffer + for i, e := range entries { + encoded, err := EncodeEntry(e) + if err != nil { + return nil, fmt.Errorf("wal: encoding entry %d: %w", i, err) + } + entriesBuf.Write(encoded) + } + + entriesSize := uint32(entriesBuf.Len()) + if entriesSize > MaxWalBatchEntriesSize { + return nil, fmt.Errorf("wal: entries size %d exceeds maximum %d", entriesSize, MaxWalBatchEntriesSize) + } + + // Build batch: header(18) + entries. + buf := make([]byte, WalBatchHeaderSize+entriesSize) + binary.LittleEndian.PutUint16(buf[0:2], 0) // flags + binary.LittleEndian.PutUint64(buf[2:10], baseSequence) + binary.LittleEndian.PutUint32(buf[10:14], uint32(len(entries))) + binary.LittleEndian.PutUint32(buf[14:18], entriesSize) + copy(buf[WalBatchHeaderSize:], entriesBuf.Bytes()) + + return buf, nil +} + +// DecodeWalBatch decodes a WAL batch from raw bytes. +// The data must contain the full batch (header + entries). +func DecodeWalBatch(data []byte) (*WalBatch, error) { + if len(data) < WalBatchHeaderSize { + return nil, fmt.Errorf("wal: batch data too short: %d < %d", len(data), WalBatchHeaderSize) + } + + flags := binary.LittleEndian.Uint16(data[0:2]) + baseSeq := binary.LittleEndian.Uint64(data[2:10]) + entryCount := binary.LittleEndian.Uint32(data[10:14]) + entriesSize := binary.LittleEndian.Uint32(data[14:18]) + + if entryCount == 0 { + return nil, errors.New("wal: batch entry count is zero") + } + if entryCount > MaxWalBatchEntryCount { + return nil, fmt.Errorf("wal: entry count %d exceeds maximum %d", entryCount, MaxWalBatchEntryCount) + } + if entriesSize > MaxWalBatchEntriesSize { + return nil, fmt.Errorf("wal: entries size %d exceeds maximum %d", entriesSize, MaxWalBatchEntriesSize) + } + + expectedLen := WalBatchHeaderSize + entriesSize + if uint32(len(data)) < expectedLen { + return nil, fmt.Errorf("wal: entries size mismatch: header says %d bytes, have %d bytes after header", + entriesSize, len(data)-WalBatchHeaderSize) + } + + entries := make([]byte, entriesSize) + copy(entries, data[WalBatchHeaderSize:WalBatchHeaderSize+entriesSize]) + + return &WalBatch{ + Flags: flags, + BaseSequence: baseSeq, + EntryCount: entryCount, + EntriesSize: entriesSize, + Entries: entries, + }, nil +} + +// FragmentState represents the state of the fragment collector state machine. +type FragmentState uint8 + +const ( + // FragmentIdle means no fragments are being collected. + FragmentIdle FragmentState = 0 + // FragmentCollecting means fragments are being accumulated. + FragmentCollecting FragmentState = 1 +) + +// maxFragmentBufferSize is the maximum total bytes the fragment buffer can hold: +// Batch Header size + max entries size. +var maxFragmentBufferSize = uint32(WalBatchHeaderSize) + MaxWalBatchEntriesSize + +// FragmentCollector reassembles WAL batches from physical record fragments. +// The state machine transitions between Idle and Collecting based on the +// record type (Full, First, Middle, Last). +type FragmentCollector struct { + state FragmentState + buf bytes.Buffer +} + +// NewFragmentCollector creates a new FragmentCollector in Idle state. +func NewFragmentCollector() *FragmentCollector { + return &FragmentCollector{ + state: FragmentIdle, + } +} + +// Reset clears all collected data and returns the collector to Idle state. +func (fc *FragmentCollector) Reset() { + fc.state = FragmentIdle + fc.buf.Reset() +} + +// Append feeds a physical record fragment to the collector. +// The state machine enforces valid transitions: +// - Idle + Full → collect payload, stay Idle (complete) +// - Idle + First → collect payload, → Collecting +// - Idle + Middle → error +// - Idle + Last → error +// - Collecting + Middle → collect payload +// - Collecting + Last → collect payload, → Idle (complete) +// - Collecting + Full → error +// - Collecting + First → error +func (fc *FragmentCollector) Append(recType uint8, payload []byte) error { + switch fc.state { + case FragmentIdle: + switch recType { + case RecFull: + // Complete batch in one record. + if err := fc.checkBufferCapacity(len(payload)); err != nil { + return err + } + fc.buf.Write(payload) + // State stays Idle — batch is complete. + return nil + case RecFirst: + // Start collecting fragments. + if err := fc.checkBufferCapacity(len(payload)); err != nil { + return err + } + fc.buf.Write(payload) + fc.state = FragmentCollecting + return nil + case RecMiddle, RecLast: + return fmt.Errorf("wal: unexpected fragment type %d in Idle state", recType) + default: + return fmt.Errorf("wal: invalid fragment type %d", recType) + } + + case FragmentCollecting: + switch recType { + case RecMiddle: + if err := fc.checkBufferCapacity(len(payload)); err != nil { + return err + } + fc.buf.Write(payload) + return nil + case RecLast: + if err := fc.checkBufferCapacity(len(payload)); err != nil { + return err + } + fc.buf.Write(payload) + fc.state = FragmentIdle + return nil + case RecFull, RecFirst: + return fmt.Errorf("wal: unexpected fragment type %d in Collecting state", recType) + default: + return fmt.Errorf("wal: invalid fragment type %d", recType) + } + + default: + return fmt.Errorf("wal: invalid fragment collector state %d", fc.state) + } +} + +// IsComplete reports whether a complete batch has been collected. +// A batch is complete when the state returns to Idle after collecting data. +func (fc *FragmentCollector) IsComplete() bool { + return fc.state == FragmentIdle && fc.buf.Len() > 0 +} + +// BatchData returns the collected batch bytes. Only valid when IsComplete() is true. +func (fc *FragmentCollector) BatchData() []byte { + return fc.buf.Bytes() +} + +// State returns the current fragment collector state. +func (fc *FragmentCollector) State() FragmentState { + return fc.state +} + +// checkBufferCapacity ensures the total collected bytes do not exceed the limit. +func (fc *FragmentCollector) checkBufferCapacity(additional int) error { + newSize := uint32(fc.buf.Len() + additional) + if newSize > maxFragmentBufferSize { + return fmt.Errorf("wal: fragment buffer size %d exceeds maximum %d", newSize, maxFragmentBufferSize) + } + return nil +} diff --git a/wal/batch_test.go b/wal/batch_test.go new file mode 100644 index 0000000..c0becff --- /dev/null +++ b/wal/batch_test.go @@ -0,0 +1,350 @@ +package wal + +import ( + "bytes" + "encoding/binary" + "fmt" + "testing" +) + +// helper: make a Put+Inline entry. +func makePutEntry(key, value string) *WalEntry { + return &WalEntry{ + OpType: OpPut, + ValueKind: VKInline, + Key: []byte(key), + Value: []byte(value), + } +} + +// helper: make a Delete entry. +func makeDeleteEntry(key string) *WalEntry { + return &WalEntry{ + OpType: OpDelete, + ValueKind: VKNone, + Key: []byte(key), + Value: nil, + } +} + +func TestBatchRoundtrip(t *testing.T) { + // Build 10 entries: 5 Put+Inline, 3 Delete, 2 Put+Inline with empty value. + entries := []*WalEntry{ + makePutEntry("key1", "val1"), + makePutEntry("key2", "val2"), + makePutEntry("key3", "val3"), + makePutEntry("key4", "val4"), + makePutEntry("key5", "val5"), + makeDeleteEntry("key6"), + makeDeleteEntry("key7"), + makeDeleteEntry("key8"), + makePutEntry("key9", ""), // empty value + makePutEntry("key10", ""), // empty value + } + + const baseSeq uint64 = 42 + + // Encode. + encoded, err := EncodeWalBatch(baseSeq, entries) + if err != nil { + t.Fatalf("EncodeWalBatch: %v", err) + } + + // Decode. + batch, err := DecodeWalBatch(encoded) + if err != nil { + t.Fatalf("DecodeWalBatch: %v", err) + } + + // Verify header fields. + if batch.Flags != 0 { + t.Errorf("Flags = %d, want 0", batch.Flags) + } + if batch.BaseSequence != baseSeq { + t.Errorf("BaseSequence = %d, want %d", batch.BaseSequence, baseSeq) + } + if batch.EntryCount != 10 { + t.Errorf("EntryCount = %d, want 10", batch.EntryCount) + } + if batch.EntriesSize != uint32(len(batch.Entries)) { + t.Errorf("EntriesSize = %d, len(Entries) = %d", batch.EntriesSize, len(batch.Entries)) + } + if uint32(len(encoded)) != WalBatchHeaderSize+batch.EntriesSize { + t.Errorf("total encoded = %d, want %d", len(encoded), WalBatchHeaderSize+batch.EntriesSize) + } + + // Parse entries from Entries bytes. + offset := 0 + for i, want := range entries { + got, consumed, err := DecodeEntry(batch.Entries[offset:]) + if err != nil { + t.Fatalf("DecodeEntry[%d]: %v", i, err) + } + if got.OpType != want.OpType { + t.Errorf("entry[%d].OpType = %d, want %d", i, got.OpType, want.OpType) + } + if got.ValueKind != want.ValueKind { + t.Errorf("entry[%d].ValueKind = %d, want %d", i, got.ValueKind, want.ValueKind) + } + if !bytes.Equal(got.Key, want.Key) { + t.Errorf("entry[%d].Key = %q, want %q", i, got.Key, want.Key) + } + if !bytes.Equal(got.Value, want.Value) { + t.Errorf("entry[%d].Value = %q, want %q", i, got.Value, want.Value) + } + offset += consumed + } + + if offset != len(batch.Entries) { + t.Errorf("parsed %d bytes, Entries region is %d bytes", offset, len(batch.Entries)) + } +} + +func TestFragmentCollector(t *testing.T) { + // Create a batch large enough to span multiple records (~100 KB). + var entries []*WalEntry + for i := 0; i < 5000; i++ { + key := fmt.Sprintf("key_%06d", i) + val := fmt.Sprintf("val_%06d_%050s", i, "x") // ~57 bytes value + entries = append(entries, makePutEntry(key, val)) + } + + encoded, err := EncodeWalBatch(1, entries) + if err != nil { + t.Fatalf("EncodeWalBatch: %v", err) + } + + // Split into records. + records := SplitIntoRecords(encoded) + if len(records) < 2 { + t.Fatalf("expected multiple records for ~100KB batch, got %d", len(records)) + } + + // Decode each physical record to get (type, payload). + collector := NewFragmentCollector() + + for i, rec := range records { + decoded, _, err := DecodePhysicalRecord(rec) + if err != nil { + t.Fatalf("DecodePhysicalRecord[%d]: %v", i, err) + } + + // Verify state transitions. + switch { + case i == 0: + if decoded.Type != RecFirst { + t.Errorf("record[0] type = %d, want RecFirst(%d)", decoded.Type, RecFirst) + } + if collector.State() != FragmentIdle { + t.Errorf("before append[0] state = %d, want FragmentIdle", collector.State()) + } + case i == len(records)-1: + if decoded.Type != RecLast { + t.Errorf("record[%d] type = %d, want RecLast(%d)", i, decoded.Type, RecLast) + } + if collector.State() != FragmentCollecting { + t.Errorf("before append[%d] state = %d, want FragmentCollecting", i, collector.State()) + } + default: + if decoded.Type != RecMiddle { + t.Errorf("record[%d] type = %d, want RecMiddle(%d)", i, decoded.Type, RecMiddle) + } + if collector.State() != FragmentCollecting { + t.Errorf("before append[%d] state = %d, want FragmentCollecting", i, collector.State()) + } + } + + if err := collector.Append(decoded.Type, decoded.Payload); err != nil { + t.Fatalf("Append[%d]: %v", i, err) + } + } + + // After all fragments, batch should be complete. + if !collector.IsComplete() { + t.Fatal("expected IsComplete() after all fragments") + } + + // Collected data should match original encoded batch. + if !bytes.Equal(collector.BatchData(), encoded) { + t.Errorf("collected data length = %d, want %d", len(collector.BatchData()), len(encoded)) + } + + // Verify the batch can be decoded. + batch, err := DecodeWalBatch(collector.BatchData()) + if err != nil { + t.Fatalf("DecodeWalBatch from collected: %v", err) + } + if batch.EntryCount != uint32(len(entries)) { + t.Errorf("EntryCount = %d, want %d", batch.EntryCount, len(entries)) + } +} + +func TestFragmentCollectorSingleRecord(t *testing.T) { + // Small batch fits in one record. + entries := []*WalEntry{makePutEntry("k", "v")} + encoded, err := EncodeWalBatch(1, entries) + if err != nil { + t.Fatalf("EncodeWalBatch: %v", err) + } + + records := SplitIntoRecords(encoded) + if len(records) != 1 { + t.Fatalf("expected 1 record, got %d", len(records)) + } + + decoded, _, err := DecodePhysicalRecord(records[0]) + if err != nil { + t.Fatalf("DecodePhysicalRecord: %v", err) + } + if decoded.Type != RecFull { + t.Errorf("type = %d, want RecFull(%d)", decoded.Type, RecFull) + } + + collector := NewFragmentCollector() + if err := collector.Append(decoded.Type, decoded.Payload); err != nil { + t.Fatalf("Append: %v", err) + } + if !collector.IsComplete() { + t.Fatal("expected IsComplete()") + } + if !bytes.Equal(collector.BatchData(), encoded) { + t.Error("collected data mismatch") + } +} + +func TestFragmentIllegalTransitions(t *testing.T) { + tests := []struct { + name string + prepFunc func(fc *FragmentCollector) // set up initial state + recType uint8 + }{ + { + name: "Idle+RecMiddle", + prepFunc: func(fc *FragmentCollector) {}, + recType: RecMiddle, + }, + { + name: "Idle+RecLast", + prepFunc: func(fc *FragmentCollector) {}, + recType: RecLast, + }, + { + name: "Collecting+RecFull", + prepFunc: func(fc *FragmentCollector) { + _ = fc.Append(RecFirst, []byte("data")) + }, + recType: RecFull, + }, + { + name: "Collecting+RecFirst", + prepFunc: func(fc *FragmentCollector) { + _ = fc.Append(RecFirst, []byte("data")) + }, + recType: RecFirst, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + fc := NewFragmentCollector() + tt.prepFunc(fc) + err := fc.Append(tt.recType, []byte("payload")) + if err == nil { + t.Errorf("expected error for %s, got nil", tt.name) + } + }) + } +} + +func TestFragmentBufferSizeLimit(t *testing.T) { + // Exceed the maximum fragment buffer size. + fc := NewFragmentCollector() + + // Start collecting. + if err := fc.Append(RecFirst, make([]byte, maxFragmentBufferSize-1)); err != nil { + t.Fatalf("Append First: %v", err) + } + if fc.State() != FragmentCollecting { + t.Fatal("expected FragmentCollecting state") + } + + // This should exceed the limit. + err := fc.Append(RecMiddle, make([]byte, 10)) + if err == nil { + t.Fatal("expected error for exceeding buffer size limit") + } +} + +func TestFragmentReset(t *testing.T) { + fc := NewFragmentCollector() + if err := fc.Append(RecFirst, []byte("data")); err != nil { + t.Fatalf("Append: %v", err) + } + if fc.State() != FragmentCollecting { + t.Fatal("expected FragmentCollecting") + } + + fc.Reset() + if fc.State() != FragmentIdle { + t.Errorf("state after Reset = %d, want FragmentIdle", fc.State()) + } + if fc.IsComplete() { + t.Error("IsComplete() should be false after Reset") + } + if fc.buf.Len() != 0 { + t.Errorf("buffer len after Reset = %d, want 0", fc.buf.Len()) + } +} + +func TestBatchValidationErrors(t *testing.T) { + t.Run("EntryCountZero", func(t *testing.T) { + // Manually craft a batch with EntryCount=0. + buf := make([]byte, WalBatchHeaderSize) + binary.LittleEndian.PutUint16(buf[0:2], 0) + binary.LittleEndian.PutUint64(buf[2:10], 1) + binary.LittleEndian.PutUint32(buf[10:14], 0) // entryCount = 0 + binary.LittleEndian.PutUint32(buf[14:18], 0) + _, err := DecodeWalBatch(buf) + if err == nil { + t.Fatal("expected error for entryCount=0") + } + }) + + t.Run("EntryCountExceedsMax", func(t *testing.T) { + buf := make([]byte, WalBatchHeaderSize) + binary.LittleEndian.PutUint16(buf[0:2], 0) + binary.LittleEndian.PutUint64(buf[2:10], 1) + binary.LittleEndian.PutUint32(buf[10:14], MaxWalBatchEntryCount+1) + binary.LittleEndian.PutUint32(buf[14:18], 1) + _, err := DecodeWalBatch(buf) + if err == nil { + t.Fatalf("expected error for entryCount > max") + } + }) + + t.Run("EntriesSizeMismatch", func(t *testing.T) { + buf := make([]byte, WalBatchHeaderSize) + binary.LittleEndian.PutUint16(buf[0:2], 0) + binary.LittleEndian.PutUint64(buf[2:10], 1) + binary.LittleEndian.PutUint32(buf[10:14], 1) + binary.LittleEndian.PutUint32(buf[14:18], 100) // entriesSize=100 but no data after header + _, err := DecodeWalBatch(buf) + if err == nil { + t.Fatal("expected error for entriesSize mismatch") + } + }) + + t.Run("DataTooShort", func(t *testing.T) { + _, err := DecodeWalBatch([]byte{1, 2, 3}) + if err == nil { + t.Fatal("expected error for data too short") + } + }) + + t.Run("EmptyEntriesSlice", func(t *testing.T) { + _, err := EncodeWalBatch(1, nil) + if err == nil { + t.Fatal("expected error for empty entries slice") + } + }) +} diff --git a/wal/validate.go b/wal/validate.go new file mode 100644 index 0000000..484f670 --- /dev/null +++ b/wal/validate.go @@ -0,0 +1,112 @@ +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 +} diff --git a/wal/validate_test.go b/wal/validate_test.go new file mode 100644 index 0000000..19741fa --- /dev/null +++ b/wal/validate_test.go @@ -0,0 +1,106 @@ +package wal + +import ( + "strings" + "testing" + + "github.com/dailz/go-kv/config" +) + +func makeEntry(opType uint8, valueKind uint8, keySize, valSize int) *WalEntry { + key := make([]byte, keySize) + for i := range key { + key[i] = byte('a' + i%26) + } + val := make([]byte, valSize) + for i := range val { + val[i] = byte('x') + } + return &WalEntry{OpType: opType, ValueKind: valueKind, Key: key, Value: val} +} + +func TestValidateBatchLimitsPass(t *testing.T) { + cfg := config.Defaults() + entries := make([]*WalEntry, 100) + for i := range entries { + entries[i] = makeEntry(OpPut, VKInline, 10, 10) + } + if err := ValidateBatchLimits(entries, &cfg); err != nil { + t.Fatalf("expected nil error, got: %v", err) + } +} + +func TestValidateEntryCountExceeded(t *testing.T) { + cfg := config.Defaults() + entries := make([]*WalEntry, 10001) + for i := range entries { + entries[i] = makeEntry(OpPut, VKInline, 10, 10) + } + err := ValidateBatchLimits(entries, &cfg) + if err == nil { + t.Fatal("expected error for entry count exceeded") + } + if !strings.Contains(err.Error(), "entry count") { + t.Fatalf("expected error containing 'entry count', got: %v", err) + } +} + +func TestValidateKeyTooBig(t *testing.T) { + cfg := config.Defaults() + // 5KB key exceeds 4KB limit + entries := []*WalEntry{makeEntry(OpPut, VKInline, 5*1024, 10)} + err := ValidateBatchLimits(entries, &cfg) + if err == nil { + t.Fatal("expected error for key too big") + } + if !strings.Contains(err.Error(), "key") { + t.Fatalf("expected error containing 'key', got: %v", err) + } +} + +func TestValidateValueTooBig(t *testing.T) { + cfg := config.Defaults() + // 5KB inline value exceeds 4KB limit + entries := []*WalEntry{makeEntry(OpPut, VKInline, 10, 5*1024)} + err := ValidateBatchLimits(entries, &cfg) + if err == nil { + t.Fatal("expected error for value too big") + } + if !strings.Contains(err.Error(), "value") { + t.Fatalf("expected error containing 'value', got: %v", err) + } +} + +func TestValidateEmptyBatch(t *testing.T) { + cfg := config.Defaults() + err := ValidateBatchLimits(nil, &cfg) + if err == nil { + t.Fatal("expected error for empty batch") + } + if !strings.Contains(err.Error(), "entry count") && !strings.Contains(err.Error(), "> 0") { + t.Fatalf("expected error about empty batch, got: %v", err) + } + + err = ValidateBatchLimits([]*WalEntry{}, &cfg) + if err == nil { + t.Fatal("expected error for empty batch") + } +} + +func TestValidateTotalSizeExceeded(t *testing.T) { + cfg := config.Defaults() + // Create entries that collectively exceed MaxBatchSize (4MB). + // Each entry with 1000-byte key + 1000-byte value ≈ 2012 bytes encoded. + // 2100 entries × ~2012 ≈ ~4.2MB > 4MB + entries := make([]*WalEntry, 2100) + for i := range entries { + entries[i] = makeEntry(OpPut, VKInline, 1000, 1000) + } + err := ValidateBatchLimits(entries, &cfg) + if err == nil { + t.Fatal("expected error for total size exceeded") + } + if !strings.Contains(err.Error(), "batch size") { + t.Fatalf("expected error containing 'batch size', got: %v", err) + } +}