fix(storage): keep the data file a checkpoint image; journal checkpoints

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)

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 (`<db>-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 (`<db>-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.

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.

internal/api/engine.go

@@ -38,9 +38,16 @@ func NewEngine(dbPath, walPath string, poolSize int) (*Engine, error) {
 	if err != nil {
 		return nil, err
 	}
+	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. 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
@@ -63,7 +70,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()

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
 }
 

internal/api/engine_lostwrite_test.go

@@ -0,0 +1,186 @@
+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)
+	}
+	// 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)
+	}
+	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)
+					}
+				})
+			}
+		}
+	}
+}

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

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"

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
 		}
 	}