fix(pipeline): scalar Del-of-old-value must not wipe new Set in ProcessCount

Bug: scalar @count writes were nondeterministically losing data under
concurrent transactions. Roughly half the deltas committed by the
new mutation pipeline contained only a [DeleteAll] posting and no
Set, so reads at maxTs returned an empty value list.

Root cause: in ProcessSingle, handleOldDeleteForSingle appends a
synthetic Del-of-old-value to postings[uid] alongside the user's
Set, so InsertTokenizerIndexes / ProcessReverse / count diffing can
see the prior value. ProcessCount then iterates the postings and
calls list.updateMutationLayer(post, singleUidUpdate=true, ...) on
each. For non-Lang scalar predicates fingerprintEdge returns
math.MaxUint64, so the synthetic Del and the user Set both have
Uid == math.MaxUint64. The first iteration (Set new) leaves
mutationMap.currentEntries = [DeleteAll, Set new]; the second
iteration (Del old) finds the Set we just inserted via findPosting
and applies updateMutationLayer in singleUidUpdate mode, which
unconditionally rewrites currentEntries to [DeleteAll] (the Del
branch never appends mpost) — wiping the new value.

Fix: in ProcessCount, when iterating a !isListEdge predicate's
postings, if the list contains a Set/Ovr posting, treat any Del as
synthetic and skip it for the data-list update. Standalone user Dels
(no accompanying Set) are still applied. Index/reverse/count diffing
already happen before ProcessCount runs and aren't affected.

Repro: TestPipelineCountIndexConcurrent in worker/sort_test.go is a
new conflict-aware in-process harness that mirrors the systest
TestCountIndexConcurrentSetDelScalarPredicate. It runs 200
contending transactions setting <0x1> <name> "name<rand>" against a
"string @index(exact) @count" schema with a fakeOracle that
implements the same hasConflict algorithm as
dgraph/cmd/zero/oracle.go. Pre-fix the test fails roughly 50% of
runs with an empty data list and the wrong count buckets; post-fix
it is stable across 20+ -count iterations and under -race.

Existing tests (TestScalarPredicateIntCount, *RevCount, *Count,
TestSingleUidReplacement, TestDeleteSetWithVarEdgeCorruptsData,
TestStringIndexWithLang, TestMultipleTxnListCount, TestGetScalarList,
TestDatetime) all pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <[email protected]>

Author: matthewmcneely

posting/index.go

@@ -540,7 +540,31 @@ func (mp *MutationPipeline) ProcessCount(ctx context.Context, pipeline *Predicat
 		list.Lock()
 		prevCount := list.GetLength(mp.txn.StartTs)
 
+		// For scalar (non-list) predicates, handleOldDeleteForSingle may have
+		// appended a synthetic Del-of-old-value alongside the user's Set, so
+		// that InsertTokenizerIndexes / ProcessReverse / count diffing can see
+		// the prior value. The synthetic Del must NOT be applied to the data
+		// list: scalar value postings all share Uid == math.MaxUint64
+		// (fingerprintEdge returns MaxUint64 for non-Lang scalars), and
+		// updateMutationLayer in singleUidUpdate mode would overwrite the
+		// just-inserted Set with [DeleteAll] and drop the new value entirely.
+		// A user-initiated Del (no accompanying Set) must still be applied.
+		skipSyntheticDel := false
+		if !isListEdge {
+			hasSet := false
+			for _, post := range postingList.Postings {
+				if post.Op == Set || post.Op == Ovr {
+					hasSet = true
+					break
+				}
+			}
+			skipSyntheticDel = hasSet
+		}
+
 		for _, post := range postingList.Postings {
+			if skipSyntheticDel && post.Op == Del {
+				continue
+			}
 			found, _, _ := list.findPosting(post.StartTs, post.Uid)
 			if found {
 				if post.Op == Set && isListEdge {

worker/sort_test.go

@@ -11,11 +11,14 @@ import (
 	"math"
 	"math/rand"
 	"os"
+	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/dgraph-io/badger/v4"
 	bpb "github.com/dgraph-io/badger/v4/pb"
+	"github.com/dgraph-io/dgo/v250/protos/api"
 	"github.com/dgraph-io/dgraph/v25/posting"
 	"github.com/dgraph-io/dgraph/v25/protos/pb"
 	"github.com/dgraph-io/dgraph/v25/schema"
@@ -409,6 +412,179 @@ func TestCount(t *testing.T) {
 	require.Equal(t, subjects, len(uids.Uids))
 }
 
+// fakeOracle is an in-memory stand-in for the zero Oracle. It hands out
+// monotonically increasing timestamps and rejects commits whose conflict
+// keys overlap a higher commitTs — same algorithm as
+// dgraph/cmd/zero/oracle.go's hasConflict.
+type fakeOracle struct {
+	mu        sync.Mutex
+	nextTs    uint64
+	keyCommit map[uint64]uint64 // conflict-key fingerprint -> commitTs
+	committed atomic.Int64
+	aborted   atomic.Int64
+}
+
+func newFakeOracle(initialTs uint64) *fakeOracle {
+	return &fakeOracle{nextTs: initialTs, keyCommit: map[uint64]uint64{}}
+}
+
+func (o *fakeOracle) reserveStartTs() uint64 {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	o.nextTs++
+	return o.nextTs
+}
+
+// tryCommit mirrors zero/oracle.go: for each conflict key, if a later
+// commitTs already exists, abort. Else stamp all keys with a fresh
+// commitTs and return it.
+func (o *fakeOracle) tryCommit(startTs uint64, conflictKeys []uint64) (uint64, bool) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	for _, k := range conflictKeys {
+		if last, ok := o.keyCommit[k]; ok && last > startTs {
+			o.aborted.Add(1)
+			return 0, false
+		}
+	}
+	o.nextTs++
+	commitTs := o.nextTs
+	for _, k := range conflictKeys {
+		o.keyCommit[k] = commitTs
+	}
+	o.committed.Add(1)
+	return commitTs, true
+}
+
+// runPipelineTxn drives a single mutation through the new pipeline with
+// real conflict-aware commit semantics. Returns (committed, error).
+func runPipelineTxn(t *testing.T, ps *badger.DB, oracle *fakeOracle,
+	edges []*pb.DirectedEdge) bool {
+	t.Helper()
+	startTs := oracle.reserveStartTs()
+	txn := posting.Oracle().RegisterStartTs(startTs)
+
+	if err := newRunMutations(context.Background(), edges, txn); err != nil {
+		t.Fatalf("pipeline failed at startTs=%d: %v", startTs, err)
+	}
+
+	// FillContext bridges plists -> deltas (via Update) and emits the
+	// txn's conflict keys as base-36 strings on ctx.Keys.
+	ctxApi := &api.TxnContext{}
+	txn.FillContext(ctxApi, 1, false)
+
+	keys := make([]uint64, 0, len(ctxApi.Keys))
+	for _, k := range ctxApi.Keys {
+		ki, err := strconv.ParseUint(k, 36, 64)
+		require.NoError(t, err)
+		keys = append(keys, ki)
+	}
+
+	commitTs, ok := oracle.tryCommit(startTs, keys)
+	if !ok {
+		return false
+	}
+	writer := posting.NewTxnWriter(ps)
+	require.NoError(t, txn.CommitToDisk(writer, commitTs))
+	require.NoError(t, writer.Flush())
+	txn.UpdateCachedKeys(commitTs)
+	return true
+}
+
+// TestPipelineCountIndexConcurrent mirrors the systest's
+// TestCountIndexConcurrentSetDelScalarPredicate at unit-test scope: many
+// concurrent transactions setting <0x1> <name> "name<rand>" against a
+// scalar string predicate with @index(exact) @count, with real
+// conflict-checking commit semantics. After everything settles, the data
+// list for 0x1 should hold exactly one value, the count(1) bucket should
+// reference exactly 0x1, and no other count bucket should reference 0x1.
+func TestPipelineCountIndexConcurrent(t *testing.T) {
+	dir, err := os.MkdirTemp("", "storetest_")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+
+	ps, err := badger.OpenManaged(badger.DefaultOptions(dir))
+	require.NoError(t, err)
+	defer ps.Close()
+	posting.Init(ps, 0, false)
+	Init(ps)
+	posting.Oracle().ResetTxns()
+
+	require.NoError(t, schema.ParseBytes(
+		[]byte(`name: string @index(exact) @count .`), 1))
+
+	pred := x.AttrInRootNamespace("name")
+	const target uint64 = 1
+
+	oracle := newFakeOracle(10)
+
+	const (
+		numRoutines  = 10
+		txnsPerRoute = 20
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(numRoutines)
+	for r := 0; r < numRoutines; r++ {
+		go func(seed int) {
+			defer wg.Done()
+			rnd := rand.New(rand.NewSource(int64(seed)))
+			for i := 0; i < txnsPerRoute; i++ {
+				value := []byte(fmt.Sprintf("name%d", rnd.Intn(10000)))
+				// Retry on conflict — same as a client doing dg.NewTxn().Mutate().
+				// Each attempt uses a fresh edge: makePostingFromEdge mutates
+				// edge.ValueId during processing, and reusing the object across
+				// attempts would make ValidateAndConvert see it as a uid edge.
+				// Real production gets a freshly-deserialized edge per Raft apply.
+				for attempt := 0; attempt < 100; attempt++ {
+					edge := &pb.DirectedEdge{
+						Entity:    target,
+						Attr:      pred,
+						Value:     value,
+						ValueType: pb.Posting_STRING,
+						Op:        pb.DirectedEdge_SET,
+					}
+					if runPipelineTxn(t, ps, oracle, []*pb.DirectedEdge{edge}) {
+						break
+					}
+				}
+			}
+		}(r)
+	}
+	wg.Wait()
+
+	t.Logf("committed=%d aborted=%d", oracle.committed.Load(), oracle.aborted.Load())
+
+	// Verify final state: exactly one value on 0x1, that uid in count(1) only.
+	readTxn := posting.Oracle().RegisterStartTs(math.MaxUint64)
+
+	dataKey := x.DataKey(pred, target)
+	dpl, err := readTxn.Get(dataKey)
+	require.NoError(t, err)
+	// Scalar string predicate: AllValues returns the live posting list values
+	// (one entry for a non-list scalar with a current value).
+	vals, err := dpl.AllValues(math.MaxUint64)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(vals),
+		"scalar predicate should retain exactly one value, got %v", vals)
+
+	for c := 0; c <= 5; c++ {
+		ck := x.CountKey(pred, uint32(c), false)
+		cpl, err := readTxn.Get(ck)
+		require.NoError(t, err)
+		cuids, err := cpl.Uids(posting.ListOptions{ReadTs: math.MaxUint64})
+		require.NoError(t, err)
+		switch c {
+		case 1:
+			require.Equal(t, []uint64{target}, cuids.Uids,
+				"count(1) bucket must contain exactly the target uid")
+		default:
+			require.NotContains(t, cuids.Uids, target,
+				"count(%d) bucket must not contain the target uid", c)
+		}
+	}
+}
+
 func TestDeleteSetWithVarEdgeCorruptsData(t *testing.T) {
 	// Setup temporary directory for Badger DB
 	dir, err := os.MkdirTemp("", "storetest_")