first draft of surfacing the reason transaction abort/error occured

Author: rahst12

dgraph/cmd/alpha/txn_test.go

@@ -21,6 +21,11 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 
 	"github.com/dgraph-io/dgo/v250"
 	"github.com/dgraph-io/dgo/v250/protos/api"
@@ -248,6 +253,64 @@ func TestConflict(t *testing.T) {
 	require.True(t, bytes.Equal(resp.Json, []byte("{\"me\":[{\"name\":\"Manish\"}]}")))
 }
 
+// TestConflictAbortReason proves the server emits the categorized abort reason on the
+// gRPC status of a write-write conflict: the code stays codes.Aborted (so existing
+// clients keep retrying) and the message is prefixed with "conflict: ". This is exactly
+// the unflattened status a gRPC client such as dgraph4j receives and parses into
+// TxnConflictException.AbortReason.
+//
+// It must inspect the raw CommitOrAbort response rather than dgo's high-level
+// Txn.Commit, because dgo intentionally replaces any codes.Aborted error with the
+// static dgo.ErrAborted (txn.go), discarding the reason for the Go client.
+func TestConflictAbortReason(t *testing.T) {
+	op := &api.Operation{}
+	op.DropAll = true
+	require.NoError(t, dg.Alter(context.Background(), op))
+
+	// First transaction creates a node with a name.
+	txn := dg.NewTxn()
+	mu := &api.Mutation{}
+	mu.SetJson = []byte(`{"name": "Manish"}`)
+	assigned, err := txn.Mutate(context.Background(), mu)
+	require.NoError(t, err)
+	require.Len(t, assigned.Uids, 1)
+	var uid string
+	for _, u := range assigned.Uids {
+		uid = u
+	}
+
+	// Second transaction writes the same predicate on the same uid -> conflicts.
+	txn2 := dg.NewTxn()
+	mu = &api.Mutation{}
+	mu.SetJson = []byte(fmt.Sprintf(`{"uid": %q, "name": "Manish"}`, uid))
+	resp2, err := txn2.Mutate(context.Background(), mu)
+	require.NoError(t, err)
+	require.NotEmpty(t, resp2.GetTxn().GetKeys(), "mutation response must carry conflict keys")
+
+	// First commit wins; its commitTs is now greater than txn2's startTs.
+	require.NoError(t, txn.Commit(context.Background()))
+
+	// Commit the loser via the raw gRPC stub so we observe the unflattened status the
+	// server sends (dgo's Txn.Commit would replace it with the reasonless ErrAborted).
+	conn, err := grpc.NewClient(alphaSockAdd, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	defer func() { _ = conn.Close() }()
+	raw := api.NewDgraphClient(conn)
+
+	// TestMain enables ACL, so the raw stub (unlike the logged-in dg client) must
+	// carry the access JWT; CommitOrAbort goes through the same auth interceptor.
+	ctx := metadata.NewOutgoingContext(context.Background(),
+		metadata.Pairs("accessJwt", hc.AccessJwt))
+	_, err = raw.CommitOrAbort(ctx, resp2.GetTxn())
+	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(), "conflict: "),
+		"abort reason should be categorized as conflict; got %q", st.Message())
+}
+
 func TestConflictTimeout(t *testing.T) {
 	var uid string
 	txn := dg.NewTxn()

dgraph/cmd/zero/oracle.go

@@ -18,6 +18,8 @@ import (
 	"go.opentelemetry.io/otel"
 	attribute "go.opentelemetry.io/otel/attribute"
 	trace "go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/dgraph-io/badger/v4/y"
 	"github.com/dgraph-io/dgo/v250/protos/api"
@@ -338,21 +340,56 @@ func (s *Server) proposeTxn(ctx context.Context, src *api.TxnContext) error {
 	return nil
 }
 
+// Abort-reason codes. When Zero decides to abort a transaction it surfaces the category to the
+// client as the prefix of a codes.Aborted gRPC status message, formatted as "<code>: <detail>".
+// api.TxnContext (external dgo module) has no field for the reason, so it rides on the error;
+// the status code stays codes.Aborted, so existing abort handling is unaffected. The dgraph4j
+// client parses these prefixes into TxnConflictException.AbortReason — keep the two in sync.
+const (
+	abortReasonConflict      = "conflict"
+	abortReasonStaleStartTs  = "stale-startts"
+	abortReasonPredicateMove = "predicate-move"
+)
+
+// abortReason builds the wire string the client parses: "<code>: <detail>".
+func abortReason(code, detail string) string {
+	return code + ": " + detail
+}
+
 func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	span := trace.SpanFromContext(ctx)
 	span.SetAttributes(attribute.Int64("startTs", int64(src.StartTs)))
 	if src.Aborted {
+		// Client-initiated discard (txn.Discard); not a server-decided abort, so no reason.
 		return s.proposeTxn(ctx, src)
 	}
 
+	// abortWithReason marks the txn aborted, proposes it (advancing the watermark exactly as the
+	// previous code did), and then surfaces the categorized reason to the caller as a gRPC
+	// Aborted status. proposeTxn still runs identically — only the post-propose return changes.
+	abortWithReason := func(reason string) error {
+		span.SetAttributes(attribute.Bool("abort", true))
+		src.Aborted = true
+		if err := s.proposeTxn(ctx, src); err != nil {
+			return err
+		}
+		return status.Error(codes.Aborted, reason)
+	}
+
 	// Use the start timestamp to check if we have a conflict, before we need to assign a commit ts.
 	s.orc.RLock()
 	conflict := s.orc.hasConflict(src)
+	// A txn whose startTs predates this leader's lease is aborted by hasConflict, but it's a
+	// leader change rather than a write-write conflict — report it distinctly.
+	stale := src.StartTs < s.orc.startTxnTs
 	s.orc.RUnlock()
 	if conflict {
-		span.SetAttributes(attribute.Bool("abort", true))
-		src.Aborted = true
-		return s.proposeTxn(ctx, src)
+		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"))
 	}
 
 	checkPreds := func() error {
@@ -385,9 +422,9 @@ func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 		return nil
 	}
 	if err := checkPreds(); err != nil {
-		span.SetAttributes(attribute.Bool("abort", true))
-		src.Aborted = true
-		return s.proposeTxn(ctx, src)
+		// checkPreds builds rich messages, e.g. "Commits on predicate %s are blocked due to
+		// predicate move" — forward them instead of swallowing the reason.
+		return abortWithReason(abortReason(abortReasonPredicateMove, err.Error()))
 	}
 
 	num := pb.Num{Val: 1, Type: pb.Num_TXN_TS}
@@ -402,16 +439,27 @@ func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	span.SetAttributes(attribute.Int64("nodeId", int64(s.Node.Id)))
 	span.AddEvent(fmt.Sprintf("TXN Context: %+v", src))
 
+	aborted := false
 	if err := s.orc.commit(src); err != nil {
 		span.SetAttributes(attribute.Bool("abort", true))
 		src.Aborted = true
+		aborted = true
 	}
 	if err := ctx.Err(); err != nil {
 		span.SetAttributes(attribute.Bool("abort", true))
 		src.Aborted = true
+		aborted = true
 	}
 	// Propose txn should be used to set watermark as done.
-	return s.proposeTxn(ctx, src)
+	if err := s.proposeTxn(ctx, src); err != nil {
+		return err
+	}
+	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 nil
 }
 
 // CommitOrAbort either commits a transaction or aborts it.

edgraph/server.go

@@ -646,6 +646,10 @@ func (s *Server) doMutate(ctx context.Context, qc *queryContext, resp *api.Respo
 		if err == dgo.ErrAborted {
 			err = status.Error(codes.Aborted, err.Error())
 			resp.Txn.Aborted = true
+		} else if status.Code(err) == codes.Aborted {
+			// Server-decided abort carrying a categorized reason; err is already a codes.Aborted
+			// status (e.g. "conflict: ...", "predicate-move: ...") — surface it unchanged.
+			resp.Txn.Aborted = true
 		}
 
 		return err
@@ -2085,6 +2089,12 @@ func (s *Server) CommitOrAbort(ctx context.Context, tc *api.TxnContext) (*api.Tx
 
 		return tctx, status.Error(codes.Aborted, err.Error())
 	}
+	if status.Code(err) == codes.Aborted {
+		// Server-decided abort carrying a categorized reason; err is already a codes.Aborted
+		// status (e.g. "conflict: ...", "predicate-move: ...") — surface it unchanged.
+		tctx.Aborted = true
+		return tctx, err
+	}
 	tctx.StartTs = tc.StartTs
 	tctx.CommitTs = commitTs
 	return tctx, err

worker/mutation.go

@@ -20,7 +20,9 @@ import (
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/dgraph-io/badger/v4"
@@ -908,6 +910,16 @@ func CommitOverNetwork(ctx context.Context, tc *api.TxnContext) (uint64, error)
 	tctx, err := zc.CommitOrAbort(ctx, tc)
 
 	if err != nil {
+		// Zero signals a server-decided abort as a codes.Aborted status carrying a categorized
+		// reason (e.g. "conflict: ...", "predicate-move: ..."). Record the abort metric and
+		// forward the reason rather than treating it as a generic transport error or flattening
+		// it to the reasonless dgo.ErrAborted.
+		if status.Code(err) == codes.Aborted {
+			if !clientDiscard {
+				ostats.Record(ctx, x.TxnAborts.M(1))
+			}
+			return 0, err
+		}
 		span.AddEvent("Error in CommitOrAbort", trace.WithAttributes(
 			attribute.String("error", err.Error())))
 		return 0, err