From 069adff0a48523a4087b65ea025ad001fb9562fb Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 28 Oct 2026 16:37:09 +0000
Subject: [PATCH 1/2] feat(engine): values up to 2 MiB, page-slot reuse, and
 WAL frame cap

Values were capped at 4,000 bytes, one slotted-page tuple. Larger values (up
to MaxValueSize, 2 MiB) are now stored as a CRC-checked chain of chunk tuples
headed by a manifest; the B+ tree flags the manifest RID with bit 15 of the
slot ID. Small values take the existing path unchanged, existing files need no
migration, and the WAL, recovery and replication still log whole values.
Batches may carry up to 8 MiB of values. See docs/adr/0001-large-values.md.

- Reads verify chunk magic, chain length and CRC (ErrCorruptValue); overwrite
  and delete free a chain only after it validates.
- gRPC: MessageSizeServerOptions / MessageSizeDialOption raise the 4 MiB
  default; the server, followers and beastctl use them. Write-error status
  mapping is shared in grpc_status.go.
- WAL: frames over 32 MiB are refused on write, and a torn header claiming a
  huge length now reads as a torn tail instead of allocating up to 4 GiB.
- Storage: InsertTuple reuses tombstoned slots. Previously a recycled page's
  slot directory grew until it could not host a full tuple, and pages too full
  for one request were dropped from the reuse list; 300 overwrites of a 512 KB
  value grew the file to 2,120 pages, now 266, and 600 small overwrites added
  264 pages, now 3.

Tests: boundary sizes, overwrite/delete space reuse, crash and clean restart,
CAS and batches on large values, scans, corruption detection, a mixed-size
oracle across restarts, gRPC 6 MiB batches, and catch-up plus live replication
of 2 MiB values. Full suite passes with -race.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017mWMfRGkKdRgKc5mtSsWEG
---
 ROADMAP.md                             |   2 +-
 cmd/beastctl/client.go                 |   2 +
 cmd/server/main.go                     |   4 +-
 docs/adr/0001-large-values.md          |  39 +++++
 docs/roadmap-plan.md                   |   9 +-
 docs/v1-semantics.md                   |   5 +-
 internal/api/engine.go                 |   2 +-
 internal/api/engine_batch.go           |   6 +-
 internal/api/engine_conditional.go     |   2 +-
 internal/api/engine_large.go           | 196 +++++++++++++++++++++++++
 internal/api/engine_large_test.go      | 185 +++++++++++++++++++++++
 internal/api/engine_ops.go             |  36 ++---
 internal/api/engine_space.go           |  62 ++++++--
 internal/api/grpc_server.go            |  44 +-----
 internal/api/grpc_server_test.go       |  51 +++++++
 internal/api/grpc_status.go            |  37 +++++
 internal/api/limits_readonly_test.go   |  14 +-
 internal/replication/large_rep_test.go |  84 +++++++++++
 internal/storage/page.go               |  36 +++--
 internal/storage/page_test.go          |  17 +++
 internal/wal/record.go                 |   9 ++
 internal/wal/wal.go                    |   4 +
 internal/wal/wal_test.go               |  10 ++
 23 files changed, 750 insertions(+), 106 deletions(-)
 create mode 100644 docs/adr/0001-large-values.md
 create mode 100644 internal/api/engine_large.go
 create mode 100644 internal/api/engine_large_test.go
 create mode 100644 internal/api/grpc_status.go
 create mode 100644 internal/replication/large_rep_test.go

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..ff26145
--- /dev/null
+++ b/internal/api/engine_large.go
@@ -0,0 +1,196 @@
+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 the summed value bytes of one BatchWrite.
+	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 validation is
+// left in place rather than risk tombstoning tuples that belong to other keys.
+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
+	if e.walkChain(m, func(r storage.RID, _ []byte) { chunks = append(chunks, r) }) != nil {
+		return
+	}
+	for _, r := range chunks {
+		e.freeTuple(r)
+	}
+	e.freeTuple(manifestOf(rid))
+}
+
+// checkValueSizes enforces per-value and per-batch limits before anything is logged.
+func checkValueSizes(ops []BatchOperation) error {
+	total := 0
+	for _, op := range ops {
+		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_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_ops.go b/internal/api/engine_ops.go
index 49c9e96..c637262 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 values sum to more than MaxBatchBytes (8 MiB).
+	ErrBatchTooLarge = errors.New("batch values exceed maximum supported total of 8 MiB")
 	// 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..d235958
--- /dev/null
+++ b/internal/api/grpc_status.go
@@ -0,0 +1,37 @@
+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
+// so a maximal batch (MaxBatchBytes of values plus keys and framing) fits.
+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)
+	}
+}

From 259eff73ebab3605ee43c163ccc552b04e9b0731 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Wed, 28 Oct 2026 17:32:05 +0000
Subject: [PATCH 2/2] fix(engine): verify a chain's CRC before freeing it;
 count WAL framing in batch limit

- freeValue checked chunk magic and chain termination but not total length
  or CRC, so a mis-linked chain that still ended cleanly after the right
  number of chunks could free tuples owned by another key. It now verifies
  length and CRC first and leaves a failing chain in place.
- MaxBatchBytes counted only value bytes. Each WAL batch operation adds 15
  bytes of framing, so many tiny puts could commit a WAL frame larger than
  the replication message limit, and followers could never catch up on it.
  The limit now covers the encoded frame.

Both reported by Codex review on #1; regression tests reproduce each.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017mWMfRGkKdRgKc5mtSsWEG
---
 internal/api/engine_large.go             | 34 +++++--------
 internal/api/engine_large_review_test.go | 61 ++++++++++++++++++++++++
 internal/api/engine_limits.go            | 26 ++++++++++
 internal/api/engine_ops.go               |  4 +-
 internal/api/grpc_status.go              |  6 ++-
 5 files changed, 105 insertions(+), 26 deletions(-)
 create mode 100644 internal/api/engine_large_review_test.go
 create mode 100644 internal/api/engine_limits.go

diff --git a/internal/api/engine_large.go b/internal/api/engine_large.go
index ff26145..a18be3e 100644
--- a/internal/api/engine_large.go
+++ b/internal/api/engine_large.go
@@ -17,7 +17,7 @@ import (
 const (
 	// MaxValueSize is the largest value accepted by Put, PutIf and BatchWrite.
 	MaxValueSize = 2 << 20
-	// MaxBatchBytes bounds the summed value bytes of one BatchWrite.
+	// MaxBatchBytes bounds one BatchWrite's encoded WAL frame: its values plus 15 bytes per operation.
 	MaxBatchBytes = 8 << 20
 
 	largeSlotFlag = 0x8000
@@ -156,8 +156,10 @@ func (e *Engine) readValue(rid storage.RID) ([]byte, error) {
 	return out, nil
 }
 
-// freeValue releases the tuples behind rid. A chain that fails validation is
-// left in place rather than risk tombstoning tuples that belong to other keys.
+// 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)
@@ -168,7 +170,13 @@ func (e *Engine) freeValue(rid storage.RID) {
 		return
 	}
 	var chunks []storage.RID
-	if e.walkChain(m, func(r storage.RID, _ []byte) { chunks = append(chunks, r) }) != nil {
+	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 {
@@ -176,21 +184,3 @@ func (e *Engine) freeValue(rid storage.RID) {
 	}
 	e.freeTuple(manifestOf(rid))
 }
-
-// checkValueSizes enforces per-value and per-batch limits before anything is logged.
-func checkValueSizes(ops []BatchOperation) error {
-	total := 0
-	for _, op := range ops {
-		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_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_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 c637262..a164e58 100644
--- a/internal/api/engine_ops.go
+++ b/internal/api/engine_ops.go
@@ -18,8 +18,8 @@ var (
 	ErrRecordExists = errors.New("record already exists")
 	// 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 values sum to more than MaxBatchBytes (8 MiB).
-	ErrBatchTooLarge = errors.New("batch values exceed maximum supported total of 8 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")
 )
diff --git a/internal/api/grpc_status.go b/internal/api/grpc_status.go
index d235958..0277d09 100644
--- a/internal/api/grpc_status.go
+++ b/internal/api/grpc_status.go
@@ -8,8 +8,10 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-// MaxGRPCMessageSize is the gRPC message limit servers and clients should use
-// so a maximal batch (MaxBatchBytes of values plus keys and framing) fits.
+// 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.