From 425847b48d2253284c7aaebde761f30876b9d7d4 Mon Sep 17 00:00:00 2001 From: ChromaBeast Date: Fri, 13 Nov 2026 11:16:21 +0530 Subject: [PATCH 1/2] fix(storage): keep the data file a checkpoint image; journal checkpoints Crash recovery replays the WAL from the meta page's checkpoint LSN on top of whatever pages are in the data file. That replay is logical and not idempotent: once a dirty page was evicted after a checkpoint, the data file already held later changes, and replaying them again (tuple inserts, slot reuse, compaction, splits) lost writes, resurrected deleted tuples or panicked in SlottedPage.Compact. A new lost-write crash test failed 15 of 20 seeds on the previous code, with or without the per-page fsync. The data file now only ever holds a complete checkpoint image: - Evicted dirty pages go to a scratch spill file (-spill): no fsync, no WAL barrier, discarded on restart. ReadPage prefers the spilled copy. - A checkpoint (BufferPoolManager.FlushAll -> DiskManager.WriteCheckpoint) writes every dirty and spilled page to a CRC-framed journal (-journal), fsyncs it, copies the pages into the data file, fsyncs that, then empties the journal with truncate+fsync. On open (and before the next checkpoint) a complete journal is re-applied and an incomplete one discarded. - The WAL barrier now runs once per checkpoint instead of per page write. - WritePage and FlushPage are gone; the engine's checkpoint paths (Close, Checkpoint, CreateSnapshot, recovery) share persistCheckpointLocked. - A zero meta page means setup never completed, whatever the file length. - beastctl restore removes a stale journal beside the destination; v0 migration removes its temporary journal. This also removes the fsync per evicted page that made eviction-heavy work (TestLargeValuesReplicate: thousands of F_FULLFSYNCs) collide with the fsync-bound TestEngineSystemWorkloadBenchmark in parallel CI runs. Locally, in a full `go test ./... -race`: benchmark 790-980 -> ~1820-2070 ops/s, TestLargeValuesReplicate 17-38 s -> 7-9 s. Tests: internal/storage/simfile simulates power loss (any subset of unsynced writes and size changes lost). New tests cover checkpoint atomicity across a crash at any point, spill semantics, leftover-journal installation, and engine recovery after random crashes with a tiny pool and after checkpoints that fail mid-journal or mid-install. Both were checked to fail when evicted pages leak into the data file or pages are installed before journaling. --- cmd/beastctl/restore.go | 6 + docs/architecture.md | 5 +- docs/upgrade-spec.md | 2 +- internal/api/engine.go | 10 +- internal/api/engine_checkpoint.go | 43 +++-- internal/api/engine_lostwrite_test.go | 182 ++++++++++++++++++ internal/api/engine_snapshot.go | 15 +- internal/api/migration.go | 3 + internal/api/recovery.go | 10 +- internal/storage/buffer_pool.go | 44 ++--- internal/storage/buffer_pool_eviction.go | 4 +- internal/storage/buffer_pool_wal.go | 14 +- internal/storage/disk_journal.go | 201 ++++++++++++++++++++ internal/storage/disk_manager.go | 162 +++++++++++----- internal/storage/disk_manager_crash_test.go | 163 ++++++++++++++++ internal/storage/disk_manager_test.go | 8 +- internal/storage/disk_spill.go | 31 +++ internal/storage/simfile/simfile.go | 159 ++++++++++++++++ internal/storage/wal_barrier_test.go | 65 ++----- 19 files changed, 949 insertions(+), 178 deletions(-) create mode 100644 internal/api/engine_lostwrite_test.go create mode 100644 internal/storage/disk_journal.go create mode 100644 internal/storage/disk_manager_crash_test.go create mode 100644 internal/storage/disk_spill.go create mode 100644 internal/storage/simfile/simfile.go diff --git a/cmd/beastctl/restore.go b/cmd/beastctl/restore.go index 94f7b03..d4ae64a 100644 --- a/cmd/beastctl/restore.go +++ b/cmd/beastctl/restore.go @@ -9,6 +9,7 @@ import ( "time" "github.com/ChromaBeast/beastdb/internal/api" + "github.com/ChromaBeast/beastdb/internal/storage" ) // RestoreResult captures recovery timing and metadata from a restore drill. @@ -70,6 +71,11 @@ func runRestore(backupDir, destDir string, force bool) (*RestoreResult, error) { } _ = os.Remove(destWAL) + // A journal left by the previous database (e.g. a crash mid-checkpoint) + // belongs to that database; applied to the restored file it would corrupt it. + if err := os.Remove(storage.JournalPath(destDB)); err != nil && !errors.Is(err, os.ErrNotExist) { + return nil, fmt.Errorf("remove stale journal: %w", err) + } // 4. Verify that restored database can be opened cleanly by engine engine, err := api.NewEngine(destDB, destWAL, 20) diff --git a/docs/architecture.md b/docs/architecture.md index 96d531e..4a362b4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -56,7 +56,7 @@ A modular, zero-allocation database architecture built from first principles in 2. **Buffer Pool Pin:** The targeted 4KB disk page is fetched into a memory frame and pinned. 3. **Slotted Page Update:** The tuple is inserted into the slotted page, updating the slot array and free space pointer. 4. **Index Update:** If indexed, the B+ tree is traversed; keys and record pointers are updated, splitting nodes if page capacity is exceeded. -5. **Frame Unpin:** The page frame is marked dirty and unpinned, allowing the background clock-sweep evictor to flush it later. +5. **Frame Unpin:** The page frame is marked dirty and unpinned. If the clock-sweep evictor later needs the frame, the page goes to the spill file; it reaches the data file only at the next checkpoint. ### Read Execution Flow 1. **Cache Check:** Point queries check the sharded in-memory cache first for sub-microsecond responses. @@ -66,6 +66,7 @@ A modular, zero-allocation database architecture built from first principles in --- ## 3. Storage Invariants -- **Write-Ahead Invariant:** Log records must hit disk before dirty database pages containing those modifications are written. Because group commit applies a write to its page before the group's `fsync`, the buffer pool enforces this with a **WAL barrier** (`WAL.EnsureDurable`) called before every dirty-page write (eviction, `FlushPage`, `FlushAll`). If the barrier fails, the page is not written. The barrier costs no extra `fsync` when the WAL is already durable up to the current LSN. +- **Checkpoint Image:** The data file only ever holds a complete checkpoint image. Dirty pages evicted between checkpoints go to a scratch spill file (`-spill`), which is never fsynced and is discarded on restart, so evictions cost no `fsync`. A checkpoint (`FlushAll`) installs every dirty and spilled page atomically: it writes and fsyncs them all to a journal (`-journal`, framed and CRC-checked), copies them into the data file, fsyncs it, and only then empties the journal. On open, a complete journal is re-applied and an incomplete one discarded. Recovery replays the WAL after the checkpoint LSN onto exactly that image; logical replay is not idempotent against pages that already hold later changes, which is why evicted pages must never reach the data file early. +- **Write-Ahead Invariant:** Log records must hit disk before dirty database pages containing those modifications are written. Because group commit applies a write to its page before the group's `fsync`, the buffer pool runs a **WAL barrier** (`WAL.EnsureDurable`) before a checkpoint writes pages to the data file. If the barrier fails, nothing is written. The barrier costs no extra `fsync` when the WAL is already durable up to the current LSN. - **Page Isolation:** Every 4KB page maintains its own self-describing header, slot array, and CRC checksum. - **Pin Safety:** A page frame with `pin_count > 0` cannot be evicted from the buffer pool under any circumstances. diff --git a/docs/upgrade-spec.md b/docs/upgrade-spec.md index 2aace34..e4b2465 100644 --- a/docs/upgrade-spec.md +++ b/docs/upgrade-spec.md @@ -26,7 +26,7 @@ Source: review of 2026-11-12. Complements [ROADMAP.md](../ROADMAP.md) and [v1-se ## Known limits of what shipped - Readers and replication are still blocked for the duration of each group's fsync (the engine keeps one global lock). Lifting this needs MVCC or latch coupling. -- The barrier can add an extra fsync when a dirty page is evicted under memory pressure. +- Evicted dirty pages go to a non-durable spill file until the next checkpoint, so the spill file grows with the number of distinct pages dirtied since then. Auto-checkpoint bounds it along with the WAL. - `LeaderServer.StreamWAL` replays the WAL file from disk and can briefly read a record that is appended but not yet fsynced. - `TestEngineSystemWorkloadBenchmark` has a fixed 1000 ops/s floor and is timing-sensitive on a loaded machine. - Checkpoint LSN/age metrics only appear after the first checkpoint since process start; the buffer pool hit ratio is cumulative. diff --git a/internal/api/engine.go b/internal/api/engine.go index aac8236..59b068c 100644 --- a/internal/api/engine.go +++ b/internal/api/engine.go @@ -38,7 +38,12 @@ func NewEngine(dbPath, walPath string, poolSize int) (*Engine, error) { if err != nil { return nil, err } + return openEngine(disk, dbPath, walPath, poolSize) +} +// openEngine builds the engine on an open data file. dbPath is used only to +// recreate or migrate the file; crash tests pass a simulated file. +func openEngine(disk *storage.DiskManager, dbPath, walPath string, poolSize int) (*Engine, error) { bpm := storage.NewBufferPoolManager(disk, poolSize) w, err := wal.OpenWAL(walPath, true) if err != nil { @@ -63,7 +68,10 @@ func NewEngine(dbPath, walPath string, poolSize int) (*Engine, error) { data := metaPage.Data() - if disk.NumPages() == 1 && isAllZeros(data) { + // A zero meta page means initialization never reached its first sync + // (no write is accepted before that), so there is nothing to recover. The + // file may be longer than one page because growth is reserved in chunks. + if isAllZeros(data) { _ = bpm.UnpinPage(storage.MetaPageID, false) _ = disk.Close() _ = w.Close() diff --git a/internal/api/engine_checkpoint.go b/internal/api/engine_checkpoint.go index 6a98813..ac04326 100644 --- a/internal/api/engine_checkpoint.go +++ b/internal/api/engine_checkpoint.go @@ -22,12 +22,7 @@ func (e *Engine) Close() error { } } - recordErr(e.updateMeta(func(m *storage.MetaData) { - m.LastCheckpointLSN = e.wal.CurrentLSN() - m.ActiveDataPageID = e.activeDataPage - m.RootPageID = e.tree.RootPageID() - })) - recordErr(e.bpm.FlushAll()) + recordErr(e.persistCheckpointLocked(e.wal.CurrentLSN())) recordErr(e.wal.Close()) recordErr(e.disk.Close()) return firstErr @@ -47,27 +42,35 @@ func (e *Engine) Checkpoint() error { return fmt.Errorf("checkpoint sync wal: %w", err) } + // Dirty pages and the meta page must be durable before the WAL that covers + // them is archived. + ckptLSN := e.wal.CurrentLSN() + if err := e.persistCheckpointLocked(ckptLSN); err != nil { + return fmt.Errorf("checkpoint: %w", err) + } + if err := e.wal.Checkpoint(); err != nil { + return err + } + e.lastCkptLSN.Store(ckptLSN) + e.lastCkptUnix.Store(time.Now().Unix()) + return nil +} + +// persistCheckpointLocked records lsn as the recovery start point in the meta +// page and installs every page, meta included, in the data file as one atomic +// checkpoint. Recovery redoes the WAL after lsn on exactly this image. +// Caller holds e.mu exclusively. +func (e *Engine) persistCheckpointLocked(lsn uint64) error { if err := e.updateMeta(func(m *storage.MetaData) { - m.LastCheckpointLSN = e.wal.CurrentLSN() + m.LastCheckpointLSN = lsn m.ActiveDataPageID = e.activeDataPage m.RootPageID = e.tree.RootPageID() }); err != nil { - return fmt.Errorf("checkpoint update meta: %w", err) + return fmt.Errorf("update meta: %w", err) } if err := e.bpm.FlushAll(); err != nil { - return fmt.Errorf("checkpoint flush dirty pages: %w", err) - } - // Dirty pages (and any file extension) must be durable before the WAL that - // covers them is archived. - if err := e.disk.Sync(); err != nil { - return fmt.Errorf("checkpoint sync data file: %w", err) + return fmt.Errorf("write checkpoint: %w", err) } - ckptLSN := e.wal.CurrentLSN() - if err := e.wal.Checkpoint(); err != nil { - return err - } - e.lastCkptLSN.Store(ckptLSN) - e.lastCkptUnix.Store(time.Now().Unix()) return nil } diff --git a/internal/api/engine_lostwrite_test.go b/internal/api/engine_lostwrite_test.go new file mode 100644 index 0000000..e777ab5 --- /dev/null +++ b/internal/api/engine_lostwrite_test.go @@ -0,0 +1,182 @@ +package api + +import ( + "bytes" + "fmt" + "math/rand" + "path/filepath" + "testing" + + "github.com/ChromaBeast/beastdb/internal/storage" + "github.com/ChromaBeast/beastdb/internal/storage/simfile" +) + +// simDisk is a data file and its journal, both simulated so a test can crash them. +type simDisk struct{ data, journal *simfile.File } + +func newSimDisk(dir string) simDisk { + db := filepath.Join(dir, "sim.bin") + return simDisk{data: simfile.New(db), journal: simfile.New(storage.JournalPath(db))} +} + +func (s simDisk) crash(rng *rand.Rand, keep float64) simDisk { + return simDisk{data: s.data.Crash(rng, keep), journal: s.journal.Crash(rng, keep)} +} + +// openEngineOnSim opens an engine on the simulated files with a fresh spill file. +func openEngineOnSim(t *testing.T, s simDisk, walPath string, poolSize int) *Engine { + t.Helper() + disk, err := storage.OpenDiskManagerFiles(s.data, s.journal, simfile.New("spill")) + if err != nil { + t.Fatal(err) + } + e, err := openEngine(disk, s.data.Name(), walPath, poolSize) + if err != nil { + t.Fatalf("open engine: %v", err) + } + return e +} + +// crashEngine abandons e without flushing its buffer pool, as a power cut would. +// Every acknowledged write is already durable in the WAL (writes return only +// after the group fsync), so closing the WAL handle loses nothing extra. +func crashEngine(e *Engine) { _ = e.wal.Close() } + +func lostWriteValue(rng *rand.Rand, tag string) []byte { + n := 8 + rng.Intn(64) + if rng.Intn(10) == 0 { + n = 2000 + rng.Intn(9000) // some values span several pages + } + v := bytes.Repeat([]byte{byte('a' + rng.Intn(26))}, n) + copy(v, tag) + return v +} + +// A crash can lose any write that was not fsynced, in the data file, the +// journal or the spill file, and the file sizes may or may not have persisted. +// With a tiny pool, pages are evicted constantly. Recovery (checkpoint image + +// WAL replay) must still rebuild exactly the committed state. +func TestCrashLosingUnsyncedPageWritesRecovers(t *testing.T) { + seeds := int64(24) + if testing.Short() { + seeds = 6 + } + for seed := int64(1); seed <= seeds; seed++ { + t.Run(fmt.Sprintf("seed%d", seed), func(t *testing.T) { + rng := rand.New(rand.NewSource(seed)) + walPath := filepath.Join(t.TempDir(), "lost.wal") + s := newSimDisk(t.TempDir()) + e := openEngineOnSim(t, s, walPath, 8) // tiny pool: constant eviction + oracle := map[uint64][]byte{} + + for round := 0; round < 4; round++ { + for op := 0; op < 80; op++ { + key := uint64(rng.Intn(100) + 1) + tag := fmt.Sprintf("s%d-r%d-o%d", seed, round, op) + switch rng.Intn(10) { + case 0, 1: + if err := e.Delete(key); err != nil { + t.Fatal(err) + } + delete(oracle, key) + case 2: + ops := []BatchOperation{ + {Type: BatchOpPut, Key: key, Value: lostWriteValue(rng, tag+"a")}, + {Type: BatchOpPut, Key: uint64(rng.Intn(100) + 1), Value: lostWriteValue(rng, tag+"b")}, + {Type: BatchOpDelete, Key: uint64(rng.Intn(100) + 1)}, + } + if err := e.BatchWrite(ops); err != nil { + t.Fatal(err) + } + oracle[ops[0].Key] = ops[0].Value + oracle[ops[1].Key] = ops[1].Value + delete(oracle, ops[2].Key) + case 3: + if rng.Intn(4) == 0 { + if err := e.Checkpoint(); err != nil { + t.Fatal(err) + } + } + default: + v := lostWriteValue(rng, tag) + if err := e.Put(key, v); err != nil { + t.Fatal(err) + } + oracle[key] = v + } + } + + crashEngine(e) + keep := []float64{0, 0.3, 0.7, 1}[rng.Intn(4)] + s = s.crash(rng, keep) + e = openEngineOnSim(t, s, walPath, 8) + verifyAgainstOracle(t, e, oracle) + } + if err := e.Close(); err != nil { + t.Fatal(err) + } + }) + } +} + +// A checkpoint that fails part-way (while writing its journal, or while +// installing pages into the data file) and is followed by a crash must recover +// every committed write. When the engine keeps running instead, the next +// checkpoint must install the leftover journal before its own. +func TestCheckpointCrashMidFlushKeepsCommittedData(t *testing.T) { + for _, target := range []string{"journal", "data"} { + for failAfter := 0; failAfter < 10; failAfter++ { + for _, keepRunning := range []bool{false, true} { + name := fmt.Sprintf("%s-fail%d-continue%v", target, failAfter, keepRunning) + t.Run(name, func(t *testing.T) { + rng := rand.New(rand.NewSource(int64(failAfter))) + walPath := filepath.Join(t.TempDir(), "ckpt.wal") + s := newSimDisk(t.TempDir()) + e := openEngineOnSim(t, s, walPath, 16) + oracle := map[uint64][]byte{} + put := func(k uint64, tag string) { + v := lostWriteValue(rng, tag) + if err := e.Put(k, v); err != nil { + t.Fatal(err) + } + oracle[k] = v + } + for k := uint64(1); k <= 100; k++ { + put(k, fmt.Sprintf("k%d", k)) + } + + f := s.data + if target == "journal" { + f = s.journal + } + f.FailWritesAfter(failAfter) + ckErr := e.Checkpoint() + f.FailWritesAfter(-1) + if ckErr == nil && target == "data" { + t.Fatalf("checkpoint succeeded despite a data write failure after %d writes", failAfter) + } + + if keepRunning { + for k := uint64(1); k <= 100; k += 3 { + put(k, fmt.Sprintf("after-%d", k)) + } + if err := e.Checkpoint(); err != nil { + t.Fatalf("checkpoint after a failed one: %v", err) + } + for k := uint64(2); k <= 100; k += 5 { + put(k, fmt.Sprintf("tail-%d", k)) + } + } + + crashEngine(e) + s = s.crash(rng, []float64{0, 0.5, 1}[failAfter%3]) + e = openEngineOnSim(t, s, walPath, 16) + verifyAgainstOracle(t, e, oracle) + if err := e.Close(); err != nil { + t.Fatal(err) + } + }) + } + } + } +} diff --git a/internal/api/engine_snapshot.go b/internal/api/engine_snapshot.go index 85b9da4..86cc706 100644 --- a/internal/api/engine_snapshot.go +++ b/internal/api/engine_snapshot.go @@ -7,8 +7,6 @@ import ( "os" "path/filepath" "time" - - "github.com/ChromaBeast/beastdb/internal/storage" ) var ( @@ -34,16 +32,9 @@ func (e *Engine) CreateSnapshot(destDir string) (string, error) { e.mu.Lock() defer e.mu.Unlock() - // 1. Force all dirty buffer pool frames to disk and update Meta Page - if err := e.updateMeta(func(m *storage.MetaData) { - m.LastCheckpointLSN = e.wal.CurrentLSN() - m.ActiveDataPageID = e.activeDataPage - m.RootPageID = e.tree.RootPageID() - }); err != nil { - return "", fmt.Errorf("snapshot update meta: %w", err) - } - if err := e.bpm.FlushAll(); err != nil { - return "", fmt.Errorf("snapshot flush dirty pages: %w", err) + // 1. Force all dirty buffer pool frames to disk, then the Meta Page + if err := e.persistCheckpointLocked(e.wal.CurrentLSN()); err != nil { + return "", fmt.Errorf("snapshot: %w", err) } // 2. Perform physical file clone of the flushed .bin file diff --git a/internal/api/migration.go b/internal/api/migration.go index 1d0bc04..a2ad4b4 100644 --- a/internal/api/migration.go +++ b/internal/api/migration.go @@ -127,6 +127,9 @@ func migrateV0ToV1(dbPath, walPath string, poolSize int) error { if err := eng.Close(); err != nil { return fmt.Errorf("failed to close temporary v1 engine: %w", err) } + // The temporary engine's journal is empty after Close; the installed + // database gets its own journal next to dbPath on open. + _ = os.Remove(storage.JournalPath(tmpDB)) bakDB := dbPath + ".v0.bak" bakWAL := walPath + ".v0.bak" diff --git a/internal/api/recovery.go b/internal/api/recovery.go index cd8e249..2b03c81 100644 --- a/internal/api/recovery.go +++ b/internal/api/recovery.go @@ -3,7 +3,6 @@ package api import ( "encoding/binary" - "github.com/ChromaBeast/beastdb/internal/storage" "github.com/ChromaBeast/beastdb/internal/wal" ) @@ -30,14 +29,7 @@ func (e *Engine) recoverFromWAL(lastCheckpointLSN uint64) error { } if replayedCount > 0 { - if err := e.updateMeta(func(m *storage.MetaData) { - m.LastCheckpointLSN = highestReplayedLSN - m.ActiveDataPageID = e.activeDataPage - m.RootPageID = e.tree.RootPageID() - }); err != nil { - return err - } - if err := e.bpm.FlushAll(); err != nil { + if err := e.persistCheckpointLocked(highestReplayedLSN); err != nil { return err } } diff --git a/internal/storage/buffer_pool.go b/internal/storage/buffer_pool.go index 7d1c176..0dc1232 100644 --- a/internal/storage/buffer_pool.go +++ b/internal/storage/buffer_pool.go @@ -26,10 +26,11 @@ type BufferPoolManager struct { hits, misses atomic.Uint64 // FetchPage cache hit/miss counters - // walBarrier, when set, is called before any dirty page is written to the + // walBarrier, when set, is called before a checkpoint writes pages to the // data file. It must return only once every WAL record appended so far is // durable, which preserves the write-ahead invariant even though writes are - // applied to pages before their group's fsync completes. + // applied to pages before their group's fsync completes. Evictions go to + // the non-durable spill file and do not need it. walBarrier func() error } @@ -127,38 +128,31 @@ func (bpm *BufferPoolManager) UnpinPage(pageID uint64, isDirty bool) error { return nil } -// FlushPage forces a cached page to disk if dirty. -func (bpm *BufferPoolManager) FlushPage(pageID uint64) error { +// FlushAll checkpoints the pool: every dirty frame plus every spilled page is +// installed in the data file atomically (see DiskManager.WriteCheckpoint), +// after the WAL barrier. On error no frame is marked clean. +func (bpm *BufferPoolManager) FlushAll() error { bpm.mu.Lock() defer bpm.mu.Unlock() - idx, exists := bpm.pageTable[pageID] - if !exists { - return ErrPageNotFound + var pages []PageImage + for _, frame := range bpm.frames { + if frame.Page != nil && frame.IsDirty { + pages = append(pages, PageImage{ID: frame.PageID, Data: frame.Page.Data()}) + } } - - frame := bpm.frames[idx] - if frame.Page != nil && frame.IsDirty { - if err := bpm.writeBack(frame); err != nil { + if bpm.walBarrier != nil { + if err := bpm.walBarrier(); err != nil { return err } - frame.IsDirty = false } - return nil -} - -// FlushAll flushes all dirty pages across all frames to persistent storage. -func (bpm *BufferPoolManager) FlushAll() error { - bpm.mu.Lock() - defer bpm.mu.Unlock() - + if err := bpm.disk.WriteCheckpoint(pages); err != nil { + return err + } for _, frame := range bpm.frames { - if frame.Page != nil && frame.IsDirty { - if err := bpm.writeBack(frame); err != nil { - return err - } + if frame.Page != nil { frame.IsDirty = false } } - return bpm.disk.Sync() + return nil } diff --git a/internal/storage/buffer_pool_eviction.go b/internal/storage/buffer_pool_eviction.go index 8695808..fb5f4fa 100644 --- a/internal/storage/buffer_pool_eviction.go +++ b/internal/storage/buffer_pool_eviction.go @@ -37,7 +37,7 @@ func (bpm *BufferPoolManager) loadPage(pageID uint64) (*SlottedPage, error) { return page, nil } -// evictFrame flushes (if dirty) and clears the given frame for reuse. +// evictFrame spills the frame's page (if dirty) and clears the frame for reuse. // Caller must hold bpm.mu (write lock). func (bpm *BufferPoolManager) evictFrame(idx int) error { victim := bpm.frames[idx] @@ -45,7 +45,7 @@ func (bpm *BufferPoolManager) evictFrame(idx int) error { return nil } if victim.IsDirty { - if err := bpm.writeBack(victim); err != nil { + if err := bpm.disk.SpillPage(victim.PageID, victim.Page.Data()); err != nil { return err } victim.IsDirty = false diff --git a/internal/storage/buffer_pool_wal.go b/internal/storage/buffer_pool_wal.go index 9ba7820..7e93a0e 100644 --- a/internal/storage/buffer_pool_wal.go +++ b/internal/storage/buffer_pool_wal.go @@ -1,19 +1,9 @@ package storage -// SetWALBarrier installs the write-ahead barrier. Call it before the pool is shared. +// SetWALBarrier installs the write-ahead barrier that FlushAll runs before a +// checkpoint writes pages to the data file. Call it before the pool is shared. func (bpm *BufferPoolManager) SetWALBarrier(fn func() error) { bpm.mu.Lock() defer bpm.mu.Unlock() bpm.walBarrier = fn } - -// writeBack writes one dirty frame to the data file after the WAL barrier. -// Caller must hold bpm.mu (write lock). -func (bpm *BufferPoolManager) writeBack(f *Frame) error { - if bpm.walBarrier != nil { - if err := bpm.walBarrier(); err != nil { - return err - } - } - return bpm.disk.WritePage(f.PageID, f.Page.Data()) -} diff --git a/internal/storage/disk_journal.go b/internal/storage/disk_journal.go new file mode 100644 index 0000000..900c9ee --- /dev/null +++ b/internal/storage/disk_journal.go @@ -0,0 +1,201 @@ +package storage + +import ( + "bufio" + "bytes" + "encoding/binary" + "errors" + "hash/crc32" + "io" +) + +// Journal layout: +// +// header : magic "BDBJRNL1" | page count (uint64 LE) +// entries : page ID (uint64 LE) | 4096-byte page image (count times) +// trailer : CRC-32 (IEEE, uint32 LE) over header and entries | "JEND" +// +// A journal is complete only if its size, magic, trailer and CRC all agree. +// The data file is written only after a complete journal has been fsynced, so +// an incomplete journal means the data file was not touched and can be discarded. +const ( + journalMagic = "BDBJRNL1" + journalEnd = "JEND" + journalHeaderSize = 16 + journalEntrySize = 8 + PageSize + journalTrailerSize = 8 + journalBufSize = 1 << 20 +) + +var errJournalNotInstalled = errors.New("disk: checkpoint journal failed verification after write") + +// PageImage is one page to install at a checkpoint. +type PageImage struct { + ID uint64 + Data []byte +} + +// WriteCheckpoint atomically replaces the data file's image with the current +// state: pages (the buffer pool's dirty frames, which win over any spilled +// copy) plus every page in the spill file. Either all of it reaches the data +// file or, after a crash, none of it does. +func (d *DiskManager) WriteCheckpoint(pages []PageImage) error { + d.mu.Lock() + defer d.mu.Unlock() + + // A journal left complete by an earlier checkpoint that failed part-way must + // be installed first; otherwise its pages and this one's would mix. + if _, err := d.installJournalLocked(); err != nil { + return err + } + + inPool := make(map[uint64]struct{}, len(pages)) + for _, p := range pages { + inPool[p.ID] = struct{}{} + } + var spilled []uint64 + for id := range d.spillSlots { + if _, ok := inPool[id]; !ok { + spilled = append(spilled, id) + } + } + if len(pages)+len(spilled) == 0 { + return nil + } + + if err := d.writeJournalLocked(pages, spilled); err != nil { + return err + } + installed, err := d.installJournalLocked() + if err != nil { + return err + } + if !installed { + return errJournalNotInstalled + } + return d.resetSpillLocked() +} + +// writeJournalLocked writes and fsyncs a complete journal. Caller holds d.mu. +func (d *DiskManager) writeJournalLocked(pages []PageImage, spilled []uint64) error { + bw := bufio.NewWriterSize(io.NewOffsetWriter(d.journal, 0), journalBufSize) + sum := crc32.NewIEEE() + w := io.MultiWriter(bw, sum) + + var hdr [journalHeaderSize]byte + copy(hdr[:8], journalMagic) + binary.LittleEndian.PutUint64(hdr[8:], uint64(len(pages)+len(spilled))) + _, _ = w.Write(hdr[:]) + + var id [8]byte + for _, p := range pages { + if len(p.Data) < PageSize { + return ErrDiskBufferTooSmall + } + binary.LittleEndian.PutUint64(id[:], p.ID) + _, _ = w.Write(id[:]) + _, _ = w.Write(p.Data[:PageSize]) + } + buf := make([]byte, PageSize) + for _, pid := range spilled { + clear(buf) + if _, err := d.spill.ReadAt(buf, d.spillSlots[pid]*PageSize); err != nil && !errors.Is(err, io.EOF) { + return err + } + binary.LittleEndian.PutUint64(id[:], pid) + _, _ = w.Write(id[:]) + _, _ = w.Write(buf) + } + + var trailer [journalTrailerSize]byte + binary.LittleEndian.PutUint32(trailer[:4], sum.Sum32()) + copy(trailer[4:], journalEnd) + _, _ = bw.Write(trailer[:]) + if err := bw.Flush(); err != nil { // reports the first write error, if any + return err + } + return d.journal.Sync() +} + +// installJournalLocked copies a complete journal into the data file, fsyncs +// it, then empties the journal durably. An incomplete journal is discarded. +// It reports whether a journal was installed. Safe to repeat after a crash at +// any point: re-applying full page images is idempotent. Caller holds d.mu +// (or has exclusive access during open). +func (d *DiskManager) installJournalLocked() (bool, error) { + info, err := d.journal.Stat() + if err != nil { + return false, err + } + size := info.Size() + if size == 0 { + return false, nil + } + + count, complete, err := d.verifyJournalLocked(size) + if err != nil { + return false, err + } + if complete { + r := bufio.NewReaderSize(io.NewSectionReader(d.journal, journalHeaderSize, int64(count)*journalEntrySize), journalBufSize) + var id [8]byte + page := make([]byte, PageSize) + for i := uint64(0); i < count; i++ { + if _, err := io.ReadFull(r, id[:]); err != nil { + return false, err + } + if _, err := io.ReadFull(r, page); err != nil { + return false, err + } + if _, err := d.file.WriteAt(page, int64(binary.LittleEndian.Uint64(id[:])*PageSize)); err != nil { + return false, err + } + } + if err := d.file.Sync(); err != nil { + return false, err + } + } + // Empty the journal durably (truncate, not delete, so this also holds where + // a directory cannot be fsynced): a stale journal re-applied after a later + // checkpoint would roll pages back. + if err := d.journal.Truncate(0); err != nil { + return false, err + } + if err := d.journal.Sync(); err != nil { + return false, err + } + return complete, nil +} + +// verifyJournalLocked checks the journal's framing and checksum and returns +// its page count. complete is false for a journal that was never fully written. +func (d *DiskManager) verifyJournalLocked(size int64) (count uint64, complete bool, err error) { + if size < journalHeaderSize+journalTrailerSize { + return 0, false, nil + } + var hdr [journalHeaderSize]byte + if _, err := d.journal.ReadAt(hdr[:], 0); err != nil { + return 0, false, err + } + if !bytes.Equal(hdr[:8], []byte(journalMagic)) { + return 0, false, nil + } + count = binary.LittleEndian.Uint64(hdr[8:]) + body := size - journalHeaderSize - journalTrailerSize + if count > uint64(body)/journalEntrySize || int64(count)*journalEntrySize != body { + return 0, false, nil + } + + sum := crc32.NewIEEE() + if _, err := io.Copy(sum, io.NewSectionReader(d.journal, 0, size-journalTrailerSize)); err != nil { + return 0, false, err + } + var trailer [journalTrailerSize]byte + if _, err := d.journal.ReadAt(trailer[:], size-journalTrailerSize); err != nil && !errors.Is(err, io.EOF) { + return 0, false, err + } + if string(trailer[4:]) != journalEnd || binary.LittleEndian.Uint32(trailer[:4]) != sum.Sum32() { + return 0, false, nil + } + return count, true, nil +} diff --git a/internal/storage/disk_manager.go b/internal/storage/disk_manager.go index 229563f..da09267 100644 --- a/internal/storage/disk_manager.go +++ b/internal/storage/disk_manager.go @@ -3,7 +3,9 @@ package storage import ( "errors" "io" + "io/fs" "os" + "path/filepath" "sync" ) @@ -12,53 +14,102 @@ var ( ErrDiskBufferTooSmall = errors.New("disk: buffer smaller than 4096 bytes") ) -// DiskManager manages physical read/write operations against the database file. +// File is the subset of *os.File the disk manager uses. Crash tests substitute +// an in-memory file that can drop writes that were never synced. +type File interface { + io.ReaderAt + io.WriterAt + Truncate(size int64) error + Sync() error + Close() error + Stat() (fs.FileInfo, error) + Name() string +} + +// DiskManager owns the data file and its two companions. +// +// Durability model: the data file only ever holds a complete checkpoint image. +// Pages evicted between checkpoints go to the spill file, which is scratch +// space (never fsynced, discarded on restart). WriteCheckpoint replaces the +// image atomically: every page is first written and fsynced to the journal, +// then copied into the data file, and only then is the journal emptied. On +// open, a complete journal is re-applied and an incomplete one discarded. +// +// Recovery therefore always starts from exactly the state the last checkpoint +// captured and redoes the WAL from there, which logical replay requires: it +// is not idempotent against pages that already contain later changes. type DiskManager struct { - file *os.File - numPages uint64 mu sync.Mutex + file File // data file: a checkpoint image + journal File // empty except while a checkpoint is being installed + spill File // evicted pages since the last checkpoint; not durable + numPages uint64 + + spillSlots map[uint64]int64 // pageID -> slot in the spill file + spillNext int64 // next free spill slot + + removeOnClose string // spill file path, removed on Close ("" for injected files) } -// OpenDiskManager opens or creates a database file for 4KB aligned page I/O. +// JournalPath and SpillPath name the companion files of a data file. +func JournalPath(dbPath string) string { return dbPath + "-journal" } +func SpillPath(dbPath string) string { return dbPath + "-spill" } + +// OpenDiskManager opens or creates a database file and its journal and spill files. func OpenDiskManager(path string) (*DiskManager, error) { - file, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0600) + data, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0600) if err != nil { return nil, err } - - info, err := file.Stat() + journal, err := os.OpenFile(JournalPath(path), os.O_RDWR|os.O_CREATE, 0600) if err != nil { - file.Close() + data.Close() return nil, err } + spill, err := os.OpenFile(SpillPath(path), os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600) + if err != nil { + data.Close() + journal.Close() + return nil, err + } + syncDir(path) // make the new files' directory entries durable - numPages := uint64(info.Size() / PageSize) - - return &DiskManager{ - file: file, - numPages: numPages, - }, nil + d, err := OpenDiskManagerFiles(data, journal, spill) + if err != nil { + data.Close() + journal.Close() + spill.Close() + return nil, err + } + d.removeOnClose = SpillPath(path) + return d, nil } -// ReadPage reads a 4KB block from disk at offset pageID * PageSize. -func (d *DiskManager) ReadPage(pageID uint64, pageData []byte) error { - if len(pageData) < PageSize { - return ErrDiskBufferTooSmall +// OpenDiskManagerFiles builds a disk manager on already-open files, first +// finishing or discarding a checkpoint that a crash interrupted. The spill +// file is expected to be empty. +func OpenDiskManagerFiles(data, journal, spill File) (*DiskManager, error) { + d := &DiskManager{ + file: data, + journal: journal, + spill: spill, + spillSlots: make(map[uint64]int64), } - - d.mu.Lock() - defer d.mu.Unlock() - - offset := int64(pageID * PageSize) - _, err := d.file.ReadAt(pageData[:PageSize], offset) - if err != nil && !errors.Is(err, io.EOF) { - return err + if _, err := d.installJournalLocked(); err != nil { + return nil, err } - return nil + info, err := data.Stat() + if err != nil { + return nil, err + } + d.numPages = uint64(info.Size() / PageSize) + return d, nil } -// WritePage writes a 4KB block to disk at offset pageID * PageSize and flushes. -func (d *DiskManager) WritePage(pageID uint64, pageData []byte) error { +// ReadPage reads the current version of a page: from the spill file if it was +// evicted since the last checkpoint, otherwise from the data file. A page past +// the end of the data file reads as zeros. +func (d *DiskManager) ReadPage(pageID uint64, pageData []byte) error { if len(pageData) < PageSize { return ErrDiskBufferTooSmall } @@ -66,54 +117,58 @@ func (d *DiskManager) WritePage(pageID uint64, pageData []byte) error { d.mu.Lock() defer d.mu.Unlock() - offset := int64(pageID * PageSize) - if _, err := d.file.WriteAt(pageData[:PageSize], offset); err != nil { + src, offset := d.file, int64(pageID*PageSize) + if slot, ok := d.spillSlots[pageID]; ok { + src, offset = d.spill, slot*PageSize + } + _, err := src.ReadAt(pageData[:PageSize], offset) + if err != nil && !errors.Is(err, io.EOF) { return err } - - return d.file.Sync() + return nil } -// AllocatePage atomically reserves the next 4KB slot in the database file. +// AllocatePage hands out the next page ID. The page reaches the data file at +// the next checkpoint (every allocated page is dirty until then). func (d *DiskManager) AllocatePage() (uint64, error) { d.mu.Lock() defer d.mu.Unlock() - pageID := d.numPages - newSize := int64((pageID + 1) * PageSize) - - if err := d.file.Truncate(newSize); err != nil { - return 0, err - } - d.numPages++ return pageID, nil } -// NumPages returns the total page count managed on disk. +// NumPages returns the number of allocated pages. func (d *DiskManager) NumPages() uint64 { d.mu.Lock() defer d.mu.Unlock() return d.numPages } -// Sync flushes all pending file writes to durable physical storage. +// Sync flushes the data file. Checkpoints already sync; this is for callers +// that copy the file. func (d *DiskManager) Sync() error { d.mu.Lock() defer d.mu.Unlock() return d.file.Sync() } -// Close flushes and releases the database file handle. +// Close releases the files and removes the spill file. Pages not written by +// a checkpoint are dropped, as after a crash; callers checkpoint first. func (d *DiskManager) Close() error { d.mu.Lock() defer d.mu.Unlock() - if err := d.file.Sync(); err != nil { - d.file.Close() - return err + err := d.file.Sync() + for _, f := range []File{d.file, d.journal, d.spill} { + if cerr := f.Close(); err == nil { + err = cerr + } } - return d.file.Close() + if d.removeOnClose != "" { + _ = os.Remove(d.removeOnClose) + } + return err } // Path returns the physical filesystem path of the database file. @@ -122,3 +177,14 @@ func (d *DiskManager) Path() string { defer d.mu.Unlock() return d.file.Name() } + +// syncDir fsyncs the directory holding path so newly created files survive a +// crash. Best effort: not every platform can fsync a directory. +func syncDir(path string) { + dir, err := os.Open(filepath.Dir(path)) + if err != nil { + return + } + _ = dir.Sync() + _ = dir.Close() +} diff --git a/internal/storage/disk_manager_crash_test.go b/internal/storage/disk_manager_crash_test.go new file mode 100644 index 0000000..41544b9 --- /dev/null +++ b/internal/storage/disk_manager_crash_test.go @@ -0,0 +1,163 @@ +package storage + +import ( + "bytes" + "fmt" + "math/rand" + "testing" + + "github.com/ChromaBeast/beastdb/internal/storage/simfile" +) + +type simDisk struct{ data, journal *simfile.File } + +func newSimDisk() simDisk { + return simDisk{data: simfile.New("data"), journal: simfile.New("journal")} +} + +func (s simDisk) open(t *testing.T) *DiskManager { + t.Helper() + d, err := OpenDiskManagerFiles(s.data, s.journal, simfile.New("spill")) + if err != nil { + t.Fatal(err) + } + return d +} + +func (s simDisk) crash(rng *rand.Rand, keep float64) simDisk { + return simDisk{data: s.data.Crash(rng, keep), journal: s.journal.Crash(rng, keep)} +} + +func filledPage(b byte) []byte { return bytes.Repeat([]byte{b}, PageSize) } + +// readAll returns the first byte of every page, which identifies its version. +func readAll(t *testing.T, d *DiskManager, n uint64) []byte { + t.Helper() + out := make([]byte, n) + buf := make([]byte, PageSize) + for id := uint64(0); id < n; id++ { + clear(buf) + if err := d.ReadPage(id, buf); err != nil { + t.Fatal(err) + } + if !bytes.Equal(buf, filledPage(buf[0])) { + t.Fatalf("page %d is torn: mixes versions", id) + } + out[id] = buf[0] + } + return out +} + +// A crash at any point of a checkpoint (journal write, data install, journal +// reset) must leave the data file holding either the whole old image or the +// whole new one after reopen, never a mix. +func TestCheckpointIsAtomicAcrossCrash(t *testing.T) { + const pages = 12 + for _, target := range []string{"journal", "data"} { + for failAfter := 0; failAfter <= pages+1; failAfter++ { + for seed := int64(0); seed < 6; seed++ { + name := fmt.Sprintf("%s-fail%d-seed%d", target, failAfter, seed) + t.Run(name, func(t *testing.T) { + rng := rand.New(rand.NewSource(seed)) + s := newSimDisk() + d := s.open(t) + var old []PageImage + for i := 0; i < pages; i++ { + id, _ := d.AllocatePage() + old = append(old, PageImage{ID: id, Data: filledPage('A')}) + } + if err := d.WriteCheckpoint(old); err != nil { + t.Fatal(err) + } + + // New image: half the pages via spill, half passed directly. + var next []PageImage + for i := 0; i < pages; i++ { + if i%2 == 0 { + if err := d.SpillPage(uint64(i), filledPage('B')); err != nil { + t.Fatal(err) + } + } else { + next = append(next, PageImage{ID: uint64(i), Data: filledPage('B')}) + } + } + f := s.data + if target == "journal" { + f = s.journal + } + f.FailWritesAfter(failAfter) + err := d.WriteCheckpoint(next) + f.FailWritesAfter(-1) + + keep := []float64{0, 0.5, 1}[seed%3] + r := s.crash(rng, keep).open(t) + got := readAll(t, r, pages) + allA, allB := bytes.Repeat([]byte{'A'}, pages), bytes.Repeat([]byte{'B'}, pages) + switch { + case bytes.Equal(got, allB): + case bytes.Equal(got, allA): + if err == nil && keep == 1 { + t.Fatalf("checkpoint reported success but the old image came back") + } + default: + t.Fatalf("mixed image after crash: %q (checkpoint err %v)", got, err) + } + }) + } + } + } +} + +// Evicted pages go to the spill file: no fsync, readable until the next +// checkpoint, and never visible in the data file after a crash. +func TestSpilledPagesAreScratchOnly(t *testing.T) { + s := newSimDisk() + d := s.open(t) + for i := 0; i < 4; i++ { + _, _ = d.AllocatePage() + } + if err := d.WriteCheckpoint([]PageImage{{0, filledPage('A')}, {1, filledPage('A')}, {2, filledPage('A')}, {3, filledPage('A')}}); err != nil { + t.Fatal(err) + } + syncs := s.data.Syncs() + s.journal.Syncs() + for round := 0; round < 50; round++ { + if err := d.SpillPage(uint64(round%4), filledPage('S')); err != nil { + t.Fatal(err) + } + } + if got := s.data.Syncs() + s.journal.Syncs(); got != syncs { + t.Fatalf("spilling pages cost %d fsyncs, want 0", got-syncs) + } + if got := readAll(t, d, 4); !bytes.Equal(got, []byte("SSSS")) { + t.Fatalf("spilled pages read back as %q", got) + } + r := s.crash(rand.New(rand.NewSource(1)), 1).open(t) + if got := readAll(t, r, 4); !bytes.Equal(got, []byte("AAAA")) { + t.Fatalf("after crash the data file holds %q, want the checkpoint image", got) + } +} + +// A checkpoint that failed after its journal was complete leaves that journal +// behind. The next checkpoint must install it before writing its own, so the +// data file never mixes the two. +func TestFailedCheckpointJournalIsInstalledFirst(t *testing.T) { + s := newSimDisk() + d := s.open(t) + for i := 0; i < 3; i++ { + _, _ = d.AllocatePage() + } + if err := d.WriteCheckpoint([]PageImage{{0, filledPage('A')}, {1, filledPage('A')}, {2, filledPage('A')}}); err != nil { + t.Fatal(err) + } + s.data.FailWritesAfter(1) // journal completes, install stops after one page + if err := d.WriteCheckpoint([]PageImage{{0, filledPage('B')}, {1, filledPage('B')}, {2, filledPage('B')}}); err == nil { + t.Fatal("expected the install to fail") + } + s.data.FailWritesAfter(-1) + if err := d.WriteCheckpoint([]PageImage{{2, filledPage('C')}}); err != nil { + t.Fatal(err) + } + if got := readAll(t, d, 3); !bytes.Equal(got, []byte("BBC")) { + t.Fatalf("data file holds %q, want BBC", got) + } +} diff --git a/internal/storage/disk_manager_test.go b/internal/storage/disk_manager_test.go index 97bfe00..67737bb 100644 --- a/internal/storage/disk_manager_test.go +++ b/internal/storage/disk_manager_test.go @@ -36,8 +36,8 @@ func TestDiskManager(t *testing.T) { writeBuf := make([]byte, PageSize) copy(writeBuf, "hello disk manager page 0") - if err := disk.WritePage(0, writeBuf); err != nil { - t.Fatalf("failed to write page 0: %v", err) + if err := disk.SpillPage(0, writeBuf); err != nil { + t.Fatalf("failed to spill page 0: %v", err) } readBuf := make([]byte, PageSize) @@ -49,6 +49,10 @@ func TestDiskManager(t *testing.T) { t.Fatalf("read content mismatch: %s", string(readBuf[:25])) } + // Only a checkpoint reaches the data file; spilled pages do not survive Close. + if err := disk.WriteCheckpoint([]PageImage{{ID: 0, Data: writeBuf}, {ID: 1, Data: make([]byte, PageSize)}}); err != nil { + t.Fatalf("failed to checkpoint: %v", err) + } if err := disk.Close(); err != nil { t.Fatalf("failed to close disk manager: %v", err) } diff --git a/internal/storage/disk_spill.go b/internal/storage/disk_spill.go new file mode 100644 index 0000000..2a4df31 --- /dev/null +++ b/internal/storage/disk_spill.go @@ -0,0 +1,31 @@ +package storage + +// SpillPage stores an evicted dirty page in the spill file. It is not durable +// and never touches the data file, so it needs neither an fsync nor the WAL +// barrier: after a crash the spill file is discarded and the WAL redoes the +// change from the last checkpoint image. +func (d *DiskManager) SpillPage(pageID uint64, pageData []byte) error { + if len(pageData) < PageSize { + return ErrDiskBufferTooSmall + } + + d.mu.Lock() + defer d.mu.Unlock() + + slot, ok := d.spillSlots[pageID] + if !ok { + slot = d.spillNext + d.spillNext++ + d.spillSlots[pageID] = slot + } + _, err := d.spill.WriteAt(pageData[:PageSize], slot*PageSize) + return err +} + +// resetSpillLocked empties the spill area once a checkpoint has made every +// spilled page part of the data file. Caller holds d.mu. +func (d *DiskManager) resetSpillLocked() error { + clear(d.spillSlots) + d.spillNext = 0 + return d.spill.Truncate(0) +} diff --git a/internal/storage/simfile/simfile.go b/internal/storage/simfile/simfile.go new file mode 100644 index 0000000..a9e8898 --- /dev/null +++ b/internal/storage/simfile/simfile.go @@ -0,0 +1,159 @@ +// Package simfile is an in-memory file for crash tests. It tracks what Sync has +// made durable, so a test can simulate power loss: Crash returns the last +// synced image plus an arbitrary subset of the writes made since, with the file +// size either persisted or not. That is the window a real OS leaves open when +// writes are not fsynced, and the one the engine's ordering rules must survive. +package simfile + +import ( + "errors" + "io" + "io/fs" + "math/rand" + "sync" + "time" +) + +// ErrInjected is returned by WriteAt once the budget set by FailWritesAfter runs out. +var ErrInjected = errors.New("simfile: injected write failure") + +type op struct { + off int64 + data []byte // nil for a truncate + truncate bool + size int64 +} + +// File implements the subset of *os.File the storage layer uses. +type File struct { + mu sync.Mutex + name string + data []byte // what reads see now + durable []byte // image as of the last Sync + pending []op // writes and truncates since the last Sync + syncs int + failAfter int // WriteAt calls left before ErrInjected; <0 disables +} + +// New returns an empty file. +func New(name string) *File { return &File{name: name, failAfter: -1} } + +func (f *File) Name() string { return f.name } + +func (f *File) ReadAt(p []byte, off int64) (int, error) { + f.mu.Lock() + defer f.mu.Unlock() + if off >= int64(len(f.data)) { + return 0, io.EOF + } + n := copy(p, f.data[off:]) + if n < len(p) { + return n, io.EOF + } + return n, nil +} + +func (f *File) WriteAt(p []byte, off int64) (int, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.failAfter == 0 { + return 0, ErrInjected + } + if f.failAfter > 0 { + f.failAfter-- + } + if end := off + int64(len(p)); end > int64(len(f.data)) { + f.data = resize(f.data, end) + } + copy(f.data[off:], p) + f.pending = append(f.pending, op{off: off, data: append([]byte(nil), p...)}) + return len(p), nil +} + +func (f *File) Truncate(size int64) error { + f.mu.Lock() + defer f.mu.Unlock() + f.data = resize(f.data, size) + f.pending = append(f.pending, op{truncate: true, size: size}) + return nil +} + +func (f *File) Sync() error { + f.mu.Lock() + defer f.mu.Unlock() + f.durable = append(f.durable[:0:0], f.data...) + f.pending = nil + f.syncs++ + return nil +} + +func (f *File) Close() error { return nil } + +func (f *File) Stat() (fs.FileInfo, error) { + f.mu.Lock() + defer f.mu.Unlock() + return info{name: f.name, size: int64(len(f.data))}, nil +} + +// Syncs reports how many times Sync has been called. +func (f *File) Syncs() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.syncs +} + +// FailWritesAfter makes WriteAt fail with ErrInjected after n more successful calls. +// A negative n turns injection off. +func (f *File) FailWritesAfter(n int) { + f.mu.Lock() + defer f.mu.Unlock() + f.failAfter = n +} + +// Crash simulates power loss and returns the file as a restart would find it. +// Starting from the last synced image, each unsynced write or truncate survives +// with probability keep (0 keeps none, 1 keeps all), applied in order. The file +// size then lands on either the synced size or the size reached by the kept +// operations, chosen at random, because the OS may persist data blocks and +// size metadata independently. +func (f *File) Crash(rng *rand.Rand, keep float64) *File { + f.mu.Lock() + defer f.mu.Unlock() + img := append([]byte(nil), f.durable...) + for _, o := range f.pending { + if rng.Float64() >= keep { + continue + } + if o.truncate { + img = resize(img, o.size) + continue + } + if end := o.off + int64(len(o.data)); end > int64(len(img)) { + img = resize(img, end) + } + copy(img[o.off:], o.data) + } + if rng.Intn(2) == 0 { + img = resize(img, int64(len(f.durable))) + } + return &File{name: f.name, data: img, durable: append([]byte(nil), img...), failAfter: -1} +} + +func resize(b []byte, size int64) []byte { + if size <= int64(len(b)) { + return b[:size] + } + return append(b, make([]byte, size-int64(len(b)))...) +} + +type info struct { + name string + size int64 +} + +func (i info) Name() string { return i.name } +func (i info) Size() int64 { return i.size } +func (i info) Mode() fs.FileMode { return 0o600 } +func (i info) ModTime() time.Time { return time.Time{} } +func (i info) IsDir() bool { return false } +func (i info) Sys() any { return nil } diff --git a/internal/storage/wal_barrier_test.go b/internal/storage/wal_barrier_test.go index 1e03faa..588a91b 100644 --- a/internal/storage/wal_barrier_test.go +++ b/internal/storage/wal_barrier_test.go @@ -19,7 +19,7 @@ func newBarrierPool(t *testing.T, size int) (*BufferPoolManager, *DiskManager) { // A dirty page must not reach the data file when the WAL barrier fails: // otherwise a page could be persisted ahead of the log records that describe it. -func TestWALBarrierFailureBlocksPageWrite(t *testing.T) { +func TestWALBarrierFailureBlocksCheckpoint(t *testing.T) { bpm, disk := newBarrierPool(t, 1) errBarrier := errors.New("wal not durable") bpm.SetWALBarrier(func() error { return errBarrier }) @@ -32,29 +32,19 @@ func TestWALBarrierFailureBlocksPageWrite(t *testing.T) { if err := bpm.UnpinPage(id, true); err != nil { t.Fatal(err) } - - // Eviction (pool of one) must fail because the barrier fails. - if _, _, err := bpm.NewPage(); !errors.Is(err, errBarrier) { - t.Fatalf("eviction err = %v, want barrier error", err) - } - if err := bpm.FlushPage(id); !errors.Is(err, errBarrier) { - t.Fatalf("FlushPage err = %v, want barrier error", err) - } if err := bpm.FlushAll(); !errors.Is(err, errBarrier) { t.Fatalf("FlushAll err = %v, want barrier error", err) } onDisk := make([]byte, PageSize) - if err := disk.ReadPage(id, onDisk); err != nil { - t.Fatal(err) - } - if bytes.Contains(onDisk, []byte("uncommitted")) { + if _, err := disk.file.ReadAt(onDisk, int64(id*PageSize)); err == nil && bytes.Contains(onDisk, []byte("uncommitted")) { t.Fatal("dirty page reached the data file although the WAL barrier failed") } } -// The barrier runs before every page write path, and pages are written after it. -func TestWALBarrierRunsBeforeEveryPageWrite(t *testing.T) { +// Evictions spill to scratch space and need no barrier; a checkpoint runs the +// barrier once, then installs every dirty and spilled page in the data file. +func TestWALBarrierRunsOncePerCheckpoint(t *testing.T) { bpm, disk := newBarrierPool(t, 1) calls := 0 bpm.SetWALBarrier(func() error { calls++; return nil }) @@ -68,45 +58,32 @@ func TestWALBarrierRunsBeforeEveryPageWrite(t *testing.T) { t.Fatal(err) } - // Evict via a second page (pool size 1). - _, id2, err := bpm.NewPage() + // Evict via a second page (pool size 1): goes to the spill file. + page2, id2, err := bpm.NewPage() if err != nil { t.Fatal(err) } - if calls != 1 { - t.Fatalf("barrier calls after eviction = %d, want 1", calls) + copy(page2.Data()[100:], []byte("second")) + if calls != 0 { + t.Fatalf("barrier calls after eviction = %d, want 0", calls) } - onDisk := make([]byte, PageSize) - if err := disk.ReadPage(id, onDisk); err != nil { - t.Fatal(err) - } - if !bytes.Contains(onDisk, []byte("committed")) { - t.Fatal("evicted page was not written after the barrier succeeded") - } - if err := bpm.UnpinPage(id2, true); err != nil { t.Fatal(err) } - if err := bpm.FlushPage(id2); err != nil { + if err := bpm.FlushAll(); err != nil { t.Fatal(err) } - if calls != 2 { - t.Fatalf("barrier calls after FlushPage = %d, want 2", calls) + if calls != 1 { + t.Fatalf("barrier calls after FlushAll = %d, want 1", calls) } - // Dirty again, then FlushAll. - p, err := bpm.FetchPage(id2) - if err != nil { - t.Fatal(err) - } - copy(p.Data()[200:], []byte("more")) - if err := bpm.UnpinPage(id2, true); err != nil { - t.Fatal(err) - } - if err := bpm.FlushAll(); err != nil { - t.Fatal(err) - } - if calls != 3 { - t.Fatalf("barrier calls after FlushAll = %d, want 3", calls) + for pid, want := range map[uint64]string{id: "committed", id2: "second"} { + onDisk := make([]byte, PageSize) + if _, err := disk.file.ReadAt(onDisk, int64(pid*PageSize)); err != nil { + t.Fatal(err) + } + if !bytes.Contains(onDisk, []byte(want)) { + t.Fatalf("page %d not in the data file after checkpoint", pid) + } } } From 8ce51d8b277fe8a07aaf0a6f7d6a8fdf406c800e Mon Sep 17 00:00:00 2001 From: ChromaBeast Date: Fri, 13 Nov 2026 11:50:55 +0530 Subject: [PATCH 2/2] fix(replication): ack idle progress; wait instead of sleeping in tests Windows CI on this branch failed TestLeaderFollowerReplication: "follower missing replayed key 20" after a fixed 100 ms sleep. The benchmark passed there (2065 ops/s), but the replication tests ran about twice as slow as on main, because the new engine crash tests fsynced a real WAL file on every commit and ran in parallel with them. - Engine crash tests open their WAL without per-commit fsync (openEngine gains a syncWAL flag; production always passes true). They only simulate loss in the data and journal files and close the WAL normally, so the fsyncs added nothing but disk load. Under -race: ~34 s -> ~12 s. - Replication tests wait for the follower with a 5 s deadline instead of fixed 100/150 ms sleeps (waitUntil in wait_test.go). - That exposed a real follower bug the sleeps were hiding: the interval ACK was only checked when the next record arrived, so after fewer than ackBatchSize records followed by an idle stream the leader never learned of the applied progress. The interval ACK now runs on its own ticker, and ACKs are serialized and skipped when nothing new was applied, so the leader receives them in LSN order. --- internal/api/engine.go | 10 +++--- internal/api/engine_lostwrite_test.go | 6 +++- internal/replication/batch_rep_test.go | 15 +++++---- internal/replication/follower.go | 44 ++++++++++++++++++++++---- internal/replication/replica_test.go | 35 ++++++++++---------- internal/replication/wait_test.go | 27 ++++++++++++++++ 6 files changed, 103 insertions(+), 34 deletions(-) create mode 100644 internal/replication/wait_test.go diff --git a/internal/api/engine.go b/internal/api/engine.go index 59b068c..88585f7 100644 --- a/internal/api/engine.go +++ b/internal/api/engine.go @@ -38,14 +38,16 @@ func NewEngine(dbPath, walPath string, poolSize int) (*Engine, error) { if err != nil { return nil, err } - return openEngine(disk, dbPath, walPath, poolSize) + return openEngine(disk, dbPath, walPath, poolSize, true) } // openEngine builds the engine on an open data file. dbPath is used only to -// recreate or migrate the file; crash tests pass a simulated file. -func openEngine(disk *storage.DiskManager, dbPath, walPath string, poolSize int) (*Engine, error) { +// recreate or migrate the file; crash tests pass a simulated file. syncWAL is +// always true in production; crash tests that never lose WAL writes turn it +// off so they do not fsync on every commit. +func openEngine(disk *storage.DiskManager, dbPath, walPath string, poolSize int, syncWAL bool) (*Engine, error) { bpm := storage.NewBufferPoolManager(disk, poolSize) - w, err := wal.OpenWAL(walPath, true) + w, err := wal.OpenWAL(walPath, syncWAL) if err != nil { _ = disk.Close() return nil, err diff --git a/internal/api/engine_lostwrite_test.go b/internal/api/engine_lostwrite_test.go index e777ab5..eb8d8a1 100644 --- a/internal/api/engine_lostwrite_test.go +++ b/internal/api/engine_lostwrite_test.go @@ -30,7 +30,11 @@ func openEngineOnSim(t *testing.T, s simDisk, walPath string, poolSize int) *Eng if err != nil { t.Fatal(err) } - e, err := openEngine(disk, s.data.Name(), walPath, poolSize) + // The WAL is a real file that these tests never crash (crashEngine closes it + // normally), so per-commit fsyncs add nothing but disk load: without them + // the tests stay CPU-only and do not slow fsync-bound tests running in + // parallel packages. + e, err := openEngine(disk, s.data.Name(), walPath, poolSize, false) if err != nil { t.Fatalf("open engine: %v", err) } diff --git a/internal/replication/batch_rep_test.go b/internal/replication/batch_rep_test.go index b4eeefb..ce85436 100644 --- a/internal/replication/batch_rep_test.go +++ b/internal/replication/batch_rep_test.go @@ -6,7 +6,6 @@ import ( "net" "path/filepath" "testing" - "time" "github.com/ChromaBeast/beastdb/internal/api" "google.golang.org/grpc" @@ -69,12 +68,16 @@ func TestAtomicBatchReplicationCompleteness(t *testing.T) { t.Fatalf("batch write failed: %v", err) } - // Wait for follower catch-up - time.Sleep(150 * time.Millisecond) - - // Verify all 4 operations are present on the follower + // Verify all 4 operations reach the follower for _, op := range batchOps { - val, found, err := follower.Get(op.Key) + var val []byte + var found bool + var err error + // On timeout the checks below report what is still wrong. + waitUntil(t, func() bool { + val, found, err = follower.Get(op.Key) + return err == nil && found && bytes.Equal(val, op.Value) + }) if err != nil || !found { t.Fatalf("follower missing batch record %d: found=%v, err=%v", op.Key, found, err) } diff --git a/internal/replication/follower.go b/internal/replication/follower.go index 39c85e9..e588f29 100644 --- a/internal/replication/follower.go +++ b/internal/replication/follower.go @@ -19,7 +19,7 @@ var ( ) const ( - ackBatchSize = 100 // send ACK every N records + ackBatchSize = 100 // send ACK every N records ackBatchInterval = 100 * time.Millisecond // or every 100 ms, whichever first ) @@ -30,6 +30,9 @@ type FollowerClient struct { client beastv1.ReplicationServiceClient lastAppliedLSN uint64 mu sync.RWMutex + + ackMu sync.Mutex // serializes ACK RPCs so the leader sees them in LSN order + lastAckedLSN uint64 // guarded by ackMu } // NewFollowerClient initializes a replication worker targeting the primary leader. @@ -45,7 +48,13 @@ func NewFollowerClient(replicaID string, engine *api.Engine, conn *grpc.ClientCo // SyncLoop continuously streams and applies WAL records from the Leader until context cancels. // Fix #4: ACKs are batched — sent every ackBatchSize records OR every ackBatchInterval ms, // whichever comes first. This reduces ACK round-trips from O(writes/sec) to O(writes/100). +// The interval ACK runs on its own ticker, so progress is acknowledged even when +// the stream goes idle after fewer than ackBatchSize records. func (f *FollowerClient) SyncLoop(ctx context.Context) error { + tickCtx, stopTicker := context.WithCancel(ctx) + defer stopTicker() + go f.ackTicker(tickCtx) + f.mu.Lock() fromLSN := f.lastAppliedLSN + 1 f.mu.Unlock() @@ -59,7 +68,6 @@ func (f *FollowerClient) SyncLoop(ctx context.Context) error { } var count int - lastAck := time.Now() for { msg, err := stream.Recv() @@ -92,23 +100,45 @@ func (f *FollowerClient) SyncLoop(ctx context.Context) error { f.mu.Unlock() count++ - if count >= ackBatchSize || time.Since(lastAck) >= ackBatchInterval { + if count >= ackBatchSize { f.sendAck(ctx) count = 0 - lastAck = time.Now() } } } -// sendAck fires a single ACK for the current lastAppliedLSN. +// ackTicker acknowledges applied-but-unacknowledged records every +// ackBatchInterval until ctx is cancelled. +func (f *FollowerClient) ackTicker(ctx context.Context) { + t := time.NewTicker(ackBatchInterval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + f.sendAck(ctx) + } + } +} + +// sendAck fires a single ACK for the current lastAppliedLSN, unless that LSN +// is already acknowledged. func (f *FollowerClient) sendAck(ctx context.Context) { + f.ackMu.Lock() + defer f.ackMu.Unlock() f.mu.RLock() lsn := f.lastAppliedLSN f.mu.RUnlock() - _, _ = f.client.Ack(ctx, &beastv1.ReplicationAck{ + if lsn == f.lastAckedLSN { + return + } + if _, err := f.client.Ack(ctx, &beastv1.ReplicationAck{ ReplicaId: f.replicaID, AcknowledgedLsn: lsn, - }) + }); err == nil { + f.lastAckedLSN = lsn + } } // LastAppliedLSN returns the latest Log Sequence Number applied by this replica. diff --git a/internal/replication/replica_test.go b/internal/replication/replica_test.go index aab9902..6c47f47 100644 --- a/internal/replication/replica_test.go +++ b/internal/replication/replica_test.go @@ -6,7 +6,6 @@ import ( "net" "path/filepath" "testing" - "time" "github.com/ChromaBeast/beastdb/internal/api" "google.golang.org/grpc" @@ -72,11 +71,14 @@ func TestLeaderFollowerReplication(t *testing.T) { }() // Wait for catch-up replication of initial keys - time.Sleep(100 * time.Millisecond) - for _, k := range initialKeys { - val, found, err := followerEngine.Get(k) - if err != nil || !found || !bytes.Equal(val, []byte("val-initial")) { + var val []byte + var found bool + var err error + if !waitUntil(t, func() bool { + val, found, err = followerEngine.Get(k) + return err == nil && found && bytes.Equal(val, []byte("val-initial")) + }) { t.Fatalf("follower missing replayed key %d: found=%v, err=%v", k, found, err) } } @@ -88,10 +90,12 @@ func TestLeaderFollowerReplication(t *testing.T) { t.Fatalf("leader live put failed: %v", err) } - time.Sleep(100 * time.Millisecond) - - val, found, err := followerEngine.Get(liveKey) - if err != nil || !found || !bytes.Equal(val, liveVal) { + var val []byte + var found bool + if !waitUntil(t, func() bool { + val, found, err = followerEngine.Get(liveKey) + return err == nil && found && bytes.Equal(val, liveVal) + }) { t.Fatalf("follower failed to receive live replicated key 999: found=%v, err=%v", found, err) } @@ -119,16 +123,15 @@ func TestLeaderFollowerReplication(t *testing.T) { t.Fatalf("leader delete failed: %v", err) } - time.Sleep(100 * time.Millisecond) - - _, found20, _ := followerEngine.Get(20) - if found20 { + if !waitUntil(t, func() bool { + _, found20, err := followerEngine.Get(20) + return err == nil && !found20 + }) { t.Fatalf("expected deleted key 20 to be gone from replica") } - // Verify Leader tracked the follower's Ack - ackLSN := leaderRepServer.GetReplicaAck("replica-node-1") - if ackLSN == 0 { + // Verify Leader tracked the follower's Ack (acks are batched, so wait for one) + if !waitUntil(t, func() bool { return leaderRepServer.GetReplicaAck("replica-node-1") != 0 }) { t.Fatalf("expected non-zero replica ACK LSN on leader") } } diff --git a/internal/replication/wait_test.go b/internal/replication/wait_test.go new file mode 100644 index 0000000..ac88718 --- /dev/null +++ b/internal/replication/wait_test.go @@ -0,0 +1,27 @@ +package replication + +import ( + "testing" + "time" +) + +// replicationDeadline bounds how long a test waits for the follower to apply +// what the leader committed. Each applied write is an fsync on the follower, +// which on a loaded CI runner can take far longer than on a laptop. +const replicationDeadline = 5 * time.Second + +// waitUntil polls cond until it holds or the deadline passes, and reports +// whether it held. Use it instead of a fixed sleep before asserting on a follower. +func waitUntil(t *testing.T, cond func() bool) bool { + t.Helper() + deadline := time.Now().Add(replicationDeadline) + for { + if cond() { + return true + } + if time.Now().After(deadline) { + return false + } + time.Sleep(10 * time.Millisecond) + } +}