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..88585f7 100644 --- a/internal/api/engine.go +++ b/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() 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..eb8d8a1 --- /dev/null +++ b/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) + } + }) + } + } + } +} 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/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) + } +} 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) + } } }