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..411336b --- /dev/null +++ b/abort.go @@ -0,0 +1,97 @@ +/* + * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +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. +// +// 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 +} + +// 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} +} + +// 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..c8fff71 --- /dev/null +++ b/abort_test.go @@ -0,0 +1,64 @@ +/* + * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package dgo + +import ( + "errors" + "fmt" + "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: 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: 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, + "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, + } + for msg, want := range cases { + require.Equal(t, want, ParseAbortReason(msg), "ParseAbortReason(%q)", msg) + } +} + +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" + err := 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") + + 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/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..6ef8104 100644 --- a/txn.go +++ b/txn.go @@ -23,7 +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 returned when an operation is performed on an aborted transaction. + // 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") ) @@ -207,7 +209,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 +219,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 +234,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