diff --git a/sei-db/db_engine/view/batch_update_test.go b/sei-db/db_engine/view/batch_update_test.go new file mode 100644 index 0000000000..87745c128c --- /dev/null +++ b/sei-db/db_engine/view/batch_update_test.go @@ -0,0 +1,207 @@ +package view + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// markUpdater folds a value by appending mark to whatever the key already held, so a result says +// which value it was folded onto rather than merely that a fold happened. A nil prior folds from +// "", which distinguishes a key the store held nothing for from one holding an empty value. +type markUpdater struct { + mark byte +} + +func (u markUpdater) NewValueFor(_ string, priorValue []byte) ([]byte, error) { + if priorValue == nil { + return append([]byte(""), u.mark), nil + } + return append(append([]byte{}, priorValue...), u.mark), nil +} + +// parkedUpdater holds every fold until it is released, which is what lets a test tell staging apart +// from folding. +type parkedUpdater struct { + started chan struct{} + release chan struct{} + once sync.Once +} + +func newParkedUpdater() *parkedUpdater { + return &parkedUpdater{started: make(chan struct{}), release: make(chan struct{})} +} + +func (u *parkedUpdater) NewValueFor(_ string, priorValue []byte) ([]byte, error) { + u.once.Do(func() { close(u.started) }) + <-u.release + return append(append([]byte{}, priorValue...), '+'), nil +} + +// failingUpdater fails every fold, standing in for a corrupted stored value. +type failingUpdater struct{} + +func (failingUpdater) NewValueFor(_ string, _ []byte) ([]byte, error) { + return nil, fmt.Errorf("fold refused this value") +} + +// BatchUpdate must return without folding anything. The fold here parks until released, so a +// BatchUpdate that performed it on the calling thread could never return and this test would hang +// rather than pass — reaching the assertions at all is the proof. +func TestBatchUpdateReturnsBeforeFolding(t *testing.T) { + manager, _ := newTestManager(t, map[string][]byte{"k": []byte("old")}, 4, 1<<20) + updater := newParkedUpdater() + + require.NoError(t, manager.BatchUpdate([]string{"k"}, updater)) + + // A read of a staged key must wait for its value rather than miss the write or serve the value it + // replaced. Observing "did not return" needs a deadline; the fold is parked, so any wait fails + // the same way. + type readResult struct { + value []byte + found bool + err error + } + reads := make(chan readResult, 1) + go func() { + value, found, err := manager.Get([]byte("k"), true) + reads <- readResult{value: value, found: found, err: err} + }() + + <-updater.started + select { + case got := <-reads: + t.Fatalf("a read of a staged key returned %q before its value was available", got.value) + case <-time.After(50 * time.Millisecond): + } + + close(updater.release) + got := <-reads + require.NoError(t, got.err) + require.True(t, got.found) + require.Equal(t, []byte("old+"), got.value, "the read must serve the folded value") +} + +// A staged key has to read as its new value on every read surface, not just the single-key one. +func TestBatchUpdateStagedValueIsVisibleToEveryReadPath(t *testing.T) { + seed := map[string][]byte{"a": []byte("1"), "b": []byte("2")} + manager, _ := newTestManager(t, seed, 4, 1<<20) + + require.NoError(t, manager.BatchUpdate([]string{"a", "b"}, markUpdater{mark: '+'})) + + value, found, err := manager.Get([]byte("a"), true) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, []byte("1+"), value) + + batch, err := manager.BatchGet([][]byte{[]byte("a"), []byte("b")}) + require.NoError(t, err) + require.Equal(t, []byte("1+"), batch["a"]) + require.Equal(t, []byte("2+"), batch["b"]) + + it, err := manager.Iterator(nil) + require.NoError(t, err) + defer func() { require.NoError(t, it.Close()) }() + seen := map[string][]byte{} + for ; it.Valid(); it.Next() { + seen[string(it.Key())] = append([]byte{}, it.Value()...) + } + require.NoError(t, it.Error()) + require.Equal(t, []byte("1+"), seen["a"], "an iterator must not copy an unfolded value") + require.Equal(t, []byte("2+"), seen["b"]) +} + +// Two folds staged for one key within a single version must apply in the order they were staged: the +// second folds onto the first's result, not onto the value both of them started from. +func TestBatchUpdateChainsFoldsWithinAVersion(t *testing.T) { + manager, _ := newTestManager(t, map[string][]byte{"k": []byte("a")}, 4, 1<<20) + + require.NoError(t, manager.BatchUpdate([]string{"k"}, markUpdater{mark: '1'})) + require.NoError(t, manager.BatchUpdate([]string{"k"}, markUpdater{mark: '2'})) + + value, found, err := manager.Get([]byte("k"), true) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, []byte("a12"), value) +} + +// The same key folded in consecutive versions must chain the same way, with each version's fold +// applying to the previous version's result. +func TestBatchUpdateChainsFoldsAcrossVersions(t *testing.T) { + manager, _ := newTestManager(t, map[string][]byte{"k": []byte("a")}, 4, 1<<20) + + var held []View + defer func() { + for _, v := range held { + require.NoError(t, v.Release()) + } + }() + + for _, mark := range []byte{'1', '2', '3'} { + require.NoError(t, manager.BatchUpdate([]string{"k"}, markUpdater{mark: mark})) + sealed, err := manager.Commit() + require.NoError(t, err) + require.NoError(t, sealed.Finalize(hashWrites(testHash))) + held = append(held, sealed) + } + + value, found, err := manager.Get([]byte("k"), true) + require.NoError(t, err) + require.True(t, found) + require.Equal(t, []byte("a123"), value) + + // Each sealed version's diff must carry that version's own folded value, since the diff is what + // hashing and flushing read. + for i, want := range [][]byte{[]byte("a1"), []byte("a12"), []byte("a123")} { + diff, err := held[i].GetDiff() + require.NoError(t, err) + require.Equal(t, want, diff["k"], "version %d's diff", i+1) + } +} + +// A key whose fold deletes it must read as absent and reach the diff as a nil value, which is how a +// delete is carried to the flush. +func TestBatchUpdateFoldCanDelete(t *testing.T) { + manager, _ := newTestManager(t, map[string][]byte{"k": []byte("doomed")}, 4, 1<<20) + + require.NoError(t, manager.BatchUpdate([]string{"k"}, deletingUpdater{})) + + _, found, err := manager.Get([]byte("k"), true) + require.NoError(t, err) + require.False(t, found, "a key whose fold returned nil must read as absent") + + sealed, err := manager.Commit() + require.NoError(t, err) + require.NoError(t, sealed.Finalize(hashWrites(testHash))) + defer func() { require.NoError(t, sealed.Release()) }() + + diff, err := sealed.GetDiff() + require.NoError(t, err) + value, present := diff["k"] + require.True(t, present, "a delete must appear in the diff") + require.Nil(t, value, "a delete is carried as a nil value") +} + +type deletingUpdater struct{} + +func (deletingUpdater) NewValueFor(_ string, _ []byte) ([]byte, error) { return nil, nil } + +// A fold that fails has to be reported to everything that would otherwise read its value as good: a +// reader, and the diff that hashing and flushing consume. It must also brick the manager, because a +// version missing one of its writes can never be hashed correctly. +func TestBatchUpdateFoldFailureReachesObservers(t *testing.T) { + manager, _ := newTestManager(t, map[string][]byte{"k": []byte("old")}, 4, 1<<20) + + require.NoError(t, manager.BatchUpdate([]string{"k"}, failingUpdater{})) + + _, _, err := manager.Get([]byte("k"), true) + require.Error(t, err, "a reader must not be handed the value a failed fold never produced") + require.Contains(t, err.Error(), "fold refused this value") + + // The diff consumers have to fail too rather than hash a version that is missing this key. + _, err = manager.Commit() + require.Error(t, err, "a failed fold must brick the manager") +} diff --git a/sei-db/db_engine/view/differential_test.go b/sei-db/db_engine/view/differential_test.go index 7098578f14..046e13fd9a 100644 --- a/sei-db/db_engine/view/differential_test.go +++ b/sei-db/db_engine/view/differential_test.go @@ -43,6 +43,7 @@ const ( opSet = iota opDelete opBatch + opUpdate opView ) @@ -91,6 +92,13 @@ func runDifferential(t *testing.T, shardCount, maxSize uint64, seedDB bool, seed muts := randMuts(rng, keys) require.NoError(t, manager.BatchSet(muts)) model.BatchSet(muts) + case opUpdate: + updated := randUpdateKeys(rng, keys) + require.NoError(t, manager.BatchUpdate(updated, foldUpdater{})) + for _, k := range updated { + prior, _ := model.GetLive([]byte(k)) + model.Set([]byte(k), foldedValue(prior)) + } case opView: if len(opens) >= maxOpen { releaseOldest() @@ -193,13 +201,57 @@ func pickOp(rng *testutil.TestRandom) int { return opSet case r < 60: return opDelete - case r < 80: + case r < 70: return opBatch + case r < 80: + return opUpdate default: return opView } } +// randUpdateKeys picks the keys for one BatchUpdate, deduplicated because the contract forbids a +// repeated key. +func randUpdateKeys(rng *testutil.TestRandom, keys [][]byte) []string { + n := rng.IntRange(1, 9) + seen := make(map[string]struct{}, n) + picked := make([]string, 0, n) + for i := 0; i < n; i++ { + k := string(pick(rng, keys)) + if _, ok := seen[k]; ok { + continue + } + seen[k] = struct{}{} + picked = append(picked, k) + } + return picked +} + +// foldUpdater folds through foldedValue, so the oracle can reproduce the same writes from its own +// state rather than carrying a second copy of the manager's logic. +type foldUpdater struct{} + +var _ BatchUpdater = foldUpdater{} + +func (foldUpdater) NewValueFor(_ string, priorValue []byte) ([]byte, error) { + return foldedValue(priorValue), nil +} + +// foldedValue is a pure function of the value a key already held. A key holding nothing gets one, a +// value that has grown past the cap is deleted, and anything else is extended — between them the +// create, modify and delete outcomes a fold can have. Bounded so a long run cannot grow values +// without limit. +func foldedValue(priorValue []byte) []byte { + if priorValue == nil { + return []byte("folded") + } + if len(priorValue) >= 12 { + return nil + } + // Copied rather than appended in place: priorValue aliases the manager's own stored value. + return append(append([]byte{}, priorValue...), '+') +} + func genKeys(rng *testutil.TestRandom, n int) [][]byte { keys := make([][]byte, n) keys[0] = []byte{} // include the empty key as an edge case diff --git a/sei-db/db_engine/view/pending_value.go b/sei-db/db_engine/view/pending_value.go new file mode 100644 index 0000000000..66224003ac --- /dev/null +++ b/sei-db/db_engine/view/pending_value.go @@ -0,0 +1,90 @@ +package view + +import ( + "context" + "fmt" +) + +// pendingValue is a staged value whose bytes are not yet known: the fold of one batch's changes onto +// the value its key already held. +type pendingValue struct { + // The key whose value this fold produces. + key string + + // Closed once value and err are final. + done chan struct{} + + // What the fold produced, nil for a delete. Illegal to read before the done chan is closed. + value []byte + + // The failure that stopped the fold, if it failed. Illegal to read before the done chan is closed. + err error +} + +// newPendingValue returns an unresolved staged value for the given key. +func newPendingValue(key string) *pendingValue { + return &pendingValue{key: key, done: make(chan struct{})} +} + +// inject records what the fold produced and releases every observer. Called exactly once. +func (p *pendingValue) inject(value []byte, err error) { + p.value = value + p.err = err + // Closing publishes both fields and wakes every observer at once, so none of them has to pass the + // value to the next, and a second inject panics here rather than queueing a second answer. + close(p.done) +} + +// await blocks until this value is resolved and reports what it resolved to. A nil value means the +// key was deleted. Must be called with no shard lock held, since the fold needs that lock to publish. +// +// ctx is cancelled when the manager shuts down, and shutdownError then names the cause. +func (p *pendingValue) await(ctx context.Context, shutdownError func() error) ([]byte, error) { + // Not threading.InterruptiblePull: it reports a closed channel as an error, and a close is how a + // resolved value is published here. + select { + case <-p.done: + if p.err != nil { + return nil, fmt.Errorf("staged value failed to resolve: %w", p.err) + } + return p.value, nil + case <-ctx.Done(): + return nil, fmt.Errorf("view manager shut down while awaiting a staged value: %w", shutdownError()) + } +} + +// stagedFold is one fold's two halves: where it gets the value it folds onto, and where it puts the +// result. The key is on result. +type stagedFold struct { + // Where the prior value comes from. + prior priorValueSource + + // The handle every observer of this key waits on, and where the fold injects its result. + result *pendingValue +} + +// Which of the three places a staged fold's prior value comes from. +type priorValueLocation int + +const ( + // The shard's versioned data holds it; priorValueSource.value is it. + priorValueInVersionedData priorValueLocation = 1 + // An earlier staged fold on the same key has yet to produce it; priorValueSource.pending is + // that fold. + priorValueInEarlierFold priorValueLocation = 2 + // Nothing the shard holds has it, so it comes from the read cache. + priorValueInReadCache priorValueLocation = 3 +) + +// priorValueSource is where a staged fold gets the prior value it folds onto. +type priorValueSource struct { + // Which of the three places the prior value comes from. + location priorValueLocation + + // The prior value. Meaningful exactly while location is priorValueInVersionedData, where a nil + // value is a tombstone rather than an absence. + value []byte + + // The earlier fold to await. Non-nil exactly while location is priorValueInEarlierFold. + pending *pendingValue +} diff --git a/sei-db/db_engine/view/read_cache.go b/sei-db/db_engine/view/read_cache.go index 24c6862c3f..58268ef868 100644 --- a/sei-db/db_engine/view/read_cache.go +++ b/sei-db/db_engine/view/read_cache.go @@ -20,7 +20,8 @@ import ( // Capitalized methods are the surface the shard calls; readCache is unexported, so they are not exports. // // Method postfixes state the lock contract: RLocked and WLocked require the caller to hold the read or -// write lock, Unlocked requires the caller to hold neither, and a bare name has no lock dependency. +// write lock, and Unlocked requires the caller to hold neither. A bare name touches no guarded state, +// or is external surface whose caller has no access to the lock. type readCache struct { // Cancelled when the manager shuts down; interrupts blocked waits on in-flight reads. ctx context.Context diff --git a/sei-db/db_engine/view/shard.go b/sei-db/db_engine/view/shard.go index eb9e09d5a0..f4bb6e59d2 100644 --- a/sei-db/db_engine/view/shard.go +++ b/sei-db/db_engine/view/shard.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "github.com/sei-protocol/sei-chain/sei-db/common/structures" @@ -24,8 +25,12 @@ import ( // - The database crashed. Database failures are fatal and are never recovered from, so every shard // goes out of service, not just the one that saw the failure. // +// Capitalized methods are the surface the ViewManager calls; shard is unexported, so they are not +// exports. +// // Method postfixes state the lock contract: RLocked and WLocked require the caller to hold the read or -// write lock, Unlocked requires the caller to hold neither, and a bare name has no lock dependency. +// write lock, and Unlocked requires the caller to hold neither. A bare name touches no guarded state, +// or is external surface whose caller has no access to the lock. type shard struct { // A lock to protect the shard's data. Also used by the read cache (see the cache field). lock sync.RWMutex @@ -59,6 +64,22 @@ type shard struct { // leaked iterator, since reading one after the database has closed is undefined behaviour (see // ViewManager.Close). openIterators uint64 + + // versionLatches holds a latch for each version that still has staged folds outstanding; a + // version absent from the map has none. Guarded by the shard lock. See versionLatch. + versionLatches map[uint64]*versionLatch + + // ctx is cancelled when the manager shuts down. Awaits on a staged value observe it, because a + // fold interrupted by shutdown never resolves. + ctx context.Context + + // shutdownError names the cause once ctx is cancelled. + shutdownError func() error + + // reportFoldFailure bricks the manager. A fold that cannot complete leaves a version unhashable, + // so it has to stop the whole manager rather than only this shard — the same response the read + // cache gives a failed database read. + reportFoldFailure func(error) } // A single value at a specific version. @@ -69,6 +90,22 @@ type versionedValue struct { // as block height, this is just a version number that monotonically increases over the lifetime // of a view manager instance. version uint64 + // pending is non-nil while this value's bytes are still being folded. Every path that reads value + // must check it first. + pending *pendingValue +} + +// versionLatch counts a version's unresolved staged values and lets an observer wait for the last of +// them. A latch that has opened stays open, and one left incomplete by a failure carries that failure. +type versionLatch struct { + // How many of this version's staged values have not resolved. Guarded by the shard lock. + count int + + // Closed when count reaches zero. + done chan struct{} + + // The fold failure that left this version incomplete, if one did. + err error } // Creates a new Shard. @@ -87,6 +124,9 @@ func NewShard( shutdownError func() error, // Reports a failed DB read to the manager, which bricks and stops serving reads. reportReadFailure func(error), + // Reports a fold that could not produce its value to the manager, which bricks. Distinct from + // reportReadFailure so the latched error names the failure that actually happened. + reportFoldFailure func(error), ) (*shard, error) { if maxSize == 0 { @@ -100,6 +140,11 @@ func NewShard( // failure mode this reporting exists to prevent. return nil, fmt.Errorf("reportReadFailure must be non-nil") } + if reportFoldFailure == nil { + // A fold failure leaves a version's diff incomplete. A shard that cannot report one would let + // that version be hashed as though it were whole. + return nil, fmt.Errorf("reportFoldFailure must be non-nil") + } versionDiffs := make(map[uint64]map[string][]byte) versionDiffs[1] = make(map[string][]byte) // versions start at 1 @@ -109,6 +154,11 @@ func NewShard( versionDiffs: versionDiffs, currentVersion: 1, // important: versions start at 1, not 0, to allow (version - 1) without underflow oldestVersion: 1, + versionLatches: make(map[uint64]*versionLatch), + ctx: ctx, + shutdownError: shutdownError, + + reportFoldFailure: reportFoldFailure, } s.cache = NewReadCache(ctx, config, db, readPool, &s.lock, maxSize, shutdownError, reportReadFailure) return s, nil @@ -125,8 +175,19 @@ func (s *shard) Get( // overhead to do so with little benefit. updateLru bool, ) ([]byte, bool, error) { - if value, found, done, err := s.attemptFastGetUnlocked(key, version, updateLru); done { - return value, found, err + value, found, pending, done, err := s.attemptFastGetUnlocked(key, version, updateLru) + if pending != nil { + value, err := pending.await(s.ctx, s.shutdownError) + if err != nil { + return nil, false, fmt.Errorf("get key %x at version %d: %w", key, version, err) + } + return value, value != nil, nil + } + if done { + if err != nil { + return nil, false, fmt.Errorf("get at version %d: %w", version, err) + } + return value, found, nil } // Not resolvable without mutating: classify against the DB read-cache under the write lock, @@ -138,19 +199,26 @@ func (s *shard) Get( // not just those that would have reached the DB. if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { s.lock.Unlock() - return nil, false, err + return nil, false, fmt.Errorf("get key %x: %w", key, err) } if err := s.validateVersionRLocked(version); err != nil { s.lock.Unlock() - return nil, false, err + return nil, false, fmt.Errorf("get key %x: %w", key, err) } // First, check to see if we have this value in the versioned data map. - if value, found := s.lookupVersionedRLocked(key, version); found { + if entry, found := s.lookupVersionedRLocked(key, version); found { s.lock.Unlock() + if entry.pending != nil { + value, err := entry.pending.await(s.ctx, s.shutdownError) + if err != nil { + return nil, false, fmt.Errorf("get key %x at version %d: %w", key, version, err) + } + return value, value != nil, nil + } s.metrics.reportCacheHits(1) - return value, value != nil, nil + return entry.value, entry.value != nil, nil } outcome := s.cache.LookupWLocked(key, updateLru) @@ -162,32 +230,38 @@ func (s *shard) Get( // attemptFastGetUnlocked attempts a read while holding only the read lock, reporting done when it // succeeded. A non-nil err always comes with done. A read it could not resolve without mutating is // left to the caller to redo under the write lock. +// +// A key holding an unresolved fold is reported as pending, for the caller to await once it has +// released the lock. func (s *shard) attemptFastGetUnlocked( key []byte, version uint64, updateLru bool, -) (value []byte, found bool, done bool, err error) { +) (value []byte, found bool, pending *pendingValue, done bool, err error) { s.lock.RLock() defer s.lock.RUnlock() if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { - return nil, false, true, err + return nil, false, nil, true, fmt.Errorf("key %x: %w", key, err) } if err := s.validateVersionRLocked(version); err != nil { - return nil, false, true, err + return nil, false, nil, true, fmt.Errorf("key %x: %w", key, err) } - if value, found := s.lookupVersionedRLocked(key, version); found { + if entry, found := s.lookupVersionedRLocked(key, version); found { + if entry.pending != nil { + return nil, false, entry.pending, false, nil + } s.metrics.reportCacheHits(1) - return value, value != nil, true, nil + return entry.value, entry.value != nil, nil, true, nil } value, found, ok := s.cache.AttemptFastLookupRLocked(key, updateLru) if !ok { - return nil, false, false, nil + return nil, false, nil, false, nil } s.metrics.reportCacheHits(1) - return value, found, true, nil + return value, found, nil, true, nil } // validateVersionRLocked checks that the given version is within the valid range. @@ -201,28 +275,32 @@ func (s *shard) validateVersionRLocked(version uint64) error { return nil } -// lookupVersionedRLocked checks versioned data for a key at the given version. -// Returns (value, true) if found in versioned data, (nil, false) if the read cache should be -// consulted. -func (s *shard) lookupVersionedRLocked(key []byte, version uint64) ([]byte, bool) { +// lookupVersionedRLocked checks versioned data for a key at the given version. Reports the entry and +// true when versioned data holds one, or false when the read cache should be consulted instead. +// +// The entry may be an unresolved fold, so every caller has to check its pending field before reading +// its value. Resolving one requires releasing this lock first; see pendingValue.await. +func (s *shard) lookupVersionedRLocked(key []byte, version uint64) (versionedValue, bool) { + // Converted inline rather than by the caller: the compiler elides the conversion only where it + // indexes a map directly, and every single-key read pays an allocation for it otherwise. deque, ok := s.versionedData[string(key)] if !ok { - return nil, false + return versionedValue{}, false } if version == s.oldestVersion { next := deque.PeekFront() if next.version == version { - return next.value, true + return next, true } - return nil, false + return versionedValue{}, false } for i := deque.Len() - 1; i >= 0; i-- { next := deque.Get(i) if next.version <= version { - return next.value, true + return next, true } } - return nil, false + return versionedValue{}, false } // BatchGet reads the given keys at the given version, returning a map (keyed by string(key)) of the @@ -231,19 +309,23 @@ func (s *shard) lookupVersionedRLocked(key []byte, version uint64) ([]byte, bool func (s *shard) BatchGet(keys [][]byte, version uint64) (map[string][]byte, error) { results := make(map[string][]byte, len(keys)) - unresolved, hits, err := s.attemptFastBatchGetUnlocked(keys, results, version) + unresolved, staged, hits, err := s.attemptFastBatchGetUnlocked(keys, results, version) if err != nil { - return nil, err + return nil, fmt.Errorf("batch get of %d keys at version %d: %w", len(keys), version, err) } var pending []pendingRead if len(unresolved) > 0 { var remainingHits int64 - pending, remainingHits, err = s.batchGetRemainingUnlocked(keys, unresolved, results, version) + var remainingStaged []*pendingValue + pending, remainingStaged, remainingHits, err = + s.batchGetRemainingUnlocked(keys, unresolved, results, version) if err != nil { - return nil, err + return nil, fmt.Errorf("batch get of %d unresolved keys at version %d: %w", + len(unresolved), version, err) } hits += remainingHits + staged = append(staged, remainingStaged...) } if hits > 0 { @@ -252,37 +334,60 @@ func (s *shard) BatchGet(keys [][]byte, version uint64) (map[string][]byte, erro if err := s.cache.ResolveBatchUnlocked(pending, results); err != nil { // DB errors are fatal; fail the whole batch. - return nil, err + return nil, fmt.Errorf("complete %d database reads for a batch get: %w", len(pending), err) + } + // Awaited after the DB reads and outside every lock, for the reason given on pendingValue.await. + if err := s.awaitStagedReadsUnlocked(staged, results); err != nil { + return nil, fmt.Errorf("batch get at version %d: %w", version, err) } return results, nil } +// awaitStagedReadsUnlocked completes the staged folds a batch read ran into, writing each resolved value into +// results. A deleted key resolves to nil and is left out, as it would be on any other read path. +func (s *shard) awaitStagedReadsUnlocked(staged []*pendingValue, results map[string][]byte) error { + for _, pending := range staged { + value, err := pending.await(s.ctx, s.shutdownError) + if err != nil { + return fmt.Errorf("await staged value for key %x: %w", pending.key, err) + } + if value != nil { + results[pending.key] = value + } + } + return nil +} + // attemptFastBatchGetUnlocked resolves the keys it can while holding the read lock, writing found // values into results and returning the positions in keys of those it could not resolve. func (s *shard) attemptFastBatchGetUnlocked( keys [][]byte, results map[string][]byte, version uint64, -) (unresolved []int, hits int64, err error) { +) (unresolved []int, staged []*pendingValue, hits int64, err error) { s.lock.RLock() defer s.lock.RUnlock() // Checked ahead of the versioned data so that a shard taken out of service refuses every read, // not just those that would have reached the DB. if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { - return nil, 0, err + return nil, nil, 0, fmt.Errorf("resolve what is already in memory: %w", err) } if err := s.validateVersionRLocked(version); err != nil { - return nil, 0, err + return nil, nil, 0, fmt.Errorf("resolve what is already in memory: %w", err) } for i, key := range keys { keyStr := string(key) - if value, found := s.lookupVersionedRLocked(key, version); found { + if entry, found := s.lookupVersionedRLocked(key, version); found { + if entry.pending != nil { + staged = append(staged, entry.pending) + continue + } // found includes tombstones (nil value); only non-nil values are real hits to return. - if value != nil { - results[keyStr] = value + if entry.value != nil { + results[keyStr] = entry.value } hits++ continue @@ -300,7 +405,7 @@ func (s *shard) attemptFastBatchGetUnlocked( } unresolved = append(unresolved, i) } - return unresolved, hits, nil + return unresolved, staged, hits, nil } // batchGetRemainingUnlocked classifies the keys at the given positions in keys, which are those the @@ -310,28 +415,33 @@ func (s *shard) batchGetRemainingUnlocked( indices []int, results map[string][]byte, version uint64, -) (pending []pendingRead, hits int64, err error) { +) (pending []pendingRead, staged []*pendingValue, hits int64, err error) { pending = make([]pendingRead, 0, len(indices)) s.lock.Lock() defer s.lock.Unlock() if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { - return nil, 0, err + return nil, nil, 0, fmt.Errorf("classify the remaining keys: %w", err) } if err := s.validateVersionRLocked(version); err != nil { - return nil, 0, err + return nil, nil, 0, fmt.Errorf("classify the remaining keys: %w", err) } // Redone from scratch rather than carried over from the fast pass, because the lock was released - // in between and another reader may have scheduled or completed any of these keys. + // in between and another reader may have scheduled or completed any of these keys — or staged a + // fold for one. for _, i := range indices { key := keys[i] keyStr := string(key) - if value, found := s.lookupVersionedRLocked(key, version); found { - if value != nil { - results[keyStr] = value + if entry, found := s.lookupVersionedRLocked(key, version); found { + if entry.pending != nil { + staged = append(staged, entry.pending) + continue + } + if entry.value != nil { + results[keyStr] = entry.value } hits++ continue @@ -352,7 +462,7 @@ func (s *shard) batchGetRemainingUnlocked( needsSchedule: outcome.needsSchedule, }) } - return pending, hits, nil + return pending, staged, hits, nil } // GetSizeInfo returns the current cache size (bytes) and entry count under the read lock. @@ -392,7 +502,7 @@ func (s *shard) Set(key []byte, value []byte) error { defer s.lock.Unlock() if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { - return err + return fmt.Errorf("set key %x: %w", key, err) } s.setWLocked(key, value) return nil @@ -424,7 +534,7 @@ func (s *shard) BatchSet(entries []*proto.KVPair) error { // Checked once for the whole batch rather than per key: it cannot change while we hold the lock. if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { - return err + return fmt.Errorf("batch set of %d keys: %w", len(entries), err) } for i := range entries { if entries[i].Delete { @@ -437,6 +547,359 @@ func (s *shard) BatchSet(entries []*proto.KVPair) error { return nil } +// StageUpdates reserves a slot at the current version for each key named by indices, each holding an +// unresolved fold, and reports where each of those folds gets the value it folds onto. Folds staged +// for one key apply in the order they were staged. +func (s *shard) StageUpdates( + keys []string, + indices []int, + version uint64, +) ([]stagedFold, error) { + s.lock.Lock() + defer s.lock.Unlock() + + // Checked once for the whole batch rather than per key: it cannot change while we hold the lock. + if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { + return nil, fmt.Errorf("stage %d values at version %d: %w", len(indices), version, err) + } + if version != s.currentVersion { + return nil, fmt.Errorf("staging at version %d, but the current version is %d", + version, s.currentVersion) + } + + folds := make([]stagedFold, len(indices)) + for n, index := range indices { + key := keys[index] + folds[n] = stagedFold{ + prior: s.capturePriorValueWLocked(key), + result: newPendingValue(key), + } + s.stagePendingValueWLocked(key, version, folds[n].result) + } + return folds, nil +} + +// capturePriorValueWLocked reports what a fold staged now for key would be folding on top of: the newest +// value the shard holds, resolved or not, or neither when the shard holds none and it has to come +// from the read cache or the database. +func (s *shard) capturePriorValueWLocked(key string) priorValueSource { + deque, ok := s.versionedData[key] + if !ok || deque.IsEmpty() { + return priorValueSource{location: priorValueInReadCache} + } + // The newest entry is the right one to fold onto, whether it belongs to an earlier version or to + // an earlier write within this one. It can never belong to a later version: every write lands at + // the current version, and StageUpdates refuses any other, so nothing is ever appended above it. + newest := deque.PeekBack() + if newest.pending != nil { + return priorValueSource{location: priorValueInEarlierFold, pending: newest.pending} + } + return priorValueSource{location: priorValueInVersionedData, value: newest.value} +} + +// stagePendingValueWLocked puts an unresolved fold into the versioned data at the given version, +// replacing any entry this version already had for the key. +func (s *shard) stagePendingValueWLocked(key string, version uint64, pending *pendingValue) { + entry := versionedValue{version: version, pending: pending} + + deque, ok := s.versionedData[key] + if !ok { + deque = structures.NewDeque[versionedValue]() + // Cloned because this map entry outlives the batch that created it, and Go leaves a map's + // original key in place on reassignment. The copy is per key new to this shard, not per write. + s.versionedData[strings.Clone(key)] = deque + } + if deque.IsEmpty() || deque.PeekBack().version < version { + deque.PushBack(entry) + } else { + deque.PopBack() + deque.PushBack(entry) + } + + s.markFoldStagedWLocked(version) +} + +// markFoldStagedWLocked records one more of a version's folds as outstanding, creating the version's +// latch if this is its first. +func (s *shard) markFoldStagedWLocked(version uint64) { + latch, ok := s.versionLatches[version] + if !ok { + latch = &versionLatch{done: make(chan struct{})} + s.versionLatches[version] = latch + } + latch.count++ +} + +// markFoldResolvedWLocked records one of a version's folds as no longer outstanding, opening the +// version's latch when it was the last. A version left incomplete by a failure keeps its latch. +func (s *shard) markFoldResolvedWLocked(version uint64) { + latch, ok := s.versionLatches[version] + if !ok { + return + } + latch.count-- + if latch.count > 0 { + return + } + close(latch.done) + if latch.err == nil { + delete(s.versionLatches, version) + } +} + +// awaitVersionFoldsUnlocked blocks until every fold staged in the given version has resolved, reporting +// the failure that stopped one if any did. Must be called with no lock held. +func (s *shard) awaitVersionFoldsUnlocked(version uint64) error { + s.lock.RLock() + latch, outstanding := s.versionLatches[version] + s.lock.RUnlock() + if !outstanding { + return nil + } + + select { + case <-latch.done: + case <-s.ctx.Done(): + return fmt.Errorf("view manager shut down while awaiting version %d: %w", + version, s.shutdownError()) + } + + s.lock.RLock() + defer s.lock.RUnlock() + if latch.err != nil { + return fmt.Errorf("version %d holds a value that failed to resolve: %w", version, latch.err) + } + return nil +} + +// AwaitOutstandingFolds blocks until every fold this shard has staged has resolved, whatever it +// resolved to. +// +// The wait is not interruptible: it exists to keep a shutdown from overtaking a fold, and a manager +// that has already failed has already cancelled the context an interruptible wait would observe. +// Every fold resolves its version's latch whether it produced a value or failed, so the wait +// terminates either way. +func (s *shard) AwaitOutstandingFolds() { + s.lock.RLock() + latches := make([]*versionLatch, 0, len(s.versionLatches)) + for _, latch := range s.versionLatches { + latches = append(latches, latch) + } + s.lock.RUnlock() + + for _, latch := range latches { + <-latch.done + } +} + +// FoldStagedValues folds every value a batch staged and records what each produced. Either every fold +// in the batch is recorded or none is. +func (s *shard) FoldStagedValues(folds []stagedFold, updater BatchUpdater, version uint64) { + priorValues, err := s.resolvePriorValuesUnlocked(folds) + if err != nil { + s.FailStagedFolds(folds, version, err) + return + } + + newValues := make([][]byte, len(folds)) + for n := range folds { + newValues[n], err = updater.NewValueFor(folds[n].result.key, priorValues[n]) + if err != nil { + s.FailStagedFolds(folds, version, + fmt.Errorf("fold key %x at version %d: %w", folds[n].result.key, version, err)) + return + } + } + + s.recordFoldsUnlocked(folds, newValues, version) +} + +// resolvePriorValuesUnlocked produces the value each staged fold applies on top of. +func (s *shard) resolvePriorValuesUnlocked(folds []stagedFold) ([][]byte, error) { + values := make([][]byte, len(folds)) + var needRead []int + + for n := range folds { + switch folds[n].prior.location { + case priorValueInVersionedData: + values[n] = folds[n].prior.value + case priorValueInEarlierFold: + // Awaited outside the lock, which is why this runs here rather than during staging. + value, err := folds[n].prior.pending.await(s.ctx, s.shutdownError) + if err != nil { + return nil, fmt.Errorf("await the earlier fold of key %x: %w", folds[n].result.key, err) + } + values[n] = value + case priorValueInReadCache: + needRead = append(needRead, n) + default: + // The zero value lands here, which is the point: a priorValueSource built with its + // location left unset would otherwise be served as though its value had been read. + panic(fmt.Sprintf("unexpected prior value location: %#v", folds[n].prior.location)) + } + } + if len(needRead) == 0 { + return values, nil + } + + // Read through the cache rather than the versioned lookup: versioned data already holds this + // batch's own unresolved entry for these keys, so a versioned lookup would find each fold waiting + // on itself. + results := make(map[string][]byte, len(needRead)) + unresolved, err := s.priorValuesFromCacheUnlocked(folds, needRead, results) + if err != nil { + return nil, fmt.Errorf("read %d prior values from the cache: %w", len(needRead), err) + } + if len(unresolved) > 0 { + pending, err := s.schedulePriorValueReadsUnlocked(folds, unresolved, results) + if err != nil { + return nil, fmt.Errorf("schedule %d prior value reads: %w", len(unresolved), err) + } + if err := s.cache.ResolveBatchUnlocked(pending, results); err != nil { + return nil, fmt.Errorf("complete %d prior value reads: %w", len(pending), err) + } + } + + for _, n := range needRead { + values[n] = results[folds[n].result.key] + } + return values, nil +} + +// priorValuesFromCacheUnlocked resolves the prior values the cache already holds, returning the +// positions it could not. +func (s *shard) priorValuesFromCacheUnlocked( + folds []stagedFold, + positions []int, + results map[string][]byte, +) ([]int, error) { + s.lock.RLock() + defer s.lock.RUnlock() + + if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { + return nil, fmt.Errorf("look up %d prior values: %w", len(positions), err) + } + + var unresolved []int + for _, n := range positions { + key := folds[n].result.key + value, found, ok := s.cache.AttemptFastLookupRLocked([]byte(key), false) + if !ok { + unresolved = append(unresolved, n) + continue + } + if found { + results[key] = value + } + } + return unresolved, nil +} + +// schedulePriorValueReadsUnlocked classifies the prior values the cache could not resolve, scheduling +// database reads as needed. The reads themselves are completed by the caller, outside the lock. +func (s *shard) schedulePriorValueReadsUnlocked( + folds []stagedFold, + positions []int, + results map[string][]byte, +) ([]pendingRead, error) { + pending := make([]pendingRead, 0, len(positions)) + + s.lock.Lock() + defer s.lock.Unlock() + + if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { + return nil, fmt.Errorf("classify %d prior value reads: %w", len(positions), err) + } + + for _, n := range positions { + key := folds[n].result.key + outcome := s.cache.LookupWLocked([]byte(key), false) + if outcome.immediate { + if outcome.found { + results[key] = outcome.value + } + continue + } + pending = append(pending, pendingRead{ + key: key, + entry: outcome.entry, + valueChan: outcome.valueChan, + needsSchedule: outcome.needsSchedule, + }) + } + return pending, nil +} + +// recordFoldsUnlocked stores what every fold in a batch produced: into the versioned entry staged for it, +// into the version's diff, and into the handle its observers are waiting on. +func (s *shard) recordFoldsUnlocked(folds []stagedFold, newValues [][]byte, version uint64) { + s.lock.Lock() + defer s.lock.Unlock() + + for n := range folds { + key := folds[n].result.key + if s.fillStagedValueWLocked(key, version, folds[n].result, newValues[n]) { + s.versionDiffs[version][key] = newValues[n] + } + s.markFoldResolvedWLocked(version) + // Released under the lock deliberately. A woken observer reads the versioned data, so it has + // to queue behind this hold anyway, and releasing here leaves no window in which the entry is + // filled but its observers are still parked. + folds[n].result.inject(newValues[n], nil) + } +} + +// fillStagedValueWLocked replaces a staged entry with the value its fold produced, reporting whether +// that value is still the one the key holds at this version. A value that is not must be kept out of +// the version's diff. +func (s *shard) fillStagedValueWLocked( + key string, + version uint64, + pending *pendingValue, + value []byte, +) bool { + deque, ok := s.versionedData[key] + if !ok { + // Retirement is the only thing that removes a key, and it refuses a version whose folds are + // still outstanding. Reaching here means that invariant broke, and carrying on would lose + // the write silently. + panic(fmt.Sprintf("no versioned data for staged key %x at version %d", key, version)) + } + + for i := deque.Len() - 1; i >= 0; i-- { + entry := deque.Get(i) + if entry.pending == pending { + deque.Set(i, versionedValue{value: value, version: version}) + return true + } + if entry.version < version { + break + } + } + // The entry is gone, so a later fold or plain write at this version took its place and won. At most + // one entry per version exists, so finding none means this fold's value has been superseded. + return false +} + +// FailStagedFolds records a failed fold on every value a batch staged and takes the shard out of +// service. The version keeps its latch, carrying the failure. +func (s *shard) FailStagedFolds(folds []stagedFold, version uint64, err error) { + s.lock.Lock() + if latch, ok := s.versionLatches[version]; ok && latch.err == nil { + latch.err = err + } + s.cache.TakeOutOfServiceWLocked(err) + for n := range folds { + s.markFoldResolvedWLocked(version) + folds[n].result.inject(nil, err) + } + s.lock.Unlock() + + // Reported after the lock is released: bricking takes the manager's versionLock and then every + // shard's, this one included. + s.reportFoldFailure(err) +} + // Delete deletes the value for the given key. func (s *shard) Delete(key []byte) error { return s.Set(key, nil) @@ -460,7 +923,11 @@ func (s *shard) Commit() (uint64, error) { s.lock.Unlock() - return newVersion, err + if err != nil { + return newVersion, fmt.Errorf("maintain the read cache after sealing version %d: %w", + newVersion, err) + } + return newVersion, nil } // Get the diffs for a range of versions [firstVersion, lastVersion). The returned data should not be mutated @@ -477,9 +944,19 @@ func (s *shard) GetDiffsForVersions( firstVersion, lastVersion) } + // Awaited before the lock is taken, not after: a sealed version's diff keeps being written to + // while its folds resolve, so it is frozen only once its latch has opened. This is the single + // place the diff consumers — hashing, flushing, and the retirement that follows a flush — reach + // a version's values, which is why the wait belongs here rather than at each of them. + for version := firstVersion; version < lastVersion; version++ { + if err := s.awaitVersionFoldsUnlocked(version); err != nil { + return nil, fmt.Errorf("await version %d before reading its diff: %w", version, err) + } + } + // A read lock suffices, and it matters: sort jobs for different versions call this concurrently. - // Nothing here mutates the shard, and the maps handed back are frozen — only versionDiffs at the - // current version is ever written to, so a version stops changing the moment it is no longer current. + // Nothing here mutates the shard, and the maps handed back are frozen — a version whose latch has + // opened gains no further writes. s.lock.RLock() defer s.lock.RUnlock() @@ -506,13 +983,38 @@ func (s *shard) GetDiffsForVersions( // Because the target is always the current version, each key resolves to the back of its deque — // no version scan is needed, unlike lookupVersionedRLocked, which serves reads at older versions. func (s *shard) MaterializeCurrentOverrides(lowerBound []byte, upperBound []byte) ([]kvPair, error) { + // Retried rather than resolved in place, because a fold cannot complete while this lock is held. + // One retry is the normal case: iterator construction must not race a batch write, so no further + // values are staged while the first pass's folds are awaited. + for { + pairs, staged, err := s.materializeAttemptUnlocked(lowerBound, upperBound) + if err != nil { + return nil, fmt.Errorf("materialize the current overrides: %w", err) + } + if len(staged) == 0 { + return pairs, nil + } + for _, pending := range staged { + if _, err := pending.await(s.ctx, s.shutdownError); err != nil { + return nil, fmt.Errorf("await a staged value before materializing: %w", err) + } + } + } +} + +// materializeAttemptUnlocked copies the in-memory overrides in range, or reports the folds that have +// to resolve before they can be copied. A non-empty staged result means pairs is incomplete. +func (s *shard) materializeAttemptUnlocked( + lowerBound []byte, + upperBound []byte, +) (pairs []kvPair, staged []*pendingValue, err error) { s.lock.RLock() defer s.lock.RUnlock() // Same reason the read paths check it: a shard taken out of service cannot vouch for its data, // and an iterator is just a bulk read. if err := s.cache.ErrIfOutOfServiceRLocked(); err != nil { - return nil, err + return nil, nil, fmt.Errorf("read the current overrides: %w", err) } out := make([]kvPair, 0, len(s.versionedData)) @@ -526,12 +1028,20 @@ func (s *shard) MaterializeCurrentOverrides(lowerBound []byte, upperBound []byte if upperBound != nil && key >= string(upperBound) { continue } + newest := deque.PeekBack() + if newest.pending != nil { + staged = append(staged, newest.pending) + continue + } out = append(out, kvPair{ key: []byte(key), - value: deque.PeekBack().value, + value: newest.value, }) } - return out, nil + if len(staged) > 0 { + return nil, staged, nil + } + return out, nil, nil } // Drop versions, pushing their data down into the read cache. The first version to drop must be @@ -560,6 +1070,17 @@ func (s *shard) DropVersions( lastVersion, s.currentVersion) } + // Retirement is driven off the version diffs, so a version with folds outstanding would retire + // without the keys those folds have yet to write: their versioned entries would never be dropped + // and their values would never reach the cache. Retirement only ever follows a flush, which waits + // for the same latch, so this reports a broken lifecycle rather than a race to be waited out. + for version := firstVersion; version < lastVersion; version++ { + if latch, outstanding := s.versionLatches[version]; outstanding { + return fmt.Errorf("version %d still has %d unresolved value(s) and cannot be retired", + version, latch.count) + } + } + // Combine the data from all versions being dropped. var combinedData map[string][]byte if firstVersion == lastVersion-1 { diff --git a/sei-db/db_engine/view/shard_manager.go b/sei-db/db_engine/view/shard_manager.go index 4ffb6b55f8..c55126b3d9 100644 --- a/sei-db/db_engine/view/shard_manager.go +++ b/sei-db/db_engine/view/shard_manager.go @@ -44,3 +44,10 @@ func (s *shardManager) Shard(addr []byte) uint64 { return x & s.mask } + +// ShardString is Shard for a key already held as a string. maphash.String is defined as +// Bytes(seed, []byte(addr)), and Shard's pooled Hash computes the same seeded sum, so a key lands in +// the same shard whichever form it arrives in. +func (s *shardManager) ShardString(addr string) uint64 { + return maphash.String(s.seed, addr) & s.mask +} diff --git a/sei-db/db_engine/view/shutdown_test.go b/sei-db/db_engine/view/shutdown_test.go index 564d0478ef..e5facad1e3 100644 --- a/sei-db/db_engine/view/shutdown_test.go +++ b/sei-db/db_engine/view/shutdown_test.go @@ -245,3 +245,86 @@ func TestCloseLeavesNoManagerGoroutines(t *testing.T) { 2*time.Second, 10*time.Millisecond, "manager goroutines leaked across create/use/close cycles") } + +// Close must not return while a fold staged by BatchUpdate is still reading its prior value: that +// read goes through the database Close is about to release. +func TestCloseAwaitsFoldReadingItsPriorValue(t *testing.T) { + db := newTestDB(map[string][]byte{"k": []byte("old")}) + manager := newTestManagerWithDB(t, db, 1, 4096) + + // Baseline past the construction-time initial-hash read, then gate all further DB reads. + base := db.getCalls.Load() + db.getGate = make(chan struct{}) + + // Release the gated read exactly once, and unconditionally on test failure, before the + // pool-draining cleanup registered at construction (t.Cleanup runs LIFO). + releaseGate := sync.OnceFunc(func() { close(db.getGate) }) + t.Cleanup(releaseGate) + + require.NoError(t, manager.BatchUpdate([]string{"k"}, markUpdater{mark: '+'})) + require.Eventually(t, func() bool { return db.getCalls.Load() > base }, + 2*time.Second, time.Millisecond, "the fold never reached the DB for its prior value") + + closeDone := make(chan error, 1) + go func() { closeDone <- manager.Close() }() + + select { + case <-closeDone: + t.Fatal("Close returned while a fold was still reading its prior value") + case <-time.After(100 * time.Millisecond): + } + + releaseGate() + select { + case err := <-closeDone: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("Close did not return after the fold's read was released") + } + + require.Zero(t, db.getsAfterClose.Load(), "a fold read the database after it was closed") +} + +// The production teardown order — manager, then pools, then DB (see CommitStore.Close) — must hold +// with a fold in flight. The fold here has yet to schedule the read of one of its keys, so a Close +// that returned before it resolved would leave it submitting that read to a closed pool. +func TestCloseAwaitsFoldBeforeItSchedulesItsRead(t *testing.T) { + db := newTestDB(map[string][]byte{"a": []byte("old"), "b": []byte("old")}) + readPool := threading.NewAdHocPool() + miscPool := threading.NewAdHocPool() + manager, err := NewViewManager(newTestConfig(1, 4096), db, readPool, miscPool) + require.NoError(t, err) + + // The first batch parks mid-fold, holding the value the second batch folds onto. + parked := newParkedUpdater() + require.NoError(t, manager.BatchUpdate([]string{"a"}, parked)) + <-parked.started + + // The second batch takes "a" from the parked fold and "b" from the database. Prior values are + // awaited before any read is scheduled, so this fold parks with "b" unread. + require.NoError(t, manager.BatchUpdate([]string{"a", "b"}, markUpdater{mark: '+'})) + + closeDone := make(chan error, 1) + go func() { closeDone <- manager.Close() }() + + select { + case <-closeDone: + close(parked.release) + t.Fatal("Close returned while two folds were still outstanding") + case <-time.After(100 * time.Millisecond): + } + + close(parked.release) + select { + case err := <-closeDone: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("Close did not return after the parked fold was released") + } + + // Closing the pools is what would panic on a fold that outlived Close. + readPool.Close() + miscPool.Close() + require.NoError(t, db.Close()) + require.Zero(t, db.getsAfterClose.Load(), "a fold read the database after it was closed") +} diff --git a/sei-db/db_engine/view/test_helpers_test.go b/sei-db/db_engine/view/test_helpers_test.go index 541a764fea..047ba3c5a8 100644 --- a/sei-db/db_engine/view/test_helpers_test.go +++ b/sei-db/db_engine/view/test_helpers_test.go @@ -42,6 +42,9 @@ type testDB struct { commitBlock chan struct{} getGate chan struct{} closed atomic.Bool + // Incremented when a Get reaches the store after Close. Lets tests assert that nothing read the + // database once it was released. + getsAfterClose atomic.Int64 // Batch lifecycle counters: batchesCreated increments in NewBatch, batchesClosed on a // batch's first Close. Lets tests assert every created batch is released (types.Batch // requires Close even after a successful Commit). @@ -62,6 +65,9 @@ func (d *testDB) Get(key []byte) ([]byte, error) { if d.getGate != nil { <-d.getGate } + if d.closed.Load() { + d.getsAfterClose.Add(1) + } if d.getErr != nil { return nil, d.getErr } @@ -313,9 +319,10 @@ func newTestShard(t *testing.T, maxSize uint64, db *testDB) *shard { config := DefaultTestViewManagerConfig() config.EstimatedOverheadPerEntry = 0 // A standalone shard has no manager to brick, and it takes itself out of service on a failed read - // without help, so reporting is a no-op here. + // or fold without help, so both reports are no-ops here. s, err := NewShard(context.Background(), config, db, threading.NewAdHocPool(), maxSize, func() error { return ErrViewManagerClosed }, + func(error) {}, func(error) {}) require.NoError(t, err) return s diff --git a/sei-db/db_engine/view/view_manager.go b/sei-db/db_engine/view/view_manager.go index dd0f595ae1..b25e18178e 100644 --- a/sei-db/db_engine/view/view_manager.go +++ b/sei-db/db_engine/view/view_manager.go @@ -14,6 +14,15 @@ import ( // closed normally rather than failed. Detect it with errors.Is. var ErrViewManagerClosed = errors.New("view manager closed") +// BatchUpdater produces the value to write for each of a batch's keys, from the value that key +// currently holds. One BatchUpdater serves every key in a BatchUpdate call. +type BatchUpdater interface { + // NewValueFor returns the value to write for key, or nil to delete it. priorValue is the value + // key currently holds, or nil if it holds none. Called concurrently, after BatchUpdate has + // returned, and must neither retain nor mutate priorValue. + NewValueFor(key string, priorValue []byte) ([]byte, error) +} + // ViewManager provides a read-through cache and efficient point-in-time views on top of a basic // key-value database. It also coordinates writes to the database, since efficient views require // careful staging of inserts. @@ -71,12 +80,25 @@ type ViewManager interface { // Iterator). BatchSet(updates []*proto.KVPair) error + // BatchUpdate stages a value for every key in keys, to be produced later by handing that key's + // prior value to updater. Where BatchSet takes the values, this takes a function of the values + // already stored. + // + // It returns as soon as the keys are staged, before any prior value has been read and before any + // value has been produced. From that moment the keys read as their new values: a read of one + // blocks until its value is available. A failure to produce a value is reported to whatever + // reads, hashes or flushes that key, and bricks the manager. + // + // keys must not repeat. Not visible to iterators created earlier (see Iterator). + BatchUpdate(keys []string, updater BatchUpdater) error + // Commit seals the current version as an immutable, point-in-time View and advances the // manager to a fresh mutable version. The returned View is safe to read for as long as the // caller holds a reservation on it; see View for the full lifecycle contract. // // Commit must not be called concurrently with operations on the current (mutable) - // version — Get, BatchGet, Set, Delete, BatchSet, or the construction of an Iterator. Reads of + // version — Get, BatchGet, Set, Delete, BatchSet, BatchUpdate, or the construction of an + // Iterator. Reads of // sealed views may proceed concurrently with it, and so may reads through an already-constructed // Iterator: an iterator is fixed at its creation instant, so a seal cannot disturb it. // @@ -98,7 +120,7 @@ type ViewManager interface { // Equally, it will never show them — a caller that wants later writes needs a new iterator. // Holding one is therefore safe from another thread, and does not block writes. // - // Constructing an iterator must NOT race a BatchSet. Each shard's overrides are copied under that + // Constructing an iterator must NOT race a BatchSet or a BatchUpdate. Each shard's overrides are copied under that // shard's own lock, so a batch spanning two shards during construction can leave the iterator // holding part of it — a state belonging to no single instant, reported without an error. Serialize // construction against BatchSet. Set and Delete each touch a single shard and so are seen either diff --git a/sei-db/db_engine/view/view_manager_impl.go b/sei-db/db_engine/view/view_manager_impl.go index b2124413d7..66894c8912 100644 --- a/sei-db/db_engine/view/view_manager_impl.go +++ b/sei-db/db_engine/view/view_manager_impl.go @@ -203,8 +203,8 @@ func NewViewManager( // cancellation — see the Close contract on ViewManager. shards := make([]*shard, config.ShardCount) for i := uint64(0); i < config.ShardCount; i++ { - shards[i], err = NewShard( - childCtx, config, db, readPool, sizePerShard, c.shutdownError, c.reportReadFailure) + shards[i], err = NewShard(childCtx, config, db, readPool, sizePerShard, + c.shutdownError, c.reportReadFailure, c.reportFoldFailure) if err != nil { cancel() return nil, fmt.Errorf("failed to create shard: %w", err) @@ -271,6 +271,71 @@ func (c *viewManager) BatchSet(updates []*proto.KVPair) error { return nil } +func (c *viewManager) BatchUpdate(keys []string, updater BatchUpdater) error { + work := c.partitionIndicesByShard(keys) + version := c.currentVersion + + // Staging is synchronous, and it is all this call does on the caller's thread: one lock hold per + // shard that reads no database and folds nothing. It is what makes the keys read as their new + // values before any fold has run. + staged := make([][]stagedFold, len(c.shards)) + for shardIndex := range work { + if len(work[shardIndex]) == 0 { + continue + } + folds, err := c.shards[shardIndex].StageUpdates(keys, work[shardIndex], version) + if err != nil { + // The shards that already staged are holding values nothing will ever fold, and a reader + // would park on one forever. Fail them before reporting. + c.abandonStaged(staged, version, err) + return fmt.Errorf("failed to stage update in shard: %w", err) + } + staged[shardIndex] = folds + } + + // Folding happens here, off this thread, and nothing waits for it: whatever reads, hashes or + // flushes one of these keys is what waits. + for shardIndex, folds := range staged { + if len(folds) == 0 { + continue + } + shard := c.shards[shardIndex] + c.miscPool.Submit(func() { + shard.FoldStagedValues(folds, updater, version) + }) + } + return nil +} + +// abandonStaged fails every fold staged so far, for a BatchUpdate that could not finish staging. +func (c *viewManager) abandonStaged(staged [][]stagedFold, version uint64, err error) { + for shardIndex, folds := range staged { + if len(folds) == 0 { + continue + } + c.shards[shardIndex].FailStagedFolds(folds, version, err) + } +} + +// partitionIndicesByShard groups the positions of keys by the shard each key belongs to, so each +// shard is visited once. The returned slice is indexed by shard, and a shard no key landed in holds +// an empty bucket. +// +// Buckets start out sized for an even spread, which is what the seeded hash produces; a bucket that +// lands above its share still grows on demand. +func (c *viewManager) partitionIndicesByShard(keys []string) [][]int { + work := make([][]int, len(c.shards)) + perShard := len(keys)/len(c.shards) + 1 + for index, key := range keys { + shardIndex := c.shardManager.ShardString(key) + if work[shardIndex] == nil { + work[shardIndex] = make([]int, 0, perShard) + } + work[shardIndex] = append(work[shardIndex], index) + } + return work +} + func (c *viewManager) BatchGet(keys [][]byte) (map[string][]byte, error) { return c.BatchGetAtVersion(keys, c.currentVersion) } @@ -795,6 +860,15 @@ func (c *viewManager) reportReadFailure(err error) { c.brick(fmt.Errorf("failed to read from the underlying database: %w", err)) } +// reportFoldFailure handles a fold that could not produce its value by bricking the manager. The +// latched error names the fold rather than the read that may have fed it. +// +// Must be called without the shard lock held: it acquires versionLock, and the established order is +// versionLock before any shard lock. +func (c *viewManager) reportFoldFailure(err error) { + c.brick(fmt.Errorf("failed to fold a staged value: %w", err)) +} + // brickLocked latches the fatal error, cancels the manager context, wakes backpressure waiters, and // takes every shard out of service, so callers observe the failure immediately rather than waiting // for Close. @@ -1131,6 +1205,19 @@ func (c *viewManager) Close() error { return c.closeErr } +// awaitOutstandingFolds blocks until every fold staged by BatchUpdate has resolved in every shard. +// +// A fold is the manager's only background work that no caller waits for: it runs on a pool, reads +// through to the database the manager owns, and submits that read to a pool the manager's owner +// closes once Close returns. Close calls this before cancelling, so that a fold in flight resolves +// against an open database instead of abandoning a read that would then race db.Close, and outside +// versionLock, which a failing fold takes to brick the manager. +func (c *viewManager) awaitOutstandingFolds() { + for _, s := range c.shards { + s.AwaitOutstandingFolds() + } +} + func (c *viewManager) closeInternal() error { // Tell the lifecycle runner to exit, then wait for it to report offline. The send is // buffered, so it does not block when the runner has already exited (manager failure), and @@ -1138,6 +1225,8 @@ func (c *viewManager) closeInternal() error { c.lifecycleExit <- struct{}{} <-c.lifecycleExited + c.awaitOutstandingFolds() + // Release everyone blocked on the manager's future: AwaitFlush, backpressured // View callers, and reads still awaiting results. The cancel happens under versionLock // because backpressure waiters re-check the context under that lock before parking on the diff --git a/sei-db/state_db/sc/flatkv/metrics.go b/sei-db/state_db/sc/flatkv/metrics.go index 7a83003c91..a02fb23530 100644 --- a/sei-db/state_db/sc/flatkv/metrics.go +++ b/sei-db/state_db/sc/flatkv/metrics.go @@ -16,26 +16,26 @@ var ( flatkvMeter = otel.Meter(flatkvMeterName) otelMetrics = struct { - OpenLatency metric.Float64Histogram - ApplyChangesetsLatency metric.Float64Histogram - CommitLatency metric.Float64Histogram - CommitBatchLatency metric.Float64Histogram - BatchReadOldValuesLatency metric.Float64Histogram - NumKVPairs metric.Int64Counter - PendingWrites metric.Int64Gauge - CurrentVersion metric.Int64Gauge - CatchupLatency metric.Float64Histogram - CatchupReplayNumBlocks metric.Int64Counter - SnapshotWriteLatency metric.Float64Histogram - SnapshotQueue *commonmetrics.QueueMeter - SnapshotPruneLatency metric.Float64Histogram - SnapshotPruneAttempts metric.Int64Counter - CurrentSnapshotHeight metric.Int64Gauge - RollbackLatency metric.Float64Histogram - ImportLatency metric.Float64Histogram - ImportKVPairs metric.Int64Counter - ImportWorkerFlushLatency metric.Float64Histogram - FlushLatency metric.Float64Histogram + OpenLatency metric.Float64Histogram + ApplyChangesetsLatency metric.Float64Histogram + CommitLatency metric.Float64Histogram + CommitBatchLatency metric.Float64Histogram + AccountUpdateLatency metric.Float64Histogram + NumKVPairs metric.Int64Counter + PendingWrites metric.Int64Gauge + CurrentVersion metric.Int64Gauge + CatchupLatency metric.Float64Histogram + CatchupReplayNumBlocks metric.Int64Counter + SnapshotWriteLatency metric.Float64Histogram + SnapshotQueue *commonmetrics.QueueMeter + SnapshotPruneLatency metric.Float64Histogram + SnapshotPruneAttempts metric.Int64Counter + CurrentSnapshotHeight metric.Int64Gauge + RollbackLatency metric.Float64Histogram + ImportLatency metric.Float64Histogram + ImportKVPairs metric.Int64Counter + ImportWorkerFlushLatency metric.Float64Histogram + FlushLatency metric.Float64Histogram }{ OpenLatency: must(flatkvMeter.Float64Histogram( "flatkv_open_latency", @@ -61,9 +61,11 @@ var ( metric.WithUnit("s"), metric.WithExplicitBucketBoundaries(commonmetrics.LatencyBuckets...), )), - BatchReadOldValuesLatency: must(flatkvMeter.Float64Histogram( - "flatkv_batch_read_old_values_latency", - metric.WithDescription("Time taken to batch read old FlatKV values"), + AccountUpdateLatency: must(flatkvMeter.Float64Histogram( + "flatkv_account_update_latency", + metric.WithDescription( + "Time taken to stage one block's account changes with the account store, which folds "+ + "them onto the rows they modify on its own threads"), metric.WithUnit("s"), metric.WithExplicitBucketBoundaries(commonmetrics.LatencyBuckets...), )), diff --git a/sei-db/state_db/sc/flatkv/store_apply.go b/sei-db/state_db/sc/flatkv/store_apply.go index 73d4841830..e1aeba42a7 100644 --- a/sei-db/state_db/sc/flatkv/store_apply.go +++ b/sei-db/state_db/sc/flatkv/store_apply.go @@ -87,7 +87,7 @@ func (s *CommitStore) applyChangeSets( logger.Debug("FlatKV ApplyChangeSets complete", "version", version, "changesets", len(changeSets), - "writes", len(prepared.accounts)+len(prepared.storage)+len(prepared.code)+len(prepared.misc), + "writes", prepared.accountCount()+len(prepared.storage)+len(prepared.code)+len(prepared.misc), "elapsed", obs.elapsed()) return nil } @@ -96,7 +96,7 @@ func (s *CommitStore) applyChangeSets( // ApplyChangeSets call. Nothing here reaches a store until every kind has validated — see // writeToStores. type preparedWrites struct { - accounts map[string]*vtype.AccountData + accounts *accountUpdater storage map[string]*vtype.StorageData code map[string]*vtype.CodeData misc map[string]*vtype.MiscData @@ -109,29 +109,20 @@ func (s *CommitStore) prepareWrites( ) (preparedWrites, error) { var out preparedWrites - // A nonce, codehash or balance change carries only its own field, so it has to be merged onto the - // account as it stands right now — a live read, since anything an earlier call at this height wrote - // counts. - s.phaseTimer.SetPhase("apply_change_sets_read_accounts") - readStart := time.Now() - accountOld, err := s.readAccountsForMerge(changesByType) - otelMetrics.BatchReadOldValuesLatency.Record(s.ctx, secondsSince(readStart), - metric.WithAttributes(successAttr(err))) - if err != nil { - return out, err - } - s.phaseTimer.SetPhase("apply_change_sets_gather_values") - accountUpdates, err := mergeAccountUpdates( + // Only the changeset's own field values are parsed here. Folding them onto the rows those + // accounts already hold is left to the account store, which does it off this thread; see + // accountUpdater. + accountWrites, err := newAccountUpdater( changesByType[keys.EVMKeyNonce], changesByType[keys.EVMKeyCodeHash], changesByType[keys.EVMKeyBalance], + blockHeight, ) if err != nil { - return out, fmt.Errorf("failed to gather account updates: %w", err) + return out, fmt.Errorf("prepare account writes for block %d: %w", blockHeight, err) } - newAccounts := deriveNewAccountValues(accountUpdates, accountOld, blockHeight) storageWrites, err := toStorageValues(changesByType[keys.EVMKeyStorage], blockHeight) if err != nil { @@ -148,44 +139,89 @@ func (s *CommitStore) prepareWrites( return out, fmt.Errorf("failed to parse misc changes: %w", err) } - out.accounts = newAccounts + out.accounts = accountWrites out.storage = storageWrites out.code = codeWrites out.misc = miscWrites return out, nil } -// readAccountsForMerge reads the accounts that this batch's nonce, codehash and balance changes touch, -// so those partial updates can be merged onto whole accounts. Keys come from all three kinds, since any -// one of them can name an account the others do not. -func (s *CommitStore) readAccountsForMerge( - changesByType map[keys.EVMKeyKind]map[string][]byte, -) (map[string]*vtype.AccountData, error) { - accountKinds := []keys.EVMKeyKind{keys.EVMKeyNonce, keys.EVMKeyCodeHash, keys.EVMKeyBalance} +var _ view.BatchUpdater = (*accountUpdater)(nil) + +// accountUpdater folds one block's per-field account changes onto the rows those accounts already +// hold. +// +// An account is stored as one row but written a field at a time, so a change carrying only a nonce or +// only a code hash has to be applied on top of the row as it stands. The account store does that fold +// on its own threads, after the write has been staged, so no part of it runs on the thread applying +// the block. +type accountUpdater struct { + // pending is the fields this block set, keyed by physical key. Parsed up front, so a fold can + // never fail on a malformed change. + pending map[string]*vtype.PendingAccountWrite + + // keys names every account the block touched, in the form BatchUpdate takes them. + keys []string + + // blockHeight is stamped on every row written, whether or not any field value changed, because + // GetBlockHeightModified reports it. + blockHeight int64 +} + +// newAccountUpdater parses one batch's per-field account changes into the fields to set on each +// account. Reports nil when the batch touches no account. +// +// Parsing here rather than during the fold is what keeps a malformed changeset from being discovered +// halfway through writing the block: by the time the folds run, the block has already been accepted. +func newAccountUpdater( + nonceChanges map[string][]byte, + codeHashChanges map[string][]byte, + balanceChanges map[string][]byte, + blockHeight int64, +) (*accountUpdater, error) { + pending, err := mergeAccountUpdates(nonceChanges, codeHashChanges, balanceChanges) + if err != nil { + return nil, fmt.Errorf("failed to gather account updates: %w", err) + } + if len(pending) == 0 { + return nil, nil + } - size := 0 - for _, kind := range accountKinds { - size += len(changesByType[kind]) + physKeys := make([]string, 0, len(pending)) + for key := range pending { + physKeys = append(physKeys, key) } - touched := make(map[string]struct{}, size) - for _, kind := range accountKinds { - for key := range changesByType[kind] { - touched[key] = struct{}{} + return &accountUpdater{pending: pending, keys: physKeys, blockHeight: blockHeight}, nil +} + +// NewValueFor folds this block's changes to one account onto the row it already holds. An account the +// store does not hold starts from zero, and a row left with no balance, nonce or code hash is deleted. +func (u *accountUpdater) NewValueFor(key string, priorValue []byte) ([]byte, error) { + var stored *vtype.AccountData + if priorValue != nil { + parsed, err := vtype.DeserializeAccountData(priorValue) + if err != nil { + return nil, fmt.Errorf("failed to deserialize accountDB old value: %w", err) } + stored = parsed } - if len(touched) == 0 { + + // Merge copies rather than writing through, so the value handed back does not alias the row the + // store still holds for earlier versions. + merged := u.pending[key].Merge(stored, u.blockHeight) + if merged.IsDelete() { return nil, nil } + return merged.Serialize(), nil +} - physKeys := make([][]byte, 0, len(touched)) - for key := range touched { - physKeys = append(physKeys, []byte(key)) - } - raw, err := s.accountStore.BatchGet(physKeys) - if err != nil { - return nil, fmt.Errorf("read accounts to merge onto: %w", err) +// accountCount reports how many accounts the block writes, treating a block that touches none as zero +// rather than requiring the caller to nil-check. +func (p preparedWrites) accountCount() int { + if p.accounts == nil { + return 0 } - return deserializeOldAccounts(raw) + return len(p.accounts.keys) } // writeToStores writes one successful ApplyChangeSets batch into the four data stores and records the @@ -207,11 +243,15 @@ func (s *CommitStore) writeToStores( // TODO: currently, WAL replay may replay blocks already in some stores. In the future when WAL replay is external, // we may be able to simplify this code since we will be able to assume that all stores start at the same block. - if alreadyHave[accountDBDir] < version { - if err := serializeAndPut(s.accountStore, prepared.accounts); err != nil { + if alreadyHave[accountDBDir] < version && prepared.accounts != nil { + start := time.Now() + err := s.accountStore.BatchUpdate(prepared.accounts.keys, prepared.accounts) + otelMetrics.AccountUpdateLatency.Record(s.ctx, secondsSince(start), + metric.WithAttributes(successAttr(err))) + if err != nil { return fmt.Errorf("write %s values: %w", accountDBDir, err) } - addKVPairs(s.ctx, accountDBDir, len(prepared.accounts)) + addKVPairs(s.ctx, accountDBDir, len(prepared.accounts.keys)) } if alreadyHave[storageDBDir] < version { if err := serializeAndPut(s.storageStore, prepared.storage); err != nil { @@ -259,33 +299,15 @@ func serializeAndPut[T vtype.VType](store view.ViewManager, values map[string]T) return nil } -// deserializeOldAccounts parses the account database's old values into AccountData. A partial update — -// a nonce without a codehash, say — has to be merged onto the account that is already there, which -// needs the old value in structured form rather than as bytes. -// -// raw is keyed by physical key, and a key that had no prior value maps to nil; those are dropped -// rather than deserialized, so the result holds only accounts that already existed. -func deserializeOldAccounts(raw map[string][]byte) (map[string]*vtype.AccountData, error) { - old := make(map[string]*vtype.AccountData, len(raw)) - for key, b := range raw { - if b == nil { - continue - } - v, err := vtype.DeserializeAccountData(b) - if err != nil { - return nil, fmt.Errorf("failed to deserialize accountDB old value: %w", err) - } - old[key] = v - } - return old, nil -} - // moduleOfKey extracts the owning module from a physical key. Injected into the // lthash HashCalculator so it can bucket pairs by module without importing ktype // (ktype already imports lthash). func moduleOfKey(physicalKey []byte) (string, error) { module, _, err := ktype.StripModulePrefix(physicalKey) - return module, err + if err != nil { + return "", fmt.Errorf("strip the module prefix from key %x: %w", physicalKey, err) + } + return module, nil } // classifyAndPrefix splits changeSets into per-EVMKeyKind maps whose keys are @@ -491,23 +513,3 @@ func mergeAccountUpdates( } return updates, nil } - -// Combine the pending account writes with prior values to determine the new account values. -// -// We need to take this step because accounts are split into multiple fields, and it's possible to overwrite just a -// single field (thus requiring us to copy the unmodified fields from the prior value). -func deriveNewAccountValues( - pendingWrites map[string]*vtype.PendingAccountWrite, - oldValues map[string]*vtype.AccountData, - blockHeight int64, -) map[string]*vtype.AccountData { - result := make(map[string]*vtype.AccountData, len(pendingWrites)) - - for addrStr, pendingWrite := range pendingWrites { - oldValue := oldValues[addrStr] - - newValue := pendingWrite.Merge(oldValue, blockHeight) - result[addrStr] = newValue - } - return result -} diff --git a/sei-db/state_db/sc/flatkv/store_replay_test.go b/sei-db/state_db/sc/flatkv/store_replay_test.go index 7b7393331f..af72028c14 100644 --- a/sei-db/state_db/sc/flatkv/store_replay_test.go +++ b/sei-db/state_db/sc/flatkv/store_replay_test.go @@ -411,7 +411,7 @@ func TestReplaySkipDoesNotRewindRecordedHeight(t *testing.T) { // TestReplayConvergesOnPartialAccountFieldWrites pins the one case where replaying // a block into a DB that already holds it is not obviously a no-op. An account row -// is a merge, not an overwrite: deriveNewAccountValues folds a nonce-only or +// is a merge, not an overwrite: accountUpdater folds a nonce-only or // codehash-only update onto whatever is currently on disk. Replaying a range where // different blocks touch different fields therefore rebuilds the row field by field // through intermediate values that were never on-chain. It converges because the diff --git a/sei-db/state_db/sc/flatkv/store_test.go b/sei-db/state_db/sc/flatkv/store_test.go index a2d89fc85a..0f85348596 100644 --- a/sei-db/state_db/sc/flatkv/store_test.go +++ b/sei-db/state_db/sc/flatkv/store_test.go @@ -1,6 +1,7 @@ package flatkv import ( + "fmt" "os" "path/filepath" "testing" @@ -1596,15 +1597,31 @@ func TestCrashRecoveryCorruptedAccountValueInDB(t *testing.T) { defer s2.Close() require.NoError(t, s2.LoadLatest()) - // Applying a partial nonce update reads the old account back to merge onto it, and must reject the - // corrupted row instead of merging onto garbage. + // A partial nonce update has to be folded onto the account already stored, which the account store + // does off this thread. Applying the block therefore succeeds: the corrupted row has not been read + // yet, and nothing waits for it to be. cs2 := &proto.NamedChangeSet{ Name: "evm", Changeset: proto.ChangeSet{Pairs: []*proto.KVPair{noncePair(addr, 99)}}, } - err = s2.ApplyChangeSets(s2.Version()+1, []*proto.NamedChangeSet{cs2}) - require.Error(t, err, "should fail on corrupted AccountValue") - require.Contains(t, err.Error(), "unsupported serialization version") + require.NoError(t, s2.ApplyChangeSets(s2.Version()+1, []*proto.NamedChangeSet{cs2}), + "the fold is scheduled, not performed, so applying the block cannot meet the corruption") + + // The read is what meets it. It waits for the fold rather than racing it, so this is not timing + // dependent: the account either folds before the read arrives or the read waits for it, and both + // end at the same failure. Reads report a corrupted row by panicking (see CommitStore.Get). + nonceKey := keys.BuildEVMKey(keys.EVMKeyNonce, addr[:]) + cause := func() (recovered string) { + defer func() { + if r := recover(); r != nil { + recovered = fmt.Sprint(r) + } + }() + s2.Get("evm", nonceKey) + return "" + }() + require.Contains(t, cause, "unsupported serialization version", + "reading the folded account must report the corrupted row it was folded onto") } func TestCrashRecoveryCrashAfterWALBeforeDBCommit(t *testing.T) {