Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 19 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
97 changes: 97 additions & 0 deletions abort.go
Original file line number Diff line number Diff line change
@@ -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 "<code>: <detail>" 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 "<code>: " 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
}
}
64 changes: 64 additions & 0 deletions abort_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
12 changes: 11 additions & 1 deletion errors_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}
13 changes: 8 additions & 5 deletions txn.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
)

Expand Down Expand Up @@ -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())
}
}

Expand All @@ -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 {
Expand All @@ -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
Expand Down
Loading