diff --git a/mpt/disk.go b/mpt/disk.go index efb022f..3c2c9ce 100644 --- a/mpt/disk.go +++ b/mpt/disk.go @@ -93,11 +93,12 @@ type diskTree struct { pmem *pmem.Mem mem []byte // cache of pmem.Data() - file1 File - file2 File - leaf File - closed bool - err error // sticky error + file1 File + file2 File + leaf File + persistedVersion int64 + closed bool + err error // sticky error } // broken marks the tree broken with err as the reason. @@ -211,6 +212,7 @@ func memOpen(file1, file2, disk File, op string) (_ Tree, err error) { } t.mem = t.pmem.Data() + t.persistedVersion = t.hdr().version() return t, nil } @@ -233,6 +235,7 @@ func (t *diskTree) Sync() error { if err := t.pmem.Sync(); err != nil { return t.broken(err) } + t.persistedVersion = t.hdr().version() return nil } diff --git a/mpt/disk_test.go b/mpt/disk_test.go index f101db6..dc7f2ca 100644 --- a/mpt/disk_test.go +++ b/mpt/disk_test.go @@ -171,6 +171,9 @@ func (tt *tester) reopen(minVer int64, minExact bool, format string, args ...any if version < minVer || minExact != exact { tt.t.Fatalf("reopen: %s: version = %d,%v, want ≥ %d,%v", kind, version, exact, minVer, minExact) } + if tree.PersistedVersion() != version { + tt.t.Fatalf("reopen: %s: PersistedVersion = %d, want %d", kind, tree.PersistedVersion(), version) + } if !exact { f1.readOnly = false f2.readOnly = false @@ -437,3 +440,158 @@ func TestSetOverwriteDiskSize(t *testing.T) { } }) } + +func TestPersistedVersion(t *testing.T) { + type step struct { + action string // "check", "set", "snap", "sync", "reopen" + key string + val string + version int64 + wantVersion int64 + wantPersisted int64 // expected PersistedVersion on diskTree (on memTree, equals wantVersion) + } + + tests := []struct { + name string + diskOnly bool + steps []step + }{ + { + name: "initial_empty", + steps: []step{ + {action: "check", wantVersion: 0, wantPersisted: 0}, + }, + }, + { + name: "snap_without_sync", + steps: []step{ + {action: "set", key: "k1", val: "v1", wantVersion: 0, wantPersisted: 0}, + {action: "snap", version: 10, wantVersion: 10, wantPersisted: 0}, + }, + }, + { + name: "snap_then_sync", + steps: []step{ + {action: "set", key: "k1", val: "v1", wantVersion: 0, wantPersisted: 0}, + {action: "snap", version: 10, wantVersion: 10, wantPersisted: 0}, + {action: "sync", wantVersion: 10, wantPersisted: 10}, + }, + }, + { + name: "unflushed_crash_recovery", + diskOnly: true, + steps: []step{ + {action: "set", key: "k1", val: "v1", wantVersion: 0, wantPersisted: 0}, + {action: "snap", version: 10, wantVersion: 10, wantPersisted: 0}, + {action: "sync", wantVersion: 10, wantPersisted: 10}, + {action: "set", key: "k2", val: "v2", wantVersion: 10, wantPersisted: 10}, + {action: "snap", version: 20, wantVersion: 20, wantPersisted: 10}, + {action: "reopen", wantVersion: 10, wantPersisted: 10}, + }, + }, + { + name: "negative_version_snap", + steps: []step{ + {action: "set", key: "k1", val: "v1", wantVersion: 0, wantPersisted: 0}, + {action: "snap", version: 10, wantVersion: 10, wantPersisted: 0}, + {action: "sync", wantVersion: 10, wantPersisted: 10}, + {action: "set", key: "k2", val: "v2", wantVersion: 10, wantPersisted: 10}, + {action: "snap", version: -1, wantVersion: 10, wantPersisted: 10}, + {action: "sync", wantVersion: 10, wantPersisted: 10}, + }, + }, + { + name: "multiple_sync_cycles", + steps: []step{ + {action: "set", key: "k1", val: "v1", wantVersion: 0, wantPersisted: 0}, + {action: "snap", version: 10, wantVersion: 10, wantPersisted: 0}, + {action: "sync", wantVersion: 10, wantPersisted: 10}, + {action: "set", key: "k2", val: "v2", wantVersion: 10, wantPersisted: 10}, + {action: "snap", version: 20, wantVersion: 20, wantPersisted: 10}, + {action: "sync", wantVersion: 20, wantPersisted: 20}, + {action: "set", key: "k3", val: "v3", wantVersion: 20, wantPersisted: 20}, + {action: "snap", version: 30, wantVersion: 30, wantPersisted: 20}, + {action: "sync", wantVersion: 30, wantPersisted: 30}, + }, + }, + } + + t.Run("diskTree", func(t *testing.T) { + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + f1 := new(memFile) + f2 := new(memFile) + f3 := new(memFile) + + tree, err := New(f1, f2, f3) + check(t, err) + defer tree.Close() + + for i, s := range tc.steps { + switch s.action { + case "set": + check(t, tree.Set(Key(s.key), Val(s.val))) + case "snap": + _, err := tree.Snap(s.version) + check(t, err) + case "sync": + check(t, tree.Sync()) + case "reopen": + f1Copy := &memFile{data: slices.Clone(f1.data)} + f2Copy := &memFile{data: slices.Clone(f2.data)} + f3Copy := &memFile{data: slices.Clone(f3.data)} + check(t, tree.Close()) + tree, err = New(f1Copy, f2Copy, f3Copy) + check(t, err) + case "check": + // Just assert versions. + default: + t.Fatalf("unknown action %q", s.action) + } + + if v, _ := tree.Version(); v != s.wantVersion { + t.Fatalf("step %d (%s): Version() = %d, want %d", i, s.action, v, s.wantVersion) + } + if got := tree.PersistedVersion(); got != s.wantPersisted { + t.Fatalf("step %d (%s): PersistedVersion() = %d, want %d", i, s.action, got, s.wantPersisted) + } + } + }) + } + }) + + t.Run("memTree", func(t *testing.T) { + for _, tc := range tests { + if tc.diskOnly { + continue + } + t.Run(tc.name, func(t *testing.T) { + tree := NewMemTree() + defer tree.Close() + + for i, s := range tc.steps { + switch s.action { + case "set": + check(t, tree.Set(Key(s.key), Val(s.val))) + case "snap": + _, err := tree.Snap(s.version) + check(t, err) + case "sync": + check(t, tree.Sync()) + case "check": + // Just assert versions. + default: + t.Fatalf("unknown action %q", s.action) + } + + if v, _ := tree.Version(); v != s.wantVersion { + t.Fatalf("step %d (%s): Version() = %d, want %d", i, s.action, v, s.wantVersion) + } + if got := tree.PersistedVersion(); got != s.wantVersion { + t.Fatalf("step %d (%s): PersistedVersion() = %d, want %d", i, s.action, got, s.wantVersion) + } + } + }) + } + }) +} diff --git a/mpt/dmem.go b/mpt/dmem.go index a6841fe..a82f667 100644 --- a/mpt/dmem.go +++ b/mpt/dmem.go @@ -126,6 +126,15 @@ func (t *diskTree) Version() (version int64, exact bool) { return hdr.version(), hdr.exact() } +// PersistedVersion returns the latest version number that has been +// completely written and synced to disk. +func (t *diskTree) PersistedVersion() int64 { + t.mmu.RLock() + defer t.mmu.RUnlock() + + return t.persistedVersion +} + // Set sets the value associated with key to val. func (t *diskTree) Set(key Key, val Val) error { t.mmu.RLock() diff --git a/mpt/mem.go b/mpt/mem.go index d4cbbc4..396795b 100644 --- a/mpt/mem.go +++ b/mpt/mem.go @@ -96,6 +96,12 @@ func (t *memTree) Version() (version int64, exact bool) { return t.version, t.exact } +// PersistedVersion returns the latest version number of the tree. +// For an in-memory tree, this is the same as Version. +func (t *memTree) PersistedVersion() int64 { + return t.version +} + // Snap returns a snapshot of t. func (t *memTree) Snap(version int64) (Snapshot, error) { if t.err != nil { diff --git a/mpt/tree.go b/mpt/tree.go index 1cccd07..eed2ff8 100644 --- a/mpt/tree.go +++ b/mpt/tree.go @@ -236,6 +236,10 @@ type Tree interface { // expected to replay all Set calls up to the next version. Version() (version int64, exact bool) + // PersistedVersion returns the latest version number that has been + // completely written and synced to disk. + PersistedVersion() int64 + // Close calls Sync and then closes the underlying files. Close() error }