FlatKV: background read/fold for account data

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
+// "<none>", 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("<none>"), 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")
+}

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

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

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