diff --git a/ROADMAP.md b/ROADMAP.md index 0f4f488..15ddabb 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -122,7 +122,7 @@ Order is dependency-based, not a calendar promise. Initial planning range for on ## Beyond V1 gates -See the [roadmap plan](docs/roadmap-plan.md) for post-gate phases. Shipped so far: conditional writes (`ABSENT`/`PRESENT`/`VALUE_EQUALS` on Put, Delete and BatchWrite) and limit-based scan pagination. +See the [roadmap plan](docs/roadmap-plan.md) for post-gate phases. Shipped so far: conditional writes (`ABSENT`/`PRESENT`/`VALUE_EQUALS` on Put, Delete and BatchWrite), limit-based scan pagination, and values up to 2 MiB with 8 MiB batches ([ADR 0001](docs/adr/0001-large-values.md)). ## References diff --git a/cmd/beastctl/client.go b/cmd/beastctl/client.go index 35e68e5..207f0fc 100644 --- a/cmd/beastctl/client.go +++ b/cmd/beastctl/client.go @@ -8,6 +8,7 @@ import ( "time" beastv1 "github.com/ChromaBeast/beastdb/api/proto" + "github.com/ChromaBeast/beastdb/internal/api" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" ) @@ -49,6 +50,7 @@ func DialWithToken(addr, token string) (*Client, error) { opts := []grpc.DialOption{ grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock(), + api.MessageSizeDialOption(), } if token != "" { opts = append(opts, grpc.WithPerRPCCredentials(tokenAuth{token: token})) diff --git a/cmd/server/main.go b/cmd/server/main.go index 14d0c59..624c994 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -73,7 +73,7 @@ func main() { engine.StartAutoCheckpoint(ckptCtx, 10*time.Second, int64(*walCheckpointMB)<<20) tokenStore := initTokenStore(engine) - var grpcOpts []grpc.ServerOption + grpcOpts := api.MessageSizeServerOptions() if *tlsCert != "" && *tlsKey != "" { creds, err := loadServerTLS(*tlsCert, *tlsKey, *tlsCA) if err != nil { @@ -103,7 +103,7 @@ func main() { if err != nil { log.Fatalf("Failed to initialize follower client TLS: %v", err) } - conn, err := grpc.NewClient(*leaderAddr, grpc.WithTransportCredentials(clientCreds)) + conn, err := grpc.NewClient(*leaderAddr, grpc.WithTransportCredentials(clientCreds), api.MessageSizeDialOption()) if err != nil { log.Fatalf("Failed to dial leader node: %v", err) } diff --git a/docs/adr/0001-large-values.md b/docs/adr/0001-large-values.md new file mode 100644 index 0000000..658fbbb --- /dev/null +++ b/docs/adr/0001-large-values.md @@ -0,0 +1,39 @@ +# ADR 0001: Values larger than one page + +**Status:** accepted, 28 October 2026. + +## Context + +Every value lived in one slotted-page tuple, so values were capped at 4,000 bytes. Clients +storing documents, reports, serialized aggregates or index lists had to invent their own +chunking, each with its own atomicity and corruption bugs. + +## Decision + +- Values up to `MaxRecordSize` (4,000 bytes) are stored inline exactly as before. +- Values up to `MaxValueSize` (2 MiB) are stored as a chain of chunk tuples headed by a + 24-byte manifest tuple (magic, version, length, CRC32, chunk count, first chunk RID). Each + chunk carries a magic byte and its successor's RID; page 0 (the meta page) ends the chain. +- The B+ tree points at the manifest with bit 15 of the RID's slot ID set. A 4 KB page holds + at most ~1,000 slots, so inline RIDs never use that bit. +- The WAL still logs the whole value in one frame (`OpPut`) or one batch frame, so recovery + and replication are unchanged: replay re-chains the value. A batch's values are capped at + `MaxBatchBytes` (8 MiB) and a WAL frame at 32 MiB. +- Reads verify chunk magic, chain length and CRC and return `ErrCorruptValue` on mismatch. + Overwrite and delete free the chain only after it validates, so a damaged chain is leaked + rather than allowed to tombstone tuples owned by other keys. + +## Consequences + +- No migration: existing files contain no flagged RIDs. A file containing large values, + opened by an older binary, fails reads of those keys with `ErrSlotOutOfBounds` instead of + returning manifest bytes as data. Downgrade after writing large values is not supported. +- gRPC's default 4 MiB message limit is raised with `api.MessageSizeServerOptions` and + `api.MessageSizeDialOption`; the server, followers and `beastctl` use them, and clients + that send large batches must too. +- A large write holds the engine's write lock while it writes ~260 chunk tuples per MiB. +- Writing this exposed two allocator defects, fixed alongside: inserts never reused + tombstoned slot entries, so the slot directory of a recycled page grew until it could no + longer host a full tuple; and pages too full for one request were dropped from the reuse + list for good. 300 overwrites of a 512 KB value now keep the file at two copies' worth of + pages instead of growing it eightfold. diff --git a/docs/roadmap-plan.md b/docs/roadmap-plan.md index 97c0901..0b9f92f 100644 --- a/docs/roadmap-plan.md +++ b/docs/roadmap-plan.md @@ -41,10 +41,11 @@ Order is dependency-based. Each phase ends with a runnable demo, tests and docs. ### Phase B — Scan and key ergonomics -1. **Paginated scans:** `limit` on `ScanRequest` and resume by last key + 1 are **implemented**. A server-enforced maximum page size and reverse/prefix scans remain. -2. **Reverse and prefix scans** over the uint64 range, with documented bounds semantics. -3. **Optional wide-key mode design.** Options to evaluate and decide with a written ADR: (a) keep uint64 and ship a documented collision-safe pattern plus a client helper, or (b) add a variable-length key space. Decision gated on benchmarks and on-disk-format migration cost. -4. **Key-layout helper library** (client side, optional convention, not required by the engine). +1. **Large values:** values up to 2 MiB, chained across pages with CRC checks, and 8 MiB batches are **implemented** ([ADR 0001](adr/0001-large-values.md)). +2. **Paginated scans:** `limit` on `ScanRequest` and resume by last key + 1 are **implemented**. A server-enforced maximum page size and reverse/prefix scans remain. +3. **Reverse and prefix scans** over the uint64 range, with documented bounds semantics. +4. **Optional wide-key mode design.** Options to evaluate and decide with a written ADR: (a) keep uint64 and ship a documented collision-safe pattern plus a client helper, or (b) add a variable-length key space. Decision gated on benchmarks and on-disk-format migration cost. +5. **Key-layout helper library** (client side, optional convention, not required by the engine). **Exit:** a client can page through a 10M-key range with bounded memory on both sides. diff --git a/docs/v1-semantics.md b/docs/v1-semantics.md index 5924c0c..366356c 100644 --- a/docs/v1-semantics.md +++ b/docs/v1-semantics.md @@ -8,7 +8,8 @@ - **Key Format:** Unsigned 64-bit integer (`uint64`). - **Value Format:** Opaque byte slice (`[]byte`). -- **Maximum Record Size:** 4,000 bytes per record (bounded by 4KB slotted page capacity). Payloads exceeding this limit return `ErrRecordTooLarge`. +- **Maximum Value Size:** 2 MiB (`MaxValueSize`). Values up to 4,000 bytes (`MaxRecordSize`) are stored inline in one slotted-page tuple; larger ones are chained across pages and checked with a CRC32 on read ([ADR 0001](adr/0001-large-values.md)). Larger payloads return `ErrRecordTooLarge`. +- **Batch Size:** the values in one `BatchWrite` may total at most 8 MiB (`MaxBatchBytes`, `ErrBatchTooLarge`). gRPC clients sending large values must raise the 4 MiB default message limit (`api.MessageSizeDialOption`). - **Key Partitioning Convention:** Applications may optionally pack higher-order bits for table/collection prefixing (e.g., bits 56–63 for partition ID), but the engine treats keys as scalar `uint64` values. --- @@ -28,7 +29,7 @@ ### `BatchWrite(puts []PutOp, deletes []DeleteOp)` * **Durability Guarantee:** The entire batch is serialized into a single `OpBatch` WAL frame and synced in a single I/O operation. * **Atomicity:** All-or-nothing. On crash recovery, uncommitted or partial batches are rolled back; fully synced batches are replayed completely. -* **Limits:** Recommended batch limit of 1,000 operations or 2 MB aggregate payload. +* **Limits:** Values in one batch may total at most 8 MiB; each value at most 2 MiB. ### Conditional writes: `Put`/`Delete` with `Condition`, `BatchWrite` with `conditions` * **Conditions:** `ABSENT` (key must not exist), `PRESENT` (key must exist), `VALUE_EQUALS` (key must exist with exactly the given bytes). Omitted or `NONE` means unconditional. diff --git a/internal/api/engine.go b/internal/api/engine.go index 85cb95f..19c13dd 100644 --- a/internal/api/engine.go +++ b/internal/api/engine.go @@ -22,7 +22,7 @@ type Engine struct { secIndex *index.SecondaryIndex readOnly bool failed error // set by halt; blocks writes until restart - reclaim map[uint64]struct{} // data pages holding tombstoned tuples + reclaim map[uint64]int // data page -> free bytes after compaction, or unknownFree mu sync.RWMutex } diff --git a/internal/api/engine_batch.go b/internal/api/engine_batch.go index a04bce4..1fff44e 100644 --- a/internal/api/engine_batch.go +++ b/internal/api/engine_batch.go @@ -46,10 +46,8 @@ func (e *Engine) BatchWriteIf(ops []BatchOperation, conds []KeyCondition) error if len(ops) == 0 { return ErrEmptyBatchWrite } - for _, op := range ops { - if op.Type == BatchOpPut && len(op.Value) > MaxRecordSize { - return ErrRecordTooLarge - } + if err := checkValueSizes(ops); err != nil { + return err } e.mu.Lock() diff --git a/internal/api/engine_conditional.go b/internal/api/engine_conditional.go index 12c432c..51e0445 100644 --- a/internal/api/engine_conditional.go +++ b/internal/api/engine_conditional.go @@ -75,7 +75,7 @@ func (e *Engine) PutIf(key uint64, value []byte, cond Condition) error { if e.IsReadOnly() { return ErrReadOnlyReplica } - if len(value) > MaxRecordSize { + if len(value) > MaxValueSize { return ErrRecordTooLarge } e.mu.Lock() diff --git a/internal/api/engine_large.go b/internal/api/engine_large.go new file mode 100644 index 0000000..a18be3e --- /dev/null +++ b/internal/api/engine_large.go @@ -0,0 +1,186 @@ +package api + +import ( + "encoding/binary" + "errors" + "hash/crc32" + + "github.com/ChromaBeast/beastdb/internal/storage" +) + +// Values up to MaxRecordSize live inline in one slotted-page tuple. Larger +// values, up to MaxValueSize, are stored as a chain of chunk tuples headed by a +// small manifest tuple; the B+ tree points at the manifest with largeSlotFlag +// set in the RID's slot ID. A 4 KB page holds at most ~1,000 slots, so bit 15 +// is never used by an inline RID, and older binaries reading a flagged RID fail +// with ErrSlotOutOfBounds instead of returning the manifest as data. +const ( + // MaxValueSize is the largest value accepted by Put, PutIf and BatchWrite. + MaxValueSize = 2 << 20 + // MaxBatchBytes bounds one BatchWrite's encoded WAL frame: its values plus 15 bytes per operation. + MaxBatchBytes = 8 << 20 + + largeSlotFlag = 0x8000 + chunkMagic = 0xBC + manifestMagic = 0xBD + manifestV1 = 1 + chunkHeader = 11 // magic(1) + next page(8) + next slot(2) + manifestSize = 24 // magic, version, length(4), crc32(4), chunks(4), first page(8), first slot(2) + chunkPayload = MaxRecordSize - chunkHeader + maxChunks = (MaxValueSize + chunkPayload - 1) / chunkPayload +) + +// ErrCorruptValue reports a large-value chain that fails structural or CRC checks. +var ErrCorruptValue = errors.New("engine: stored value chain is corrupt") + +func isLargeRID(rid storage.RID) bool { return rid.SlotID&largeSlotFlag != 0 } + +func manifestOf(rid storage.RID) storage.RID { + rid.SlotID &^= largeSlotFlag + return rid +} + +// insertValue stores value inline when it fits in one tuple, else as a chunk chain. +// Caller must hold e.mu exclusively. +func (e *Engine) insertValue(value []byte) (storage.RID, error) { + if len(value) <= MaxRecordSize { + return e.insertTuple(value) + } + n := (len(value) + chunkPayload - 1) / chunkPayload + written := make([]storage.RID, 0, n+1) + undo := func() { + for _, r := range written { + e.freeTuple(r) + } + } + buf := make([]byte, MaxRecordSize) + var next storage.RID // zero page ID ends the chain; page 0 is the meta page + // Write the tail first so every chunk can name its successor. + for i := n - 1; i >= 0; i-- { + start := i * chunkPayload + end := min(start+chunkPayload, len(value)) + t := buf[:chunkHeader+end-start] + t[0] = chunkMagic + binary.LittleEndian.PutUint64(t[1:9], next.PageID) + binary.LittleEndian.PutUint16(t[9:11], next.SlotID) + copy(t[chunkHeader:], value[start:end]) + rid, err := e.insertTuple(t) + if err != nil { + undo() + return storage.RID{}, err + } + written = append(written, rid) + next = rid + } + m := buf[:manifestSize] + m[0], m[1] = manifestMagic, manifestV1 + binary.LittleEndian.PutUint32(m[2:6], uint32(len(value))) + binary.LittleEndian.PutUint32(m[6:10], crc32.ChecksumIEEE(value)) + binary.LittleEndian.PutUint32(m[10:14], uint32(n)) + binary.LittleEndian.PutUint64(m[14:22], next.PageID) + binary.LittleEndian.PutUint16(m[22:24], next.SlotID) + rid, err := e.insertTuple(m) + if err != nil { + undo() + return storage.RID{}, err + } + rid.SlotID |= largeSlotFlag + return rid, nil +} + +type manifest struct { + length, crc, chunks uint32 + first storage.RID +} + +func (e *Engine) readManifest(rid storage.RID) (manifest, error) { + m, err := e.readTuple(manifestOf(rid)) + if err != nil { + return manifest{}, err + } + if len(m) != manifestSize || m[0] != manifestMagic || m[1] != manifestV1 { + return manifest{}, ErrCorruptValue + } + out := manifest{ + length: binary.LittleEndian.Uint32(m[2:6]), + crc: binary.LittleEndian.Uint32(m[6:10]), + chunks: binary.LittleEndian.Uint32(m[10:14]), + first: storage.RID{PageID: binary.LittleEndian.Uint64(m[14:22]), SlotID: binary.LittleEndian.Uint16(m[22:24])}, + } + if out.length > MaxValueSize || out.chunks == 0 || out.chunks > maxChunks { + return manifest{}, ErrCorruptValue + } + return out, nil +} + +// walkChain visits each chunk of a large value in order and verifies the chain +// ends exactly where the manifest says. visit receives the chunk's payload. +func (e *Engine) walkChain(m manifest, visit func(rid storage.RID, payload []byte)) error { + next := m.first + for i := uint32(0); i < m.chunks; i++ { + if next.PageID == 0 || isLargeRID(next) { + return ErrCorruptValue + } + t, err := e.readTuple(next) + if err != nil { + return err + } + if len(t) < chunkHeader || t[0] != chunkMagic { + return ErrCorruptValue + } + visit(next, t[chunkHeader:]) + next = storage.RID{PageID: binary.LittleEndian.Uint64(t[1:9]), SlotID: binary.LittleEndian.Uint16(t[9:11])} + } + if next.PageID != 0 { + return ErrCorruptValue + } + return nil +} + +// readValue returns a copy of the value at rid, reassembling and checking large values. +func (e *Engine) readValue(rid storage.RID) ([]byte, error) { + if !isLargeRID(rid) { + return e.readTuple(rid) + } + m, err := e.readManifest(rid) + if err != nil { + return nil, err + } + out := make([]byte, 0, m.length) + if err := e.walkChain(m, func(_ storage.RID, p []byte) { out = append(out, p...) }); err != nil { + return nil, err + } + if uint32(len(out)) != m.length || crc32.ChecksumIEEE(out) != m.crc { + return nil, ErrCorruptValue + } + return out, nil +} + +// freeValue releases the tuples behind rid. A chain that fails any check, +// including its length and CRC, is left in place: a mis-linked chain can end +// cleanly after the right number of chunks while running through another +// key's tuples, and freeing it would destroy that key's value. +func (e *Engine) freeValue(rid storage.RID) { + if !isLargeRID(rid) { + e.freeTuple(rid) + return + } + m, err := e.readManifest(rid) + if err != nil { + return + } + var chunks []storage.RID + size, crc := 0, uint32(0) + visit := func(r storage.RID, p []byte) { + chunks = append(chunks, r) + size += len(p) + crc = crc32.Update(crc, crc32.IEEETable, p) + } + if e.walkChain(m, visit) != nil || uint32(size) != m.length || crc != m.crc { + return + } + for _, r := range chunks { + e.freeTuple(r) + } + e.freeTuple(manifestOf(rid)) +} diff --git a/internal/api/engine_large_review_test.go b/internal/api/engine_large_review_test.go new file mode 100644 index 0000000..e7bd6a1 --- /dev/null +++ b/internal/api/engine_large_review_test.go @@ -0,0 +1,61 @@ +package api + +import ( + "encoding/binary" + "errors" + "testing" + + "github.com/ChromaBeast/beastdb/internal/storage" +) + +func chainRIDs(t *testing.T, e *Engine, key uint64) []storage.RID { + t.Helper() + rid, err := e.tree.Find(key) + if err != nil { + t.Fatal(err) + } + m, err := e.readManifest(rid) + if err != nil { + t.Fatal(err) + } + var out []storage.RID + if err := e.walkChain(m, func(r storage.RID, _ []byte) { out = append(out, r) }); err != nil { + t.Fatal(err) + } + return out +} + +// A mis-linked chain that still has the right length and terminates must not +// free chunks owned by another key: the value's length and CRC must check out first. +func TestFreeRefusesChainThatFailsCRC(t *testing.T) { + e := openLarge(t, t.TempDir()) + defer e.Close() + _ = e.Put(1, pattern(20<<10, 1)) + _ = e.Put(2, pattern(20<<10, 2)) + a, b := chainRIDs(t, e, 1), chainRIDs(t, e, 2) + page, _ := e.bpm.FetchPage(a[0].PageID) + tuple, _ := page.GetTuple(a[0].SlotID) + binary.LittleEndian.PutUint64(tuple[1:9], b[1].PageID) + binary.LittleEndian.PutUint16(tuple[9:11], b[1].SlotID) + _ = e.bpm.UnpinPage(a[0].PageID, true) + + if err := e.Delete(1); err != nil { + t.Fatal(err) + } + mustGet(t, e, 2, pattern(20<<10, 2)) +} + +// The WAL frames each batch operation with 15 bytes; that framing must count +// toward MaxBatchBytes so a committed batch always fits a replication message. +func TestBatchLimitCountsWALFraming(t *testing.T) { + e := openLarge(t, t.TempDir()) + defer e.Close() + n := MaxBatchBytes/16 + 1 // one-byte values: fits the old value-only limit + ops := make([]BatchOperation, n) + for i := range ops { + ops[i] = BatchOperation{Type: BatchOpPut, Key: uint64(i), Value: []byte{1}} + } + if err := e.BatchWrite(ops); !errors.Is(err, ErrBatchTooLarge) { + t.Fatalf("expected ErrBatchTooLarge for a batch whose WAL frame exceeds the limit, got %v", err) + } +} diff --git a/internal/api/engine_large_test.go b/internal/api/engine_large_test.go new file mode 100644 index 0000000..ed37cf6 --- /dev/null +++ b/internal/api/engine_large_test.go @@ -0,0 +1,185 @@ +package api + +import ( + "bytes" + "errors" + "math/rand" + "path/filepath" + "testing" +) + +func openLarge(t *testing.T, dir string) *Engine { + t.Helper() + e, err := NewEngine(filepath.Join(dir, "large.bin"), filepath.Join(dir, "large.wal"), 64) + if err != nil { + t.Fatalf("open engine: %v", err) + } + return e +} + +func pattern(n int, seed byte) []byte { + b := make([]byte, n) + for i := range b { + b[i] = byte(i*31) ^ seed + } + return b +} + +func mustGet(t *testing.T, e *Engine, key uint64, want []byte) { + t.Helper() + got, found, err := e.Get(key) + if err != nil || !found { + t.Fatalf("get %d: found=%v err=%v", key, found, err) + } + if !bytes.Equal(got, want) { + t.Fatalf("get %d: %d bytes back, want %d", key, len(got), len(want)) + } +} + +func TestLargeValueRoundTripAtBoundaries(t *testing.T) { + e := openLarge(t, t.TempDir()) + defer e.Close() + sizes := []int{0, 1, MaxRecordSize, MaxRecordSize + 1, chunkPayload * 3, chunkPayload*3 + 1, 1 << 20, MaxValueSize} + for i, n := range sizes { + if err := e.Put(uint64(i+1), pattern(n, byte(i))); err != nil { + t.Fatalf("put %d bytes: %v", n, err) + } + } + for i, n := range sizes { + mustGet(t, e, uint64(i+1), pattern(n, byte(i))) + } + if err := e.Put(99, make([]byte, MaxValueSize+1)); !errors.Is(err, ErrRecordTooLarge) { + t.Fatalf("expected ErrRecordTooLarge, got %v", err) + } +} + +func TestLargeValueOverwriteAndDeleteReuseSpace(t *testing.T) { + e := openLarge(t, t.TempDir()) + defer e.Close() + val := pattern(512<<10, 7) + for i := 0; i < 5; i++ { + if err := e.Put(1, val); err != nil { + t.Fatal(err) + } + } + afterWarmup := e.disk.NumPages() + for i := 0; i < 40; i++ { + _ = e.Put(1, pattern(512<<10, byte(i))) + } + if grown := e.disk.NumPages() - afterWarmup; grown > afterWarmup/2 { + t.Fatalf("overwrites leaked pages: %d pages after warm-up, %d more after 40 overwrites", afterWarmup, grown) + } + if err := e.Put(1, []byte("small again")); err != nil { + t.Fatal(err) + } + mustGet(t, e, 1, []byte("small again")) + if err := e.Delete(1); err != nil { + t.Fatal(err) + } + if _, found, _ := e.Get(1); found { + t.Fatal("deleted large value still readable") + } +} + +func TestLargeValueSurvivesCrashAndCleanRestart(t *testing.T) { + dir := t.TempDir() + e := openLarge(t, dir) + _ = e.Put(1, pattern(300<<10, 1)) + _ = e.Put(2, pattern(10, 2)) + _ = e.BatchWrite([]BatchOperation{{Type: BatchOpPut, Key: 3, Value: pattern(90<<10, 3)}, {Type: BatchOpDelete, Key: 2}}) + // Abrupt stop: WAL durable, pages never flushed. + _ = e.wal.Sync() + _ = e.wal.Close() + _ = e.disk.Close() + + e = openLarge(t, dir) + mustGet(t, e, 1, pattern(300<<10, 1)) + mustGet(t, e, 3, pattern(90<<10, 3)) + if _, found, _ := e.Get(2); found { + t.Fatal("batch delete lost in recovery") + } + if err := e.Close(); err != nil { + t.Fatal(err) + } + e = openLarge(t, dir) + defer e.Close() + mustGet(t, e, 1, pattern(300<<10, 1)) +} + +func TestLargeValueConditionsBatchesAndScans(t *testing.T) { + e := openLarge(t, t.TempDir()) + defer e.Close() + v1, v2 := pattern(50<<10, 1), pattern(60<<10, 2) + if err := e.PutIf(1, v1, Condition{Kind: CondAbsent}); err != nil { + t.Fatal(err) + } + if err := e.PutIf(1, v2, Condition{Kind: CondValueEquals, Value: v2}); !errors.Is(err, ErrConditionFailed) { + t.Fatalf("stale compare-and-swap must fail, got %v", err) + } + if err := e.PutIf(1, v2, Condition{Kind: CondValueEquals, Value: v1}); err != nil { + t.Fatalf("compare-and-swap on a large value: %v", err) + } + big := []BatchOperation{{Type: BatchOpPut, Key: 2, Value: make([]byte, MaxValueSize)}, {Type: BatchOpPut, Key: 3, Value: make([]byte, MaxValueSize)}, + {Type: BatchOpPut, Key: 4, Value: make([]byte, MaxValueSize)}, {Type: BatchOpPut, Key: 5, Value: make([]byte, MaxValueSize)}, + {Type: BatchOpPut, Key: 6, Value: []byte("x")}} + if err := e.BatchWrite(big); !errors.Is(err, ErrBatchTooLarge) { + t.Fatalf("expected ErrBatchTooLarge, got %v", err) + } + cur, err := e.Scan(0, 10) + if err != nil { + t.Fatal(err) + } + _, rid, ok, err := cur.Next() + if err != nil || !ok { + t.Fatalf("scan found nothing: %v", err) + } + got, err := e.ReadTuple(rid) + _ = cur.Close() + if err != nil || !bytes.Equal(got, v2) { + t.Fatalf("scan must return the whole large value: %d bytes, %v", len(got), err) + } +} + +func TestCorruptChunkIsDetected(t *testing.T) { + e := openLarge(t, t.TempDir()) + defer e.Close() + _ = e.Put(1, pattern(20<<10, 9)) + rid, _ := e.tree.Find(1) + m, _ := e.readManifest(rid) + page, _ := e.bpm.FetchPage(m.first.PageID) + tuple, _ := page.GetTuple(m.first.SlotID) + tuple[chunkHeader+5] ^= 0xFF + _ = e.bpm.UnpinPage(m.first.PageID, true) + if _, _, err := e.Get(1); !errors.Is(err, ErrCorruptValue) { + t.Fatalf("expected ErrCorruptValue, got %v", err) + } +} + +func TestMixedSizeOracleAcrossRestarts(t *testing.T) { + dir := t.TempDir() + e := openLarge(t, dir) + oracle := map[uint64][]byte{} + rng := rand.New(rand.NewSource(42)) + for round := 0; round < 4; round++ { + for op := 0; op < 80; op++ { + key := uint64(rng.Intn(30) + 1) + if rng.Intn(4) == 0 { + _ = e.Delete(key) + delete(oracle, key) + continue + } + sizes := []int{rng.Intn(200), rng.Intn(MaxRecordSize), MaxRecordSize + rng.Intn(20000), rng.Intn(200 << 10)} + val := pattern(sizes[rng.Intn(len(sizes))], byte(op)) + if err := e.Put(key, val); err != nil { + t.Fatalf("put: %v", err) + } + oracle[key] = val + } + _ = e.Close() + e = openLarge(t, dir) + for k, v := range oracle { + mustGet(t, e, k, v) + } + } + _ = e.Close() +} diff --git a/internal/api/engine_limits.go b/internal/api/engine_limits.go new file mode 100644 index 0000000..90f71a8 --- /dev/null +++ b/internal/api/engine_limits.go @@ -0,0 +1,26 @@ +package api + +// walBatchOpOverhead is the WAL batch framing per operation: type (1), key +// length (2), the 8-byte key and value length (4); see wal.EncodeBatchPayload. +const walBatchOpOverhead = 15 + +// checkValueSizes enforces per-value and per-batch limits before anything is +// logged. The batch limit covers the encoded WAL frame, framing included, so a +// committed batch always fits in one replication message. +func checkValueSizes(ops []BatchOperation) error { + total := 4 // operation count + for _, op := range ops { + total += walBatchOpOverhead + if op.Type != BatchOpPut { + continue + } + if len(op.Value) > MaxValueSize { + return ErrRecordTooLarge + } + total += len(op.Value) + } + if total > MaxBatchBytes { + return ErrBatchTooLarge + } + return nil +} diff --git a/internal/api/engine_ops.go b/internal/api/engine_ops.go index 49c9e96..a164e58 100644 --- a/internal/api/engine_ops.go +++ b/internal/api/engine_ops.go @@ -9,13 +9,17 @@ import ( "github.com/ChromaBeast/beastdb/internal/wal" ) +// MaxRecordSize is the largest value stored inline in a single slotted-page +// tuple; larger values up to MaxValueSize are chained (see engine_large.go). const MaxRecordSize = 4000 var ( // ErrRecordExists indicates a key is already present during PutIfAbsent. ErrRecordExists = errors.New("record already exists") - // ErrRecordTooLarge indicates the payload exceeds the 4,000 byte limit. - ErrRecordTooLarge = errors.New("record value exceeds maximum supported size of 4000 bytes") + // ErrRecordTooLarge indicates a value exceeds MaxValueSize (2 MiB). + ErrRecordTooLarge = errors.New("record value exceeds maximum supported size of 2 MiB") + // ErrBatchTooLarge indicates a batch's encoded WAL frame exceeds MaxBatchBytes (8 MiB). + ErrBatchTooLarge = errors.New("batch exceeds maximum supported size of 8 MiB (values plus 15 bytes per operation)") // ErrReadOnlyReplica indicates mutation attempts on a read-only replica node. ErrReadOnlyReplica = errors.New("writes are not permitted on read-only replica") ) @@ -25,7 +29,7 @@ func (e *Engine) Put(key uint64, value []byte) error { if e.IsReadOnly() { return ErrReadOnlyReplica } - if len(value) > MaxRecordSize { + if len(value) > MaxValueSize { return ErrRecordTooLarge } e.mu.Lock() @@ -38,7 +42,7 @@ func (e *Engine) PutIfAbsent(key uint64, value []byte) error { if e.IsReadOnly() { return ErrReadOnlyReplica } - if len(value) > MaxRecordSize { + if len(value) > MaxValueSize { return ErrRecordTooLarge } e.mu.Lock() @@ -85,23 +89,13 @@ func (e *Engine) Get(key uint64) ([]byte, bool, error) { return nil, false, err } - page, err := e.bpm.FetchPage(rid.PageID) - if err != nil { - return nil, false, err - } - - tuple, err := page.GetTuple(rid.SlotID) - _ = e.bpm.UnpinPage(rid.PageID, false) + val, err := e.readValue(rid) if err != nil { return nil, false, err } - - val := make([]byte, len(tuple)) - copy(val, tuple) return val, true, nil } - // Scan returns a streaming cursor iterator over the requested key range. // The engine's read lock is held for the cursor's entire lifetime and released // in cursor.Close() — this prevents concurrent writes from splitting a leaf @@ -116,19 +110,25 @@ func (e *Engine) Scan(startKey, endKey uint64) (*index.Cursor, error) { return cursor, nil } -// ReadTuple retrieves tuple bytes directly for a given RID. +// ReadTuple returns a copy of the value a B+ tree RID refers to, reassembling +// chained large values. func (e *Engine) ReadTuple(rid storage.RID) ([]byte, error) { + return e.readValue(rid) +} + +// readTuple copies the raw bytes of one slotted-page tuple. +func (e *Engine) readTuple(rid storage.RID) ([]byte, error) { page, err := e.bpm.FetchPage(rid.PageID) if err != nil { return nil, err } tuple, err := page.GetTuple(rid.SlotID) - _ = e.bpm.UnpinPage(rid.PageID, false) if err != nil { + _ = e.bpm.UnpinPage(rid.PageID, false) return nil, err } - val := make([]byte, len(tuple)) copy(val, tuple) + _ = e.bpm.UnpinPage(rid.PageID, false) return val, nil } diff --git a/internal/api/engine_space.go b/internal/api/engine_space.go index e0d7bd1..b2ca3f7 100644 --- a/internal/api/engine_space.go +++ b/internal/api/engine_space.go @@ -34,10 +34,12 @@ func (e *Engine) insertTuple(value []byte) (storage.RID, error) { _ = e.bpm.UnpinPage(e.activeDataPage, true) return storage.RID{PageID: e.activeDataPage, SlotID: slotID}, nil } + leftover := page.FreeSpace() _ = e.bpm.UnpinPage(e.activeDataPage, true) if err != storage.ErrPageFull { return storage.RID{}, err } + oldActive := e.activeDataPage newPage, newID, err := e.nextDataPage(len(value)) if err != nil { @@ -48,6 +50,9 @@ func (e *Engine) insertTuple(value []byte) (storage.RID, error) { return storage.RID{}, err } e.activeDataPage = newID + if leftover >= minTrackedFree { + e.trackFree(oldActive, leftover) + } slotID, err = newPage.InsertTuple(value) _ = e.bpm.UnpinPage(newID, err == nil) if err != nil { @@ -56,32 +61,60 @@ func (e *Engine) insertTuple(value []byte) (storage.RID, error) { return storage.RID{PageID: newID, SlotID: slotID}, nil } -// nextDataPage returns a pinned page with room for size bytes: a compacted page -// with dead tuples if one fits, otherwise a freshly allocated page. +const ( + // unknownFree marks a page whose tuples were tombstoned since it was last measured. + unknownFree = -1 + // minTrackedFree drops pages too full to be worth revisiting. + minTrackedFree = 64 +) + +// nextDataPage returns a pinned page with room for size bytes: a known page +// with enough free space if one exists, otherwise a freshly allocated page. +// Pages that are too full for this request stay tracked for smaller ones. func (e *Engine) nextDataPage(size int) (*storage.SlottedPage, uint64, error) { - for id := range e.reclaim { - delete(e.reclaim, id) + need := size + storage.SlotSize + for id, free := range e.reclaim { if id == e.activeDataPage { + delete(e.reclaim, id) + continue + } + if free != unknownFree && free < need { continue } p, err := e.bpm.FetchPage(id) if err != nil { + delete(e.reclaim, id) continue } p.Compact() - if p.FreeSpace() >= size+storage.SlotSize { - _ = e.bpm.UnpinPage(id, true) - p, err = e.bpm.FetchPage(id) - if err == nil { + got := p.FreeSpace() + _ = e.bpm.UnpinPage(id, true) + if got >= need { + delete(e.reclaim, id) + if p, err = e.bpm.FetchPage(id); err == nil { return p, id, nil } continue } - _ = e.bpm.UnpinPage(id, true) + if got < minTrackedFree { + delete(e.reclaim, id) + } else { + e.reclaim[id] = got + } } return e.bpm.NewPage() } +// trackFree remembers a page's free space for later inserts. +func (e *Engine) trackFree(id uint64, free int) { + if e.reclaim == nil { + e.reclaim = make(map[uint64]int) + } + if cur, ok := e.reclaim[id]; free == unknownFree || !ok || cur != unknownFree { + e.reclaim[id] = free + } +} + // freeTuple tombstones the tuple at rid and queues its page for space reuse. func (e *Engine) freeTuple(rid storage.RID) { page, err := e.bpm.FetchPage(rid.PageID) @@ -91,10 +124,7 @@ func (e *Engine) freeTuple(rid storage.RID) { _ = page.DeleteTuple(rid.SlotID) _ = e.bpm.UnpinPage(rid.PageID, true) if rid.PageID != e.activeDataPage { - if e.reclaim == nil { - e.reclaim = make(map[uint64]struct{}) - } - e.reclaim[rid.PageID] = struct{}{} + e.trackFree(rid.PageID, unknownFree) } } @@ -106,7 +136,7 @@ func (e *Engine) applyPutInMemory(key uint64, value []byte) error { if err != nil && err != index.ErrKeyNotFound { return err } - rid, err := e.insertTuple(value) + rid, err := e.insertValue(value) if err != nil { return err } @@ -114,7 +144,7 @@ func (e *Engine) applyPutInMemory(key uint64, value []byte) error { return err } if hadOld { - e.freeTuple(oldRID) + e.freeValue(oldRID) } return nil } @@ -127,6 +157,6 @@ func (e *Engine) applyDeleteInMemory(key uint64) error { if err != nil { return err } - e.freeTuple(rid) + e.freeValue(rid) return e.tree.Delete(key) } diff --git a/internal/api/grpc_server.go b/internal/api/grpc_server.go index 9c1a9f1..1fac303 100644 --- a/internal/api/grpc_server.go +++ b/internal/api/grpc_server.go @@ -2,7 +2,6 @@ package api import ( "context" - "errors" "net" beastv1 "github.com/ChromaBeast/beastdb/api/proto" @@ -66,24 +65,15 @@ func (s *GRPCServer) Put(ctx context.Context, req *beastv1.PutRequest) (*beastv1 if s.engine.IsReadOnly() { return nil, status.Errorf(codes.FailedPrecondition, "writes are not permitted on read replica") } - if len(req.Value) > MaxRecordSize { - return nil, status.Errorf(codes.InvalidArgument, "record value exceeds max 4000 bytes: %d bytes provided", len(req.Value)) + if len(req.Value) > MaxValueSize { + return nil, status.Errorf(codes.InvalidArgument, "record value exceeds max %d bytes: %d bytes provided", MaxValueSize, len(req.Value)) } cond, err := conditionFromProto(req.GetCondition()) if err != nil { return nil, err } if err := s.engine.PutIf(req.Key, req.Value, cond); err != nil { - if errors.Is(err, ErrConditionFailed) { - return nil, status.Errorf(codes.Aborted, "precondition failed") - } - if errors.Is(err, ErrReadOnlyReplica) { - return nil, status.Errorf(codes.FailedPrecondition, "writes are not permitted on read replica") - } - if errors.Is(err, ErrRecordTooLarge) { - return nil, status.Errorf(codes.InvalidArgument, "record value exceeds max 4000 bytes") - } - return nil, status.Errorf(codes.Internal, "put error: %v", err) + return nil, writeStatus(err, "put") } return &beastv1.PutResponse{Success: true}, nil } @@ -97,15 +87,8 @@ func (s *GRPCServer) Delete(ctx context.Context, req *beastv1.DeleteRequest) (*b if err != nil { return nil, err } - err = s.engine.DeleteIf(req.Key, cond) - if errors.Is(err, ErrConditionFailed) { - return nil, status.Errorf(codes.Aborted, "precondition failed") - } - if errors.Is(err, ErrReadOnlyReplica) { - return nil, status.Errorf(codes.FailedPrecondition, "writes are not permitted on read replica") - } - if err != nil { - return nil, status.Errorf(codes.Internal, "delete error: %v", err) + if err := s.engine.DeleteIf(req.Key, cond); err != nil { + return nil, writeStatus(err, "delete") } return &beastv1.DeleteResponse{Success: true}, nil } @@ -167,11 +150,7 @@ func (s *GRPCServer) BatchWrite(ctx context.Context, req *beastv1.BatchWriteRequ return nil, status.Errorf(codes.InvalidArgument, "unsupported operation type: %v", op.OpType) } - ops[i] = BatchOperation{ - Type: opType, - Key: op.Key, - Value: op.Value, - } + ops[i] = BatchOperation{Type: opType, Key: op.Key, Value: op.Value} } if s.engine.IsReadOnly() { @@ -188,16 +167,7 @@ func (s *GRPCServer) BatchWrite(ctx context.Context, req *beastv1.BatchWriteRequ } if err := s.engine.BatchWriteIf(ops, conds); err != nil { - if errors.Is(err, ErrConditionFailed) { - return nil, status.Errorf(codes.Aborted, "precondition failed") - } - if errors.Is(err, ErrReadOnlyReplica) { - return nil, status.Errorf(codes.FailedPrecondition, "writes are not permitted on read replica") - } - if errors.Is(err, ErrRecordTooLarge) { - return nil, status.Errorf(codes.InvalidArgument, "batch contains record exceeding max 4000 bytes: %v", err) - } - return nil, status.Errorf(codes.Internal, "batch write failed: %v", err) + return nil, writeStatus(err, "batch write") } return &beastv1.BatchWriteResponse{ diff --git a/internal/api/grpc_server_test.go b/internal/api/grpc_server_test.go index c7f743e..a999c39 100644 --- a/internal/api/grpc_server_test.go +++ b/internal/api/grpc_server_test.go @@ -131,3 +131,54 @@ func BenchmarkGRPCGet(b *testing.B) { _, _ = client.Get(ctx, req) } } + +// Values above gRPC's 4 MiB default, batched, must round-trip with the message-size options. +func TestGRPCLargeValuesWithMessageSizeOptions(t *testing.T) { + dir := t.TempDir() + engine, err := NewEngine(filepath.Join(dir, "big.bin"), filepath.Join(dir, "big.wal"), 50) + if err != nil { + t.Fatal(err) + } + defer engine.Close() + server := NewGRPCServer(engine, MessageSizeServerOptions()...) + lis, _ := net.Listen("tcp", "127.0.0.1:0") + go func() { _ = server.Serve(lis) }() + defer server.Stop() + conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()), MessageSizeDialOption()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + client := beastv1.NewBeastDBServiceClient(conn) + ctx := context.Background() + + var ops []*beastv1.BatchOperation + for k := uint64(1); k <= 3; k++ { + ops = append(ops, &beastv1.BatchOperation{OpType: beastv1.BatchOpType_BATCH_OP_TYPE_PUT, Key: k, Value: bytes.Repeat([]byte{byte(k)}, MaxValueSize)}) + } + if _, err := client.BatchWrite(ctx, &beastv1.BatchWriteRequest{Operations: ops}); err != nil { + t.Fatalf("6 MiB batch: %v", err) + } + got, err := client.Get(ctx, &beastv1.GetRequest{Key: 2}) + if err != nil || !bytes.Equal(got.Value, ops[1].Value) { + t.Fatalf("get large value: %v", err) + } + stream, err := client.Scan(ctx, &beastv1.ScanRequest{StartKey: 1, EndKey: 3}) + if err != nil { + t.Fatal(err) + } + n := 0 + for { + kv, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil || len(kv.Value) != MaxValueSize { + t.Fatalf("scan large value: %v", err) + } + n++ + } + if n != 3 { + t.Fatalf("scan returned %d values", n) + } +} diff --git a/internal/api/grpc_status.go b/internal/api/grpc_status.go new file mode 100644 index 0000000..0277d09 --- /dev/null +++ b/internal/api/grpc_status.go @@ -0,0 +1,39 @@ +package api + +import ( + "errors" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// MaxGRPCMessageSize is the gRPC message limit servers and clients should use. +// A committed batch's WAL frame is at most MaxBatchBytes, so the 1 MiB margin +// covers the replication message around it; a client request with protobuf +// framing near the limit is rejected by gRPC before anything is written. +const MaxGRPCMessageSize = MaxBatchBytes + 1<<20 + +// MessageSizeServerOptions raises gRPC's 4 MiB default to MaxGRPCMessageSize. +func MessageSizeServerOptions() []grpc.ServerOption { + return []grpc.ServerOption{grpc.MaxRecvMsgSize(MaxGRPCMessageSize), grpc.MaxSendMsgSize(MaxGRPCMessageSize)} +} + +// MessageSizeDialOption is the client-side counterpart of MessageSizeServerOptions. +func MessageSizeDialOption() grpc.DialOption { + return grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(MaxGRPCMessageSize), grpc.MaxCallSendMsgSize(MaxGRPCMessageSize)) +} + +// writeStatus maps engine write errors to gRPC status codes. +func writeStatus(err error, op string) error { + switch { + case errors.Is(err, ErrConditionFailed): + return status.Errorf(codes.Aborted, "precondition failed") + case errors.Is(err, ErrReadOnlyReplica): + return status.Errorf(codes.FailedPrecondition, "writes are not permitted on read replica") + case errors.Is(err, ErrRecordTooLarge), errors.Is(err, ErrBatchTooLarge): + return status.Errorf(codes.InvalidArgument, "%s: %v", op, err) + default: + return status.Errorf(codes.Internal, "%s error: %v", op, err) + } +} diff --git a/internal/api/limits_readonly_test.go b/internal/api/limits_readonly_test.go index d34d31d..14bf9cd 100644 --- a/internal/api/limits_readonly_test.go +++ b/internal/api/limits_readonly_test.go @@ -26,17 +26,17 @@ func TestEngineRecordSizeLimit(t *testing.T) { } defer engine.Close() - // Exactly 4000 bytes should succeed - validPayload := bytes.Repeat([]byte("A"), 4000) + // Exactly MaxValueSize bytes should succeed (chained beyond MaxRecordSize) + validPayload := bytes.Repeat([]byte("A"), MaxValueSize) if err := engine.Put(1, validPayload); err != nil { - t.Fatalf("expected 4000 byte put to succeed, got: %v", err) + t.Fatalf("expected MaxValueSize put to succeed, got: %v", err) } - // 4001 bytes must fail with ErrRecordTooLarge - tooLargePayload := bytes.Repeat([]byte("B"), 4001) + // One byte more must fail with ErrRecordTooLarge + tooLargePayload := bytes.Repeat([]byte("B"), MaxValueSize+1) err = engine.Put(2, tooLargePayload) if !errors.Is(err, ErrRecordTooLarge) { - t.Fatalf("expected ErrRecordTooLarge for 4001 bytes, got: %v", err) + t.Fatalf("expected ErrRecordTooLarge for MaxValueSize+1 bytes, got: %v", err) } // BatchWrite with oversized record must also fail @@ -134,7 +134,7 @@ func TestGRPCReadOnlyAndSizeValidation(t *testing.T) { // Disable read-only to test payload size rejection engine.SetReadOnly(false) - _, err = client.Put(ctx, &beastv1.PutRequest{Key: 1, Value: bytes.Repeat([]byte("X"), 4001)}) + _, err = client.Put(ctx, &beastv1.PutRequest{Key: 1, Value: bytes.Repeat([]byte("X"), MaxValueSize+1)}) if status.Code(err) != codes.InvalidArgument { t.Fatalf("expected InvalidArgument on oversized Put, got: %v", err) } diff --git a/internal/replication/large_rep_test.go b/internal/replication/large_rep_test.go new file mode 100644 index 0000000..cbd06b2 --- /dev/null +++ b/internal/replication/large_rep_test.go @@ -0,0 +1,84 @@ +package replication + +import ( + "bytes" + "context" + "net" + "path/filepath" + "testing" + "time" + + "github.com/ChromaBeast/beastdb/internal/api" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +func filled(n int, b byte) []byte { return bytes.Repeat([]byte{b}, n) } + +// Large values and multi-megabyte batches must replicate in catch-up and live modes. +func TestLargeValuesReplicate(t *testing.T) { + dir := t.TempDir() + leader, err := api.NewEngine(filepath.Join(dir, "l.bin"), filepath.Join(dir, "l.wal"), 50) + if err != nil { + t.Fatal(err) + } + defer leader.Close() + rep := NewLeaderServer(leader.WALPath()) + srv := grpc.NewServer(api.MessageSizeServerOptions()...) + rep.RegisterService(srv) + leader.SetCommitObserver(rep) + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + go func() { _ = srv.Serve(lis) }() + defer srv.Stop() + + catchUp := filled(1500<<10, 'c') + if err := leader.Put(1, catchUp); err != nil { + t.Fatal(err) + } + + follower, err := api.NewEngine(filepath.Join(dir, "f.bin"), filepath.Join(dir, "f.wal"), 50) + if err != nil { + t.Fatal(err) + } + defer follower.Close() + conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()), api.MessageSizeDialOption()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = NewFollowerClient("big", follower, conn).SyncLoop(ctx) }() + + live := filled(api.MaxValueSize, 'l') + batch := []api.BatchOperation{ + {Type: api.BatchOpPut, Key: 3, Value: filled(api.MaxValueSize, 'x')}, + {Type: api.BatchOpPut, Key: 4, Value: filled(api.MaxValueSize, 'y')}, + {Type: api.BatchOpPut, Key: 5, Value: filled(api.MaxValueSize, 'z')}, + } + time.Sleep(100 * time.Millisecond) + if err := leader.Put(2, live); err != nil { + t.Fatal(err) + } + if err := leader.BatchWrite(batch); err != nil { + t.Fatal(err) + } + + want := map[uint64][]byte{1: catchUp, 2: live, 3: batch[0].Value, 4: batch[1].Value, 5: batch[2].Value} + deadline := time.Now().Add(5 * time.Second) + for k, v := range want { + for { + got, found, err := follower.Get(k) + if err == nil && found && bytes.Equal(got, v) { + break + } + if time.Now().After(deadline) { + t.Fatalf("follower missing key %d: found=%v len=%d err=%v", k, found, len(got), err) + } + time.Sleep(20 * time.Millisecond) + } + } +} diff --git a/internal/storage/page.go b/internal/storage/page.go index a42953f..e375f6d 100644 --- a/internal/storage/page.go +++ b/internal/storage/page.go @@ -59,31 +59,41 @@ func (p *SlottedPage) FreeSpace() int { } // InsertTuple stores variable-length data, growing data upwards and slot downwards. +// A tombstoned slot is reused before the directory grows, so a page that is +// filled and emptied repeatedly keeps its full capacity. func (p *SlottedPage) InsertTuple(tuple []byte) (uint16, error) { tupleLen := len(tuple) - needed := tupleLen + SlotSize - + h := p.Header() + slotID, reuse := p.freeSlot(h) + needed := tupleLen + if !reuse { + needed += SlotSize + } if p.FreeSpace() < needed { return 0, ErrPageFull } - h := p.Header() h.Upper -= uint16(tupleLen) copy(p.data[h.Upper:h.Upper+uint16(tupleLen)], tuple) - - slotID := h.SlotCount - WriteSlot(p.data, slotID, Slot{ - Offset: h.Upper, - Length: uint16(tupleLen), - }) - - h.SlotCount++ - h.Lower += SlotSize + WriteSlot(p.data, slotID, Slot{Offset: h.Upper, Length: uint16(tupleLen)}) + if !reuse { + h.SlotCount++ + h.Lower += SlotSize + } WritePageHeader(p.data, h) - return slotID, nil } +// freeSlot returns the first tombstoned slot, or the next new slot ID. +func (p *SlottedPage) freeSlot(h PageHeader) (uint16, bool) { + for i := uint16(0); i < h.SlotCount; i++ { + if ReadSlot(p.data, i).Offset == 0 { + return i, true + } + } + return h.SlotCount, false +} + // GetTuple retrieves the tuple bytes referenced by slotID. func (p *SlottedPage) GetTuple(slotID uint16) ([]byte, error) { h := p.Header() diff --git a/internal/storage/page_test.go b/internal/storage/page_test.go index faf3066..3e6b824 100644 --- a/internal/storage/page_test.go +++ b/internal/storage/page_test.go @@ -115,3 +115,20 @@ func BenchmarkSlottedPageGet(b *testing.B) { _, _ = page.GetTuple(slotID) } } + +// A page filled and emptied repeatedly must keep its capacity: tombstoned slots are reused. +func TestInsertReusesTombstonedSlots(t *testing.T) { + p := NewSlottedPage(1) + tuple := make([]byte, 4000) + for i := 0; i < 50; i++ { + slot, err := p.InsertTuple(tuple) + if err != nil { + t.Fatalf("round %d: page lost capacity: %v", i, err) + } + _ = p.DeleteTuple(slot) + p.Compact() + } + if h := p.Header(); h.SlotCount != 1 { + t.Fatalf("slot directory grew to %d entries", h.SlotCount) + } +} diff --git a/internal/wal/record.go b/internal/wal/record.go index 1e06159..0a22811 100644 --- a/internal/wal/record.go +++ b/internal/wal/record.go @@ -12,6 +12,10 @@ const ( // RecLen (4B) + LSN (8B) + Type (1B) + KeyLen (2B) + ValLen (4B) + CRC32 (4B) = 23 Bytes. HeaderSize = 23 + // MaxFrameSize bounds one encoded record. The engine caps batches at 8 MiB + // of values, so anything larger on disk can only be a torn or corrupt header. + MaxFrameSize = 32 << 20 + // Operation Types OpPut byte = 0x01 OpDelete byte = 0x02 @@ -23,6 +27,7 @@ var ( ErrTornWrite = errors.New("wal: torn write or CRC32 checksum mismatch") ErrRecordTooSmall = errors.New("wal: record length smaller than header") ErrPayloadTooLarge = errors.New("wal: record payload exceeds max limit") + ErrFrameTooLarge = errors.New("wal: encoded record exceeds MaxFrameSize") ) // Record represents a single state mutation logged for durability. @@ -70,6 +75,10 @@ func DecodeRecord(r io.Reader) (*Record, error) { if recLen < HeaderSize { return nil, ErrRecordTooSmall } + if recLen > MaxFrameSize { + // A torn header can claim up to 4 GiB; treat it as a torn tail instead of allocating. + return nil, ErrTornWrite + } lsn := binary.LittleEndian.Uint64(header[4:12]) opType := header[12] diff --git a/internal/wal/wal.go b/internal/wal/wal.go index d637fb6..2d038cc 100644 --- a/internal/wal/wal.go +++ b/internal/wal/wal.go @@ -66,6 +66,10 @@ func (w *WAL) Write(opType byte, key, value []byte) (uint64, error) { } encoded := EncodeRecord(rec) + if len(encoded) > MaxFrameSize { + w.currentLSN-- + return 0, ErrFrameTooLarge + } if _, err := w.file.Write(encoded); err != nil { return 0, err } diff --git a/internal/wal/wal_test.go b/internal/wal/wal_test.go index 1aedf54..62076a8 100644 --- a/internal/wal/wal_test.go +++ b/internal/wal/wal_test.go @@ -2,6 +2,7 @@ package wal import ( "bytes" + "encoding/binary" "os" "path/filepath" "testing" @@ -132,3 +133,12 @@ func BenchmarkWALAppend(b *testing.B) { _, _ = wal.Write(OpPut, key, val) } } + +// A torn header claiming a huge length must read as a torn tail, not trigger a huge allocation. +func TestDecodeRejectsOversizedFrameHeader(t *testing.T) { + header := make([]byte, HeaderSize) + binary.LittleEndian.PutUint32(header[0:4], 0xFFFFFFF0) + if _, err := DecodeRecord(bytes.NewReader(header)); err != ErrTornWrite { + t.Fatalf("expected ErrTornWrite, got %v", err) + } +}