diff --git a/posting/mvcc.go b/posting/mvcc.go index aaea3f55626..8f6449989c2 100644 --- a/posting/mvcc.go +++ b/posting/mvcc.go @@ -598,6 +598,18 @@ func (ml *MemoryLayer) updateItemInCache(key string, delta []byte, startTs, comm val.lastUpdate = commitTs if val.list != nil { + // A commit at or below minTs is already in the rolled-up version this list was read from + // (rollups are versioned newest commit + 1); it gets here when the Raft log is replayed + // after a restart. Applying it again would put a delta below minTs into the list and + // break every read it serves, so drop the entry and let the next read rebuild it. + val.list.RLock() + minTs := val.list.minTs + val.list.RUnlock() + if commitTs <= minTs { + ml.del([]byte(key)) + return + } + p := new(pb.PostingList) x.Check(proto.Unmarshal(delta, p)) @@ -706,6 +718,7 @@ func ReadPostingList(key []byte, it *badger.Iterator) (*List, error) { switch item.UserMeta() { case BitEmptyPosting: + l.minTs = item.Version() return l, nil case BitCompletePosting: if err := unmarshalOrCopy(l.plist, item); err != nil { diff --git a/posting/mvcc_test.go b/posting/mvcc_test.go index 1fab24d9520..6ab700ee6b5 100644 --- a/posting/mvcc_test.go +++ b/posting/mvcc_test.go @@ -18,6 +18,7 @@ import ( "google.golang.org/protobuf/proto" "github.com/dgraph-io/badger/v4" + bpb "github.com/dgraph-io/badger/v4/pb" "github.com/dgraph-io/dgraph/v25/codec" "github.com/dgraph-io/dgraph/v25/protos/pb" "github.com/dgraph-io/dgraph/v25/schema" @@ -643,3 +644,117 @@ func TestPostingListRead(t *testing.T) { addEdgeToUID(t, attr, 1, 7, 15, 16) assertLength(17, 3) } + +// uidsAt reads the uid list of key the way a query does. +func uidsAt(t *testing.T, key []byte, readTs uint64) []uint64 { + t.Helper() + pl, err := NoCache(readTs).GetUids(key) + require.NoError(t, err) + uids, err := pl.Uids(ListOptions{ReadTs: readTs}) + require.NoError(t, err) + return uids.Uids +} + +// rollUpKey rolls key up at rollupTs, writes the result to pstore and returns it. +func rollUpKey(t *testing.T, key []byte, rollupTs uint64) *bpb.KV { + t.Helper() + l, err := GetNoStore(key, rollupTs) + require.NoError(t, err) + kvs, err := l.Rollup(nil, rollupTs) + require.NoError(t, err) + require.Len(t, kvs, 1) + RemoveCacheFor(key) + writer := NewTxnWriter(pstore) + require.NoError(t, writer.Write(&bpb.KVList{Kv: kvs})) + require.NoError(t, writer.Flush()) + return kvs[0] +} + +// addDeleteAllToUID commits * . through Txn.UpdateCachedKeys, like a real commit. +func addDeleteAllToUID(t *testing.T, attr string, src uint64, startTs, commitTs uint64) { + t.Helper() + edge := &pb.DirectedEdge{ + Attr: attr, + Entity: src, + Value: []byte(x.Star), + Op: pb.DirectedEdge_DEL, + } + l, err := GetNoStore(x.DataKey(attr, src), startTs) + require.NoError(t, err) + addMutation(t, l, edge, Del, startTs, commitTs, false) +} + +// A read at snapshot S must not see a delete committed at C > S after rollup has rewritten the +// emptied key as a BitEmptyPosting at C+1. That needs ReadPostingList to set minTs for the marker. +func TestReadAtSnapshotAfterEmptyRollup(t *testing.T) { + useMemoryLayer(t, 10<<20) // TestMain runs without the posting list cache. + require.NoError(t, pstore.DropAll()) + attr := x.AttrInRootNamespace("emptyRollupSnapshot") + key := x.DataKey(attr, 1) + + const ( + addTs = 2 + snapshotTs = 3 + delTs = 5 + rollupTs = 10 + ) + + addEdgeToUID(t, attr, 1, 2, addTs-1, addTs) + addDelEdgeToUID(t, attr, 1, 2, delTs-1, delTs) + require.Equal(t, []uint64{2}, uidsAt(t, key, snapshotTs)) + + kv := rollUpKey(t, key, rollupTs) + require.Equal(t, []byte{BitEmptyPosting}, kv.UserMeta) + require.Equal(t, uint64(delTs+1), kv.Version) + + require.Equal(t, []uint64{2}, uidsAt(t, key, snapshotTs)) + require.Empty(t, uidsAt(t, key, delTs)) +} + +// After a restart the Raft log replays commits that are already part of the rolled-up version. +// Folding such a commit into the cached list puts a delta below minTs into it, and a delete-all +// there makes every read of the entry fail in pIterator.seek. updateItemInCache must drop the entry. +func TestReplayedCommitBelowMinTsKeepsCachedListReadable(t *testing.T) { + const ( + addTs = 2 + delTs = 5 + readdTs = 7 + rollupTs = 10 + ) + + for _, tc := range []struct { + name string + readd bool + wantMeta byte + wantVer uint64 + want []uint64 + }{ + {name: "empty", wantMeta: BitEmptyPosting, wantVer: delTs + 1, want: nil}, + {name: "complete", readd: true, wantMeta: BitCompletePosting, wantVer: readdTs + 1, want: []uint64{3}}, + } { + t.Run(tc.name, func(t *testing.T) { + useMemoryLayer(t, 10<<20) + require.NoError(t, pstore.DropAll()) + attr := x.AttrInRootNamespace("replayBelowMinTs" + tc.name) + key := x.DataKey(attr, 1) + + addEdgeToUID(t, attr, 1, 2, addTs-1, addTs) + addDeleteAllToUID(t, attr, 1, delTs-1, delTs) + if tc.readd { + addEdgeToUID(t, attr, 1, 3, readdTs-1, readdTs) + } + + kv := rollUpKey(t, key, rollupTs) + require.Equal(t, []byte{tc.wantMeta}, kv.UserMeta) + require.Equal(t, tc.wantVer, kv.Version) + + // A cache entry serves minTs <= readTs <= maxTs; the rollup's version is both. + readTs := kv.Version + require.ElementsMatch(t, tc.want, uidsAt(t, key, readTs)) + + // Replay the delete-all with its original start and commit ts. + addDeleteAllToUID(t, attr, 1, delTs-1, delTs) + require.ElementsMatch(t, tc.want, uidsAt(t, key, readTs)) + }) + } +} diff --git a/systest/integration2/rollup_empty_list_test.go b/systest/integration2/rollup_empty_list_test.go new file mode 100644 index 00000000000..08209b0a403 --- /dev/null +++ b/systest/integration2/rollup_empty_list_test.go @@ -0,0 +1,206 @@ +//go:build integration2 + +/* + * SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package main + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/dgraph-io/dgo/v250/protos/api" + "github.com/dgraph-io/dgraph/v25/dgraphapi" + "github.com/dgraph-io/dgraph/v25/dgraphtest" +) + +// Once incremental rollup has rewritten a list that a delete emptied, the empty marker must not be +// served to transactions that started before the delete, and must carry its real version in an +// incremental backup. +// +// Rollup skips a key it rolled up less than 10s ago, so the tests wait emptyRollupCooldown between +// the setup mutation and the delete, then give the delete's rollup emptyRollupWait to land. + +const ( + emptyRollupCooldown = 11 * time.Second + emptyRollupWait = 20 * time.Second + emptyRollupNodes = 10 +) + +func startEmptyRollupCluster(t *testing.T) (*dgraphtest.LocalCluster, *dgraphapi.GrpcClient, + *dgraphapi.HTTPClient) { + t.Helper() + conf := dgraphtest.NewClusterConfig().WithNumAlphas(1).WithNumZeros(1).WithReplicas(1) + c, err := dgraphtest.NewLocalCluster(conf) + require.NoError(t, err) + t.Cleanup(func() { c.Cleanup(t.Failed()) }) + require.NoError(t, c.Start()) + + gc, cleanup, err := c.Client() + require.NoError(t, err) + t.Cleanup(cleanup) + hc, err := c.HTTPClient() + require.NoError(t, err) + + require.NoError(t, gc.SetupSchema(` + friend: [uid] . + status: string .`)) + return c, gc, hc +} + +// addNodesWithOneFriend creates n nodes with one friend edge and status "before" each. +func addNodesWithOneFriend(t *testing.T, gc *dgraphapi.GrpcClient, n int) []string { + t.Helper() + var set strings.Builder + for i := 0; i < n; i++ { + fmt.Fprintf(&set, "_:x%d _:y%d .\n_:x%d \"before\" .\n", i, i, i) + } + resp, err := gc.Mutate(&api.Mutation{SetNquads: []byte(set.String()), CommitNow: true}) + require.NoError(t, err) + uids := make([]string, n) + for i := range uids { + uids[i] = resp.Uids[fmt.Sprintf("x%d", i)] + require.NotEmpty(t, uids[i]) + } + return uids +} + +// deleteAllFriends empties every node's friend list and returns the commit ts. +func deleteAllFriends(t *testing.T, gc *dgraphapi.GrpcClient, uids []string) uint64 { + t.Helper() + var del, set strings.Builder + for _, uid := range uids { + fmt.Fprintf(&del, "<%s> * .\n", uid) + fmt.Fprintf(&set, "<%s> \"after\" .\n", uid) + } + resp, err := gc.Mutate(&api.Mutation{ + DelNquads: []byte(del.String()), + SetNquads: []byte(set.String()), + CommitNow: true, + }) + require.NoError(t, err) + require.NotZero(t, resp.Txn.CommitTs) + return resp.Txn.CommitTs +} + +func countNodesWithFriend(t *testing.T, gc *dgraphapi.GrpcClient) int { + t.Helper() + resp, err := gc.Query(`{ q(func: has(friend)) { count(uid) } }`) + require.NoError(t, err) + var out struct { + Q []struct{ Count int } `json:"q"` + } + require.NoError(t, json.Unmarshal(resp.Json, &out)) + require.Len(t, out.Q, 1) + return out.Q[0].Count +} + +// A transaction that started before a delete must keep seeing the deleted edge at its own start +// ts, also after the alpha has rolled the emptied list up. +func TestReadAtEarlierStartTsAfterDeleteEmptiesList(t *testing.T) { + _, gc, _ := startEmptyRollupCluster(t) + uids := addNodesWithOneFriend(t, gc, emptyRollupNodes) + require.Equal(t, emptyRollupNodes, countNodesWithFriend(t, gc)) + + time.Sleep(emptyRollupCooldown) + + // Transaction A: its first query fixes its start ts. + ctx := context.Background() + txnA := gc.NewReadOnlyTxn() + defer func() { _ = txnA.Discard(ctx) }() + resp, err := txnA.Query(ctx, `{ q(func: uid(0x1)) { uid } }`) + require.NoError(t, err) + startTs := resp.Txn.StartTs + require.NotZero(t, startTs) + + commitTs := deleteAllFriends(t, gc, uids) + require.Greater(t, commitTs, startTs) + require.Equal(t, 0, countNodesWithFriend(t, gc)) + + // A must see every friend at its start ts, now and after the rollup rewrites the keys. + uidList := strings.Join(uids, ", ") + query := fmt.Sprintf(`{ + q(func: uid(%s)) { uid status cnt: count(friend) friend { uid } } + h(func: uid(%s)) @filter(has(friend)) { uid } + }`, uidList, uidList) + deadline := time.Now().Add(emptyRollupWait) + for round := 0; time.Now().Before(deadline); round++ { + resp, err := txnA.Query(ctx, query) + require.NoError(t, err) + require.Equal(t, startTs, resp.Txn.StartTs) + + var out struct { + Q []struct { + Uid string `json:"uid"` + Status string `json:"status"` + Cnt int `json:"cnt"` + Friend []struct { + Uid string `json:"uid"` + } `json:"friend"` + } `json:"q"` + H []struct { + Uid string `json:"uid"` + } `json:"h"` + } + require.NoError(t, json.Unmarshal(resp.Json, &out)) + require.Len(t, out.Q, emptyRollupNodes) + + missing := 0 + for _, n := range out.Q { + require.Equal(t, "before", n.Status, + "round %d: scalar read at start ts %d (delete committed at %d)", round, startTs, commitTs) + if len(n.Friend) != 1 || n.Cnt != 1 { + missing++ + } + } + require.Zero(t, missing, "round %d: %d of %d nodes lost their friend at start ts %d, "+ + "which precedes the delete committed at %d; response: %s", + round, missing, emptyRollupNodes, startTs, commitTs, resp.Json) + require.Len(t, out.H, emptyRollupNodes, + "round %d: has(friend) at start ts %d dropped nodes after the rollup", round, startTs) + + time.Sleep(500 * time.Millisecond) + } +} + +// A full backup taken while the edges exist plus an incremental backup taken after the delete and +// its rollup must restore to zero edges. +func TestRestoreAfterDeleteEmptiesListAndRollup(t *testing.T) { + c, gc, hc := startEmptyRollupCluster(t) + uids := addNodesWithOneFriend(t, gc, emptyRollupNodes) + require.Equal(t, emptyRollupNodes, countNodesWithFriend(t, gc)) + + require.NoError(t, hc.Backup(c, true, dgraphtest.DefaultBackupDir)) + + time.Sleep(emptyRollupCooldown) + deleteAllFriends(t, gc, uids) + require.Equal(t, 0, countNodesWithFriend(t, gc)) + + // Give the rollup time to rewrite the emptied keys. + deadline := time.Now().Add(emptyRollupWait / 2) + for time.Now().Before(deadline) { + require.Equal(t, 0, countNodesWithFriend(t, gc)) + time.Sleep(500 * time.Millisecond) + } + + require.NoError(t, hc.Backup(c, false, dgraphtest.DefaultBackupDir)) + + // An edge in neither backup; a correct restore drops it. + _, err := gc.Mutate(&api.Mutation{SetNquads: []byte(`_:s _:t .`), CommitNow: true}) + require.NoError(t, err) + require.Equal(t, 1, countNodesWithFriend(t, gc)) + + require.NoError(t, hc.Restore(c, dgraphtest.DefaultBackupDir, "", 0, 0)) + require.NoError(t, dgraphapi.WaitForRestore(c)) + + require.Equal(t, 0, countNodesWithFriend(t, gc), + "friend edges deleted and rolled up before the incremental backup came back after restore") +}