adding test

Author: rahst12

dgraph/cmd/alpha/txn_test.go

@@ -311,6 +311,59 @@ func TestConflictAbortReason(t *testing.T) {
 		"abort reason should be categorized as conflict; got %q", st.Message())
 }
 
+// TestPredicateMoveAbortReason proves the "predicate-move" category is surfaced on the
+// gRPC status. Zero's checkPreds aborts a commit whose predicate keys don't match the
+// tablet's serving group (the same code path that rejects commits during a predicate
+// move). We reach it deterministically by committing a real txn context whose Preds have
+// been rewritten to claim a group that doesn't serve the predicate, with conflict Keys
+// cleared so hasConflict passes and checkPreds runs.
+//
+// As with TestConflictAbortReason, it uses the raw CommitOrAbort stub to observe the
+// unflattened status the server sends.
+func TestPredicateMoveAbortReason(t *testing.T) {
+	op := &api.Operation{}
+	op.DropAll = true
+	require.NoError(t, dg.Alter(context.Background(), op))
+
+	// A normal mutation gives us a valid, fresh TxnContext (real StartTs and Preds).
+	txn := dg.NewTxn()
+	mu := &api.Mutation{SetJson: []byte(`{"name": "Alice"}`)}
+	resp, err := txn.Mutate(context.Background(), mu)
+	require.NoError(t, err)
+
+	tc := resp.GetTxn()
+	require.NotEmpty(t, tc.GetPreds(), "mutation response must carry predicate keys")
+
+	// Rewrite each predicate key's group id to a group that does not serve it, and drop
+	// the conflict keys so hasConflict is false and the commit reaches checkPreds.
+	doctored := make([]string, 0, len(tc.GetPreds()))
+	for _, p := range tc.GetPreds() {
+		// Preds look like "<gid>-<predicate>"; claim a nonexistent group 100.
+		if idx := strings.IndexByte(p, '-'); idx >= 0 {
+			doctored = append(doctored, "100"+p[idx:])
+		}
+	}
+	require.NotEmpty(t, doctored, "expected parseable predicate keys")
+	tc.Preds = doctored
+	tc.Keys = nil
+
+	conn, err := grpc.NewClient(alphaSockAdd, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer func() { _ = conn.Close() }()
+	raw := api.NewDgraphClient(conn)
+
+	ctx := metadata.NewOutgoingContext(context.Background(),
+		metadata.Pairs("accessJwt", hc.AccessJwt))
+	_, err = raw.CommitOrAbort(ctx, tc)
+	require.Error(t, err)
+
+	st := status.Convert(err)
+	require.Equal(t, codes.Aborted, st.Code(),
+		"abort must keep codes.Aborted so existing clients still retry")
+	require.True(t, strings.HasPrefix(st.Message(), "predicate-move: "),
+		"abort reason should be categorized as predicate-move; got %q", st.Message())
+}
+
 func TestConflictTimeout(t *testing.T) {
 	var uid string
 	txn := dg.NewTxn()

dgraph/cmd/zero/oracle.go

@@ -351,11 +351,27 @@ const (
 	abortReasonPredicateMove = "predicate-move"
 )
 
+// Human-readable details paired with the conflict abort codes.
+const (
+	abortDetailConflict     = "Transaction has been aborted. Please retry"
+	abortDetailStaleStartTs = "Transaction has been aborted due to a leader change. Please retry"
+)
+
 // abortReason builds the wire string the client parses: "<code>: <detail>".
 func abortReason(code, detail string) string {
 	return code + ": " + detail
 }
 
+// conflictAbortReason returns the wire reason for a hasConflict abort, distinguishing a
+// write-write conflict from a stale start timestamp (a txn that predates the current
+// leader's lease, i.e. a leader change rather than a real conflict).
+func conflictAbortReason(stale bool) string {
+	if stale {
+		return abortReason(abortReasonStaleStartTs, abortDetailStaleStartTs)
+	}
+	return abortReason(abortReasonConflict, abortDetailConflict)
+}
+
 func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	span := trace.SpanFromContext(ctx)
 	span.SetAttributes(attribute.Int64("startTs", int64(src.StartTs)))
@@ -384,12 +400,7 @@ func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	stale := src.StartTs < s.orc.startTxnTs
 	s.orc.RUnlock()
 	if conflict {
-		if stale {
-			return abortWithReason(abortReason(abortReasonStaleStartTs,
-				"Transaction has been aborted due to a leader change. Please retry"))
-		}
-		return abortWithReason(abortReason(abortReasonConflict,
-			"Transaction has been aborted. Please retry"))
+		return abortWithReason(conflictAbortReason(stale))
 	}
 
 	checkPreds := func() error {
@@ -456,8 +467,7 @@ func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	}
 	if aborted {
 		// A late write-write conflict detected at commit time (keyCommit), or a cancelled ctx.
-		return status.Error(codes.Aborted, abortReason(abortReasonConflict,
-			"Transaction has been aborted. Please retry"))
+		return status.Error(codes.Aborted, conflictAbortReason(false))
 	}
 	return nil
 }

dgraph/cmd/zero/oracle_reason_test.go

@@ -0,0 +1,63 @@
+/*
+ * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package zero
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/dgraph-io/dgo/v250/protos/api"
+)
+
+// The abort-reason wire format is a contract with gRPC clients (e.g. dgraph4j parses the
+// "<code>: " prefix into TxnConflictException.AbortReason). These unit tests pin the
+// category prefixes and the logic that selects between them, so the contract can't drift
+// silently without an integration cluster.
+
+func TestAbortReasonFormat(t *testing.T) {
+	require.Equal(t, "conflict: boom", abortReason(abortReasonConflict, "boom"))
+	require.Equal(t, "stale-startts: x", abortReason(abortReasonStaleStartTs, "x"))
+	require.Equal(t, "predicate-move: y", abortReason(abortReasonPredicateMove, "y"))
+}
+
+func TestConflictAbortReason(t *testing.T) {
+	// Write-write conflict.
+	r := conflictAbortReason(false)
+	require.True(t, strings.HasPrefix(r, abortReasonConflict+": "),
+		"want conflict prefix, got %q", r)
+	require.Equal(t, abortReason(abortReasonConflict, abortDetailConflict), r)
+
+	// Stale start timestamp (leader change).
+	r = conflictAbortReason(true)
+	require.True(t, strings.HasPrefix(r, abortReasonStaleStartTs+": "),
+		"want stale-startts prefix, got %q", r)
+	require.Equal(t, abortReason(abortReasonStaleStartTs, abortDetailStaleStartTs), r)
+	require.Contains(t, r, "leader change")
+}
+
+// TestHasConflictStaleStartTs pins the exact discriminator commit() uses to choose the
+// stale-startts reason: a txn whose startTs predates the leader's startTxnTs lease is a
+// conflict, and is flagged stale; a fresh startTs with no conflicting keys is neither.
+func TestHasConflictStaleStartTs(t *testing.T) {
+	o := &Oracle{}
+	o.Init()
+	defer o.close()
+
+	o.updateStartTxnTs(100)
+
+	// startTs below the lease floor: hasConflict true, and the stale discriminator true.
+	stale := &api.TxnContext{StartTs: 42}
+	require.True(t, o.hasConflict(stale), "txn below startTxnTs must conflict")
+	require.True(t, stale.StartTs < o.startTxnTs, "must be flagged stale")
+	require.Equal(t, conflictAbortReason(true), conflictAbortReason(stale.StartTs < o.startTxnTs))
+
+	// startTs at/above the lease floor with no keys: not a conflict, not stale.
+	fresh := &api.TxnContext{StartTs: 100}
+	require.False(t, o.hasConflict(fresh), "fresh txn with no keys must not conflict")
+	require.False(t, fresh.StartTs < o.startTxnTs, "must not be flagged stale")
+}

systest/integration2/txn_abort_reason_test.go

@@ -0,0 +1,86 @@
+//go:build integration2
+
+/*
+ * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/status"
+
+	"github.com/dgraph-io/dgo/v250/protos/api"
+	"github.com/dgraph-io/dgraph/v25/dgraphtest"
+)
+
+// TestStaleStartTsAbortReason proves the "stale-startts" abort category end-to-end against a
+// real cluster. A transaction's start timestamp becomes "stale" when it predates the current
+// Zero leader's lease — i.e. after a leader change. We reproduce that deterministically by
+// opening a transaction, then restarting the (single) Zero: on restart Zero renews its lease and
+// advances startTxnTs past every previously-leased start ts, so committing the now-old txn aborts
+// with the stale-startts reason rather than a plain write-write conflict.
+//
+// As in the alpha-level reason tests, the commit goes through the raw CommitOrAbort stub so we
+// observe the unflattened codes.Aborted status (dgo's Txn.Commit would replace it with the
+// reasonless ErrAborted).
+func TestStaleStartTsAbortReason(t *testing.T) {
+	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)
+	defer cleanup()
+
+	ctx := context.Background()
+	require.NoError(t, gc.Alter(ctx, &api.Operation{DropAll: true}))
+
+	// Open a transaction and mutate so it gets a real (soon-to-be-stale) start ts and keys.
+	txn := gc.NewTxn()
+	resp, err := txn.Mutate(ctx, &api.Mutation{SetJson: []byte(`{"name": "Manish"}`)})
+	require.NoError(t, err)
+	tc := resp.GetTxn()
+	require.NotZero(t, tc.GetStartTs(), "mutation must yield a start ts")
+
+	// Restart Zero. On coming back up it renews its lease and sets startTxnTs to MaxTxnTs+1,
+	// which is strictly greater than the start ts leased above — making our open txn stale.
+	require.NoError(t, c.StopZero(0))
+	require.NoError(t, c.StartZero(0))
+	require.NoError(t, c.HealthCheck(false))
+
+	// Wait until a Zero leader is established again (lease renewal, hence startTxnTs bump, runs
+	// when a Zero becomes leader). Avoids racing the commit against the leaderless window.
+	require.Eventually(t, func() bool {
+		_, err := c.GetZeroLeader(0)
+		return err == nil
+	}, 60*time.Second, time.Second, "zero leader did not re-establish after restart")
+
+	// Commit the stale txn via the raw stub to observe the categorized status.
+	port, err := c.GetAlphaGrpcPublicPort(0)
+	require.NoError(t, err)
+	conn, err := grpc.NewClient("0.0.0.0:"+port, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer func() { _ = conn.Close() }()
+	raw := api.NewDgraphClient(conn)
+
+	_, err = raw.CommitOrAbort(ctx, tc)
+	require.Error(t, err, "committing a txn whose start ts predates the new leader must abort")
+
+	st := status.Convert(err)
+	require.Equal(t, codes.Aborted, st.Code(),
+		"abort must keep codes.Aborted so existing clients still retry")
+	require.True(t, strings.HasPrefix(st.Message(), "stale-startts: "),
+		"abort reason should be categorized as stale-startts; got %q", st.Message())
+}