From b08e0fb11908163a8b699336b2ecaea2078b8e2d Mon Sep 17 00:00:00 2001 From: Matthew McNeely Date: Thu, 22 Oct 2026 12:14:25 -0400 Subject: [PATCH 1/3] feat(txn): surface the server's abort reason on AbortedError Dgraph now reports why it aborted a transaction as a ": " prefix on the gRPC Aborted status (dgraph-io/dgraph#9747). Txn flattened every Aborted status to the bare ErrAborted sentinel, so that reason never reached the caller. Txn.Commit and the mutation paths of Txn.Do/Mutate now return an *AbortedError carrying the parsed Reason and the server's message. It unwraps to ErrAborted, so errors.Is(err, dgo.ErrAborted) keeps working; a direct == comparison against ErrAborted no longer matches. Parsing matches dgraph4j#290 and pydgraph#307: the text before the first colon, trimmed and case-insensitive, mapped onto conflict, predicate-move or stale-startts, with anything else (including every message from an older server) reported as AbortReasonUnknown. Co-Authored-By: Claude Opus 5.5 --- README.md | 20 +++++++++++- abort.go | 88 ++++++++++++++++++++++++++++++++++++++++++++++++++ abort_test.go | 55 +++++++++++++++++++++++++++++++ errors_test.go | 12 ++++++- txn.go | 12 ++++--- 5 files changed, 180 insertions(+), 7 deletions(-) create mode 100644 abort.go create mode 100644 abort_test.go diff --git a/README.md b/README.md index 2d746dc..6b99135 100644 --- a/README.md +++ b/README.md @@ -611,11 +611,29 @@ txn := dgraphClient.NewTxn() // Perform some queries and mutations. err := txn.Commit(ctx) -if err == y.ErrAborted { +if errors.Is(err, dgo.ErrAborted) { // Retry or handle error } ``` +An aborted transaction returns a `*dgo.AbortedError`, which wraps `dgo.ErrAborted`. Test for it with +`errors.Is`; comparing with `==` does not match. Its `Reason` field says why the server aborted the +transaction, so you can choose how to retry: + +```go +var ae *dgo.AbortedError +if errors.As(err, &ae) { + switch ae.Reason { + case dgo.AbortReasonPredicateMove: + // A predicate is moving between groups. Back off, then retry with a new transaction. + case dgo.AbortReasonConflict, dgo.AbortReasonStaleStartTs: + // Retry with a new transaction. + default: // dgo.AbortReasonUnknown + // Older server, or a cause the server does not categorize. ae.Error() has the details. + } +} +``` + ### Setting Metadata Headers Metadata headers such as authentication tokens can be set through the context of gRPC methods. Below diff --git a/abort.go b/abort.go new file mode 100644 index 0000000..dc037db --- /dev/null +++ b/abort.go @@ -0,0 +1,88 @@ +/* + * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package dgo + +import ( + "strings" +) + +// AbortReason is the category of a transaction abort, as reported by the Dgraph server. +// +// The server encodes the category as a ": " prefix on the gRPC Aborted status +// message. Categories are matched on that prefix, and an absent or unrecognized prefix yields +// AbortReasonUnknown, so a newer server may introduce categories this client does not name without +// breaking it. +type AbortReason string + +const ( + // AbortReasonConflict is a write-write conflict with another concurrent transaction; retry with + // a fresh transaction. The server cannot say which key collided: conflict keys are one-way + // fingerprints by the time they are compared, so the culprit may be the data key written + // directly, or an index or count key derived from it. On an @upsert predicate the uid is + // excluded from the comparison, so any two transactions writing the same value conflict. + AbortReasonConflict AbortReason = "conflict" + // AbortReasonPredicateMove means a predicate the transaction touched is being moved between + // groups and commits on it are blocked, or it finished moving while the transaction was open. + // Back off, then retry with a fresh transaction once the move completes. + AbortReasonPredicateMove AbortReason = "predicate-move" + // AbortReasonStaleStartTs means the transaction's start timestamp is older than the oldest + // timestamp the server can still validate against. That happens after a Zero leader change, and + // also when Zero trims its conflict map at a snapshot. Retry with a fresh transaction. + AbortReasonStaleStartTs AbortReason = "stale-startts" + // AbortReasonUnknown means no category was reported. This covers aborts from older servers, + // which do not categorize at all, and aborts a current server declines to categorize because no + // published category fits: for example a transaction already aborted by a schema change or the + // idle-transaction reaper, a cancelled request, or a predicate no group currently serves. The + // error message still explains what happened. + AbortReasonUnknown AbortReason = "unknown" +) + +// AbortedError is returned when the server aborts a transaction. It wraps ErrAborted, so +// errors.Is(err, ErrAborted) reports true; use errors.As to read the reason: +// +// var ae *dgo.AbortedError +// if errors.As(err, &ae) && ae.Reason == dgo.AbortReasonPredicateMove { +// // back off before retrying +// } +// +// Comparing with == (err == dgo.ErrAborted) does not match an AbortedError; use errors.Is. +type AbortedError struct { + // Reason is the parsed abort category. + Reason AbortReason + // Message is the server's description of the abort, including any ": " prefix. + Message string +} + +func (e *AbortedError) Error() string { + if e.Message == "" { + return ErrAborted.Error() + } + return e.Message +} + +// Unwrap returns ErrAborted, so that errors.Is(err, ErrAborted) holds for every abort. +func (e *AbortedError) Unwrap() error { + return ErrAborted +} + +// newAbortedError builds the error returned for a server-side abort from the gRPC status message. +func newAbortedError(msg string) *AbortedError { + return &AbortedError{Reason: ParseAbortReason(msg), Message: msg} +} + +// ParseAbortReason returns the abort category encoded in a server abort message. Only the first +// colon delimits the category, since several server messages contain colons in the detail that +// follows. Matching is case-insensitive and ignores surrounding whitespace. A message with no +// recognized prefix returns AbortReasonUnknown. +func ParseAbortReason(msg string) AbortReason { + code, _, _ := strings.Cut(msg, ":") + switch r := AbortReason(strings.ToLower(strings.TrimSpace(code))); r { + case AbortReasonConflict, AbortReasonPredicateMove, AbortReasonStaleStartTs: + return r + default: + return AbortReasonUnknown + } +} diff --git a/abort_test.go b/abort_test.go new file mode 100644 index 0000000..31cf202 --- /dev/null +++ b/abort_test.go @@ -0,0 +1,55 @@ +/* + * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package dgo + +import ( + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestParseAbortReason(t *testing.T) { + cases := map[string]AbortReason{ + "conflict: Transaction has been aborted. Please retry": AbortReasonConflict, + "predicate-move: Commits on predicate name are blocked due to predicate move": AbortReasonPredicateMove, + // The detail can contain colons of its own; only the first one delimits the category. + "predicate-move: Mutation done in group: 1. Predicate name assigned to 2": AbortReasonPredicateMove, + "stale-startts: Transaction start timestamp is older than ...": AbortReasonStaleStartTs, + " CONFLICT : detail": AbortReasonConflict, + "conflict": AbortReasonConflict, + + // A server without abort reasons, or a cause the server leaves uncategorized. + "Transaction has been aborted. Please retry": AbortReasonUnknown, + "Tablet for name is nil": AbortReasonUnknown, + "context deadline exceeded": AbortReasonUnknown, + "": AbortReasonUnknown, + // A category this client does not know about yet. + "predicate-unavailable: something": AbortReasonUnknown, + } + for msg, want := range cases { + require.Equal(t, want, ParseAbortReason(msg), "ParseAbortReason(%q)", msg) + } +} + +func TestAbortedError(t *testing.T) { + const msg = "predicate-move: Commits on predicate name are blocked due to predicate move" + var err error = fmt.Errorf("committing: %w", newAbortedError(msg)) + + require.ErrorIs(t, err, ErrAborted) + require.False(t, errors.Is(err, ErrFinished)) + + var ae *AbortedError + require.ErrorAs(t, err, &ae) + require.Equal(t, AbortReasonPredicateMove, ae.Reason) + require.Equal(t, msg, ae.Message) + require.Equal(t, msg, ae.Error(), "Error returns the server's message verbatim") + + // No message from the server falls back to the legacy text. + require.Equal(t, ErrAborted.Error(), newAbortedError("").Error()) + require.Equal(t, AbortReasonUnknown, newAbortedError("").Reason) +} diff --git a/errors_test.go b/errors_test.go index 05def75..3021fcb 100644 --- a/errors_test.go +++ b/errors_test.go @@ -104,5 +104,15 @@ func TestTxnErrAborted(t *testing.T) { err = txn1.Commit(ctx1) require.NoError(t, err) - require.Error(t, txn2.Commit(ctx2), dgo.ErrAborted, "2nd transaction should have aborted") + err = txn2.Commit(ctx2) + require.ErrorIs(t, err, dgo.ErrAborted, "2nd transaction should have aborted") + + var ae *dgo.AbortedError + require.ErrorAs(t, err, &ae) + t.Logf("abort reason %q: %s", ae.Reason, ae.Message) + // A server that reports abort reasons calls this a conflict; an older one reports none. + require.Contains(t, []dgo.AbortReason{dgo.AbortReasonConflict, dgo.AbortReasonUnknown}, ae.Reason, + "unexpected abort reason for %q", ae.Message) + require.Contains(t, ae.Error(), dgo.ErrAborted.Error(), + "a conflict abort keeps the legacy sentence in its message") } diff --git a/txn.go b/txn.go index a301a8b..fc145c5 100644 --- a/txn.go +++ b/txn.go @@ -23,7 +23,8 @@ var ( ErrFinished = errors.New("Transaction has already been committed or discarded") // ErrReadOnly is returned when a write/update is performed on a readonly transaction ErrReadOnly = errors.New("Readonly transaction cannot run mutations or be committed") - // ErrAborted is returned when an operation is performed on an aborted transaction. + // ErrAborted is matched by every abort the server reports. Transactions return an + // *AbortedError wrapping it, so test with errors.Is(err, ErrAborted) rather than ==. ErrAborted = errors.New("Transaction has been aborted. Please retry") ) @@ -207,7 +208,7 @@ func (txn *Txn) Do(ctx context.Context, req *api.Request) (*api.Response, error) // If the transaction was aborted, return the right error // so the caller can handle it. if s, ok := status.FromError(err); ok && s.Code() == codes.Aborted { - err = ErrAborted + err = newAbortedError(s.Message()) } } @@ -217,8 +218,9 @@ func (txn *Txn) Do(ctx context.Context, req *api.Request) (*api.Response, error) // Commit commits any mutations that have been made in the transaction. // Once Commit has been called, the lifespan of the transaction is complete. // -// Errors could be returned for various reasons. Notably, ErrAborted could be -// returned if transactions that modify the same data are being run concurrently. +// Errors could be returned for various reasons. Notably, an *AbortedError (for which +// errors.Is(err, ErrAborted) is true) could be returned if transactions that modify +// the same data are being run concurrently. Its Reason says why the server aborted. // It's up to the user to decide if they wish to retry. // In this case, the user should create a new transaction. func (txn *Txn) Commit(ctx context.Context) error { @@ -231,7 +233,7 @@ func (txn *Txn) Commit(ctx context.Context) error { err := txn.commitOrAbort(ctx) if s, ok := status.FromError(err); ok && s.Code() == codes.Aborted { - err = ErrAborted + err = newAbortedError(s.Message()) } return err From 5427550c9fac4c93e9f533e973038c3df1c88b70 Mon Sep 17 00:00:00 2001 From: Matthew McNeely Date: Thu, 22 Oct 2026 12:26:06 -0400 Subject: [PATCH 2/3] feat(txn): report AbortedError as codes.Aborted via GRPCStatus status.Code(err) on an *AbortedError returned codes.Unknown, so code that classifies errors by gRPC status (the dgraph live loader, for one) didn't see an abort. GRPCStatus reports the Aborted status with the server's message. Test fixtures now use the server's final framing, where every abort message carries "Transaction has been aborted. Please retry" after the category. Co-Authored-By: Claude Opus 5.5 --- abort.go | 9 +++++++++ abort_test.go | 27 ++++++++++++++++++--------- txn.go | 5 +++-- 3 files changed, 30 insertions(+), 11 deletions(-) diff --git a/abort.go b/abort.go index dc037db..411336b 100644 --- a/abort.go +++ b/abort.go @@ -7,6 +7,9 @@ package dgo import ( "strings" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) // AbortReason is the category of a transaction abort, as reported by the Dgraph server. @@ -68,6 +71,12 @@ func (e *AbortedError) Unwrap() error { return ErrAborted } +// GRPCStatus reports the abort as the codes.Aborted status the server sent, so status.Code(err) +// and status.Convert(err) see an abort rather than codes.Unknown. +func (e *AbortedError) GRPCStatus() *status.Status { + return status.New(codes.Aborted, e.Error()) +} + // newAbortedError builds the error returned for a server-side abort from the gRPC status message. func newAbortedError(msg string) *AbortedError { return &AbortedError{Reason: ParseAbortReason(msg), Message: msg} diff --git a/abort_test.go b/abort_test.go index 31cf202..af886ce 100644 --- a/abort_test.go +++ b/abort_test.go @@ -11,23 +11,28 @@ import ( "testing" "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func TestParseAbortReason(t *testing.T) { cases := map[string]AbortReason{ - "conflict: Transaction has been aborted. Please retry": AbortReasonConflict, - "predicate-move: Commits on predicate name are blocked due to predicate move": AbortReasonPredicateMove, + "conflict: Transaction has been aborted. Please retry": AbortReasonConflict, + "predicate-move: Transaction has been aborted. Please retry. Commits on predicate name are " + + "blocked due to predicate move": AbortReasonPredicateMove, // The detail can contain colons of its own; only the first one delimits the category. - "predicate-move: Mutation done in group: 1. Predicate name assigned to 2": AbortReasonPredicateMove, - "stale-startts: Transaction start timestamp is older than ...": AbortReasonStaleStartTs, + "predicate-move: Transaction has been aborted. Please retry. Mutation done in group: 1. " + + "Predicate name assigned to 2": AbortReasonPredicateMove, + "stale-startts: Transaction has been aborted. Please retry. Its start timestamp is older " + + "than ...": AbortReasonStaleStartTs, " CONFLICT : detail": AbortReasonConflict, "conflict": AbortReasonConflict, // A server without abort reasons, or a cause the server leaves uncategorized. - "Transaction has been aborted. Please retry": AbortReasonUnknown, - "Tablet for name is nil": AbortReasonUnknown, - "context deadline exceeded": AbortReasonUnknown, - "": AbortReasonUnknown, + "Transaction has been aborted. Please retry": AbortReasonUnknown, + "Transaction has been aborted. Please retry. Tablet for name is nil": AbortReasonUnknown, + "Transaction has been aborted. Please retry. context deadline exceeded": AbortReasonUnknown, + "": AbortReasonUnknown, // A category this client does not know about yet. "predicate-unavailable: something": AbortReasonUnknown, } @@ -37,7 +42,8 @@ func TestParseAbortReason(t *testing.T) { } func TestAbortedError(t *testing.T) { - const msg = "predicate-move: Commits on predicate name are blocked due to predicate move" + const msg = "predicate-move: Transaction has been aborted. Please retry. Commits on predicate name " + + "are blocked due to predicate move" var err error = fmt.Errorf("committing: %w", newAbortedError(msg)) require.ErrorIs(t, err, ErrAborted) @@ -49,6 +55,9 @@ func TestAbortedError(t *testing.T) { require.Equal(t, msg, ae.Message) require.Equal(t, msg, ae.Error(), "Error returns the server's message verbatim") + require.Equal(t, codes.Aborted, status.Code(err), "status.Code sees through the wrapping") + require.Equal(t, msg, status.Convert(errors.Unwrap(err)).Message()) + // No message from the server falls back to the legacy text. require.Equal(t, ErrAborted.Error(), newAbortedError("").Error()) require.Equal(t, AbortReasonUnknown, newAbortedError("").Reason) diff --git a/txn.go b/txn.go index fc145c5..6ef8104 100644 --- a/txn.go +++ b/txn.go @@ -23,8 +23,9 @@ var ( ErrFinished = errors.New("Transaction has already been committed or discarded") // ErrReadOnly is returned when a write/update is performed on a readonly transaction ErrReadOnly = errors.New("Readonly transaction cannot run mutations or be committed") - // ErrAborted is matched by every abort the server reports. Transactions return an - // *AbortedError wrapping it, so test with errors.Is(err, ErrAborted) rather than ==. + // ErrAborted is returned when the server aborts a transaction. Commit, and Mutate or Do + // with mutations, return an *AbortedError wrapping it, so test with + // errors.Is(err, ErrAborted) rather than ==. ErrAborted = errors.New("Transaction has been aborted. Please retry") ) From 7875dfe8052b9b19cf07717cde2cbb9629d0359b Mon Sep 17 00:00:00 2001 From: Matthew McNeely Date: Fri, 23 Oct 2026 13:08:58 -0400 Subject: [PATCH 3/3] test: let the wrapped abort error's type be inferred staticcheck ST1023 flagged the explicit error type on a fmt.Errorf result. Co-Authored-By: Claude Opus 5.5 --- abort_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/abort_test.go b/abort_test.go index af886ce..c8fff71 100644 --- a/abort_test.go +++ b/abort_test.go @@ -44,7 +44,7 @@ func TestParseAbortReason(t *testing.T) { func TestAbortedError(t *testing.T) { const msg = "predicate-move: Transaction has been aborted. Please retry. Commits on predicate name " + "are blocked due to predicate move" - var err error = fmt.Errorf("committing: %w", newAbortedError(msg)) + err := fmt.Errorf("committing: %w", newAbortedError(msg)) require.ErrorIs(t, err, ErrAborted) require.False(t, errors.Is(err, ErrFinished))