From 9c7d5fa605fe76673b10d42b221dcc9fdc285934 Mon Sep 17 00:00:00 2001 From: Shivaji Kharse Date: Mon, 19 Oct 2026 18:44:04 +0530 Subject: [PATCH] fix(posting): set minTs for empty rollups, drop replayed commits below minTs When a delete empties a posting list, the incremental rollup rewrites the key as a BitEmptyPosting marker at commitTs+1. ReadPostingList did not set List.minTs for that marker (it does for BitCompletePosting), so: - on a cache miss MemoryLayer.ReadData, which reads the key at math.MaxUint64, served the empty list to every read ts, including transactions that started before the delete; and - ToBackupPostingList versioned the emptied key at 0, so restoring a full plus an incremental backup kept the older, non-empty version. Set minTs = item.Version() for BitEmptyPosting as well. The line was dropped in ee99a7a85 (#9237) and the bug first shipped in v24.1.0. With minTs set, a commit replayed from the Raft log after a restart can reach updateItemInCache with commitTs <= minTs of the cached list, because the rollup that already contains it is versioned newest commit + 1. Folding it in again puts a delta below minTs into the mutable layer, and if it is a delete-all every read the entry serves fails with "deleteBelowTs must be greater than the minTs in the list" until the entry is evicted. Drop the cache entry in that case instead; the next read rebuilds it from disk. This path was already reachable for complete rollups before this change. Tests: posting/mvcc_test.go gains the reporter's snapshot-read test and a replay test for both rollup kinds; systest/integration2/rollup_empty_list_test.go reproduces the earlier-start-ts read and the backup/restore case end to end. Fixes #9856 --- posting/mvcc.go | 13 ++ posting/mvcc_test.go | 115 ++++++++++ .../integration2/rollup_empty_list_test.go | 206 ++++++++++++++++++ 3 files changed, 334 insertions(+) create mode 100644 systest/integration2/rollup_empty_list_test.go 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") +}