Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ROADMAP.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
2 changes: 2 additions & 0 deletions cmd/beastctl/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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}))
Expand Down
4 changes: 2 additions & 2 deletions cmd/server/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
}
Expand Down
39 changes: 39 additions & 0 deletions docs/adr/0001-large-values.md
Original file line number Diff line number Diff line change
@@ -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.
9 changes: 5 additions & 4 deletions docs/roadmap-plan.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
5 changes: 3 additions & 2 deletions docs/v1-semantics.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

---
Expand All @@ -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.
Expand Down
2 changes: 1 addition & 1 deletion internal/api/engine.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down
6 changes: 2 additions & 4 deletions internal/api/engine_batch.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
2 changes: 1 addition & 1 deletion internal/api/engine_conditional.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
186 changes: 186 additions & 0 deletions internal/api/engine_large.go
Original file line number Diff line number Diff line change
@@ -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))
}
61 changes: 61 additions & 0 deletions internal/api/engine_large_review_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
Loading
Loading