diff --git a/.trunk/trunk.yaml b/.trunk/trunk.yaml index 10545f03180..2335ee52e0f 100644 --- a/.trunk/trunk.yaml +++ b/.trunk/trunk.yaml @@ -16,7 +16,7 @@ runtimes: enabled: - node@22.16.0 - python@3.10.8 - - go@1.27.0 + - go@1.27.3 # This is the section where you manage your linters. (https://docs.trunk.io/check/configuration) lint: @@ -28,7 +28,7 @@ lint: enabled: - taplo@0.10.0 - svgo@4.0.1 - - golangci-lint2@2.13.1 + - golangci-lint2@2.15.0 - trivy@0.69.3 - actionlint@1.7.11 - checkov@3.2.513 diff --git a/dgraph/cmd/alpha/txn_test.go b/dgraph/cmd/alpha/txn_test.go index 54e6bc11582..e220882bba8 100644 --- a/dgraph/cmd/alpha/txn_test.go +++ b/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,118 @@ 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 the dgo version this repo builds against replaces any +// codes.Aborted error with the static dgo.ErrAborted, discarding the reason +// (dgraph-io/dgo#288 keeps it). +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 (this repo's dgo replaces 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()) +} + +// 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 transaction 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 "-"; 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() @@ -879,6 +996,6 @@ func TestTxnDiscardBeforeCommit(t *testing.T) { require.NoError(t, err, "unable to mutate") err = txn.Discard(context.Background()) - // Since client is discarding this transaction server should not throw ErrAborted err. - require.NotEqual(t, err, dgo.ErrAborted) + // Since client is discarding this transaction server should not return an error. + require.NoError(t, err) } diff --git a/dgraph/cmd/live/batch.go b/dgraph/cmd/live/batch.go index 2e3e7a7d704..757a3d9fb8d 100644 --- a/dgraph/cmd/live/batch.go +++ b/dgraph/cmd/live/batch.go @@ -7,6 +7,7 @@ package live import ( "context" + "errors" "fmt" "math" "math/rand" @@ -112,18 +113,21 @@ func handleError(err error, isRetry bool) { fmt.Printf("Connection has been possibly interrupted. Got error: %v."+ " Will retry after %s.\n", err, dur.Round(time.Second)) time.Sleep(dur) - case strings.Contains(s.Message(), "x509"): - x.Fatalf("%v", s.Message()) - case s.Code() == codes.Aborted: + // Checked before the x509 match below: an abort message can name a predicate, and a predicate + // called e.g. x509cert must not make an abort fatal. dgo flattens the Aborted status to + // ErrAborted (or, from dgraph-io/dgo#288, an error wrapping it), so check for that too. + case s.Code() == codes.Aborted || errors.Is(err, dgo.ErrAborted): if !isRetry && opt.verbose { fmt.Printf("Transaction aborted. Will retry in background.\n") } + case strings.Contains(s.Message(), "x509"): + x.Fatalf("%v", s.Message()) case strings.Contains(s.Message(), "Server overloaded."): //nolint:gosec // random generator in closed set does not require cryptographic precision dur := time.Duration(1+rand.Intn(10)) * time.Minute fmt.Printf("Server is overloaded. Will retry after %s.\n", dur.Round(time.Minute)) time.Sleep(dur) - case err != x.ErrConflict && err != dgo.ErrAborted: + case !errors.Is(err, x.ErrConflict): fmt.Printf("Error while mutating: %v s.Code %v\n", s.Message(), s.Code()) } } diff --git a/dgraph/cmd/zero/oracle.go b/dgraph/cmd/zero/oracle.go index 663f8ded0d4..4d50eb29ad3 100644 --- a/dgraph/cmd/zero/oracle.go +++ b/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" @@ -120,10 +122,19 @@ func (o *Oracle) purgeBelow(minTs uint64) { } } +// errStaleStartTs is returned by Oracle.commit when the start timestamp is below startTxnTs. It is +// decided under the same lock as the conflict check, so the late abort path in Server.commit reports +// what the oracle actually saw. Re-reading startTxnTs afterwards could mislabel the abort, because +// purgeBelow and updateStartTxnTs can both move it in between. +var errStaleStartTs = errors.New("transaction start timestamp is stale") + func (o *Oracle) commit(src *api.TxnContext) error { o.Lock() defer o.Unlock() + if o.isStaleStartTs(src) { + return errStaleStartTs + } if o.hasConflict(src) { return x.ErrConflict } @@ -324,13 +335,11 @@ func (s *Server) proposeTxn(ctx context.Context, src *api.TxnContext) error { return err } - // There might be race between this proposal trying to commit and predicate - // move aborting it. A predicate move, triggered by Zero, would abort all - // pending transactions. At the same time, a client which has already done - // mutations, can proceed to commit it. A race condition can happen here, - // with both proposing their respective states, only one can succeed after - // the proposal is done. So, check again to see the fate of the transaction - // here. + // This proposal can race with an abort proposed for the same start ts elsewhere: a TryAbort + // from an Alpha (schema update, drop-predicate, or the idle-transaction reaper), or a client + // Discard of the same transaction. Both sides propose their respective states and only one + // can succeed, so check again to see the fate of the transaction here. (A predicate move does + // not abort pending transactions; it blocks their commits in checkPreds instead.) src.CommitTs = s.orc.commitTs(src.StartTs) if src.CommitTs == 0 { src.Aborted = true @@ -338,56 +347,272 @@ 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 ": ". +// api.TxnContext (external dgo module) has no field for the reason, so the reason rides on the +// error instead; the status code stays codes.Aborted, so existing abort handling is unaffected. +// Client support for parsing these prefixes into an AbortReason enum is proposed in +// dgraph-io/dgraph4j#290, dgraph-io/pydgraph#307 and dgraph-io/dgo#288; keep the server and +// those parsers in sync. +// +// These three codes are the entire published vocabulary. A cause that cannot be mapped onto one of +// them emits the detail with *no* prefix (abortReasonUncategorized) rather than borrowing a code +// that would imply the wrong remedy. Each of those client parsers maps an absent or unrecognized +// prefix to UNKNOWN, so "no category" is a supported and truthful answer. Withholding +// also keeps a later phase purely additive: naming a new category then only adds precision to +// something that was already UNKNOWN, and never reclassifies a code clients have shipped against. +const ( + // abortReasonUncategorized means the server knows what happened and says so in the detail, + // but the cause does not correspond to any published category. It is not "the server does not + // know" — it is "there is no code for this yet". + abortReasonUncategorized = "" + abortReasonConflict = "conflict" + abortReasonStaleStartTs = "stale-startts" + abortReasonPredicateMove = "predicate-move" +) + +// Human-readable details, one per cause. Every message a client can receive for a server-decided +// abort is declared here rather than at its call site, so the published surface is a single block: +// adding a new abort means adding a constant next to the existing vocabulary, where the category it +// should pair with is visible. Entries ending in Fmt are format strings and must be given their +// arguments; the others are complete. abortReason frames each one as +// ": Transaction has been aborted. Please retry. ". +// +// category detail +// ─────────────── ────────────────────────────────────────────────────────── +// conflict abortDetailConflict +// stale-startts abortDetailStaleStartTs +// predicate-move abortDetailPredicateBlockedFmt, abortDetailGroupMismatchFmt +// (withheld) abortDetailMissingGroupIDFmt, abortDetailBadGroupIDFmt, +// abortDetailTabletNilFmt, abortDetailPreAborted, and +// ctx.Err() — the last supplied by the Go runtime, so it is +// not declared here +const ( + // A conflict is the one category the server cannot narrow. Every conflict key reaching Zero is a + // farm.Fingerprint64(key)^uid produced by GetConflictKey (posting/list.go), so by the time + // hasConflict matches one, which key produced it is already unrecoverable. Naming the + // possibilities is therefore the most this category can honestly say, and it is worth saying: a + // caller who set a single scalar value has no reason to expect that the index and count keys + // *derived* from it can conflict too. + // + // Exactly three kinds of key carry a conflict — data, index and count. Reverse keys carry none: + // IsReverse is a distinct ByteType, so a reverse key matches no case in GetConflictKey's switch, + // falls to default, and addConflictKey drops the resulting zero. Do not add "reverse" here. + // + // @upsert is named separately because it is a *rule*, not a fourth kind of key. HasUpsert is + // tested ahead of every IsData/IsIndex case, so on an @upsert predicate the existing data and + // index keys switch to getKey(key, 0) — the uid drops out and any two transactions writing the + // same value collide on the shared index key. That is the uniqueness mechanism, and in + // upsert-heavy ingest it is the most likely cause of this abort, which is why it is worth the + // extra sentence. + // + // abortReason puts abortLegacyMessage in front of this, so the wire message still starts with the + // sentence that dgraph/cmd/alpha/upsert_test.go, query/query4_test.go and systest/unique_test.go + // match against the live server. + abortDetailConflict = "Another transaction committed to one of the same keys. The conflicting " + + "key cannot be identified: it may be the data key written directly, or an index or count " + + "key derived from it. On an @upsert predicate the uid is excluded, so any two transactions " + + "writing the same value conflict" + // A start timestamp goes stale for two different reasons, so this detail deliberately names + // both rather than asserting one. Zero raises startTxnTs either when a Zero becomes leader and + // renews its leases (updateLeases, assign.go) or when it trims its conflict map while applying + // a snapshot (purgeBelow via applySnapshot, raft.go) — the second is not a leader change at + // all. Both mean the same thing to a caller: this transaction's start timestamp is older than + // the oldest timestamp Zero can still validate against, so retry with a fresh transaction. + abortDetailStaleStartTs = "Its start timestamp is older than the oldest timestamp Zero can " + + "still validate (Zero leader change, or its conflict map was trimmed at a snapshot)" + + // Reported when proposeTxn comes back with the transaction already aborted while this commit + // decided nothing itself — i.e. something else proposed an abort for the same start timestamp + // before this commit was arbitrated. Zero records no cause when that happens (an abort proposal + // carries only timestamps), so the category is withheld and the detail names the possibilities + // instead of guessing. None of them is a predicate *move* — a move blocks commits in checkPreds + // before a commit timestamp is leased, and never aborts pending transactions — so + // predicate-move would be the wrong label. The known producers are: + // + // - a schema or type update on a predicate the transaction touched, which cancels pending + // transactions so the index can be rebuilt (detectPendingTxns, worker/draft.go) + // - a drop-predicate (S * * delete) on a predicate the transaction touched (same function) + // - the Alpha leader ageing out transactions idle longer than --limit "txn-abort-after" + // (abortOldTransactions, worker/draft.go) + // - a client Discard of the same transaction racing this commit, which reaches commit()'s own + // src.Aborted early return and proposes the abort there + // + // The first three reach Zero through TryAbort (blockingAbort, worker/draft.go). + // + // Before this was handled, commit() returned nil here and the abort reached the client as a + // bare dgo.ErrAborted with no reason at all. + abortDetailPreAborted = "It was already aborted before this commit was decided, for example " + + "because a schema update or a drop-predicate cancelled pending transactions on a predicate " + + "it touched, the server aged it out after --limit \"txn-abort-after\" of idleness, or the " + + "client discarded it concurrently" + + // Details produced by checkPreds. The wording predates the abort-reason work, but commit() used to + // discard these errors, so clients have never seen them. The predicate argument is the name as + // the user wrote it (see displayPred), not the namespaced attribute. + abortDetailMissingGroupIDFmt = "Unable to find group id in %s" + abortDetailBadGroupIDFmt = "unable to parse group id from %s" + abortDetailTabletNilFmt = "Tablet for %s is nil" + abortDetailPredicateBlockedFmt = "Commits on predicate %s are blocked due to predicate move" + abortDetailGroupMismatchFmt = "Mutation done in group: %d. Predicate %s assigned to %d" +) + +// abortLegacyMessage is the text of every abort from a server without abort reasons, and the text +// of dgo.ErrAborted. abortReason puts it in front of every detail, so each abort message still +// contains it. Clients that recognise aborts by message text rely on that: the dgraph-js-http +// client treats an HTTP error as an abort only when its message contains "abort" and "retry", and +// rethrows anything else as a hard failure. Tests in this repo and user log-scrapers also +// substring-match it. Never change it, and never emit an abort that bypasses abortReason. +const abortLegacyMessage = "Transaction has been aborted. Please retry" + +// abortReason builds the wire string the client parses: ": . ". An +// empty code drops the ": " prefix, so the client reports no category, but the message still +// starts with the legacy sentence a server without this feature sent for every abort. +func abortReason(code, detail string) string { + msg := abortLegacyMessage + ". " + detail + if code == abortReasonUncategorized { + return msg + } + return code + ": " + msg +} + +// conflictAbortReason returns the wire reason for an abort decided by hasConflict, distinguishing a +// write-write conflict from a stale start timestamp. See isStaleStartTs for what "stale" means. +func conflictAbortReason(stale bool) string { + if stale { + return abortReason(abortReasonStaleStartTs, abortDetailStaleStartTs) + } + return abortReason(abortReasonConflict, abortDetailConflict) +} + +// lateAbortReason returns the wire reason for an abort decided after the commit timestamp was +// leased, and false if nothing aborted the transaction. orcErr is Oracle.commit's verdict and ctxErr +// is the request context's error, read after it. Two unrelated causes can abort here. An oracle +// rejection takes precedence: the oracle would have aborted the transaction whatever the context +// said, so that is the honest cause to report. +func lateAbortReason(orcErr, ctxErr error) (string, bool) { + switch { + case orcErr != nil: + // Oracle.commit re-runs the conflict check, and a start timestamp that went stale since the + // early check in commit() (a lease renewal or snapshot purge in between) is not a write-write + // conflict. Oracle.commit reports that case as errStaleStartTs so it can be told apart. + return conflictAbortReason(errors.Is(orcErr, errStaleStartTs)), true + case ctxErr != nil: + // The context was cancelled or timed out; no conflict occurred. No published category + // describes this, so report the context's own message with no category rather than claiming + // a conflict that did not happen. The gRPC status code stays codes.Aborted because the + // transaction genuinely did abort; switching to Canceled or DeadlineExceeded would change + // retry behaviour for clients that already treat Aborted as retryable. + return abortReason(abortReasonUncategorized, ctxErr.Error()), true + } + return "", false +} + +// isStaleStartTs reports whether this transaction's start timestamp (src.StartTs) is below the +// floor Zero can still validate against (startTxnTs). hasConflict aborts such a transaction, but +// the cause is not a write-write conflict, so it is reported under its own category. See +// abortDetailStaleStartTs for the two ways that floor rises. Callers must hold at least a read +// lock on the oracle. +func (o *Oracle) isStaleStartTs(src *api.TxnContext) bool { + return src.StartTs < o.startTxnTs +} + +// checkPreds reports whether any predicate this transaction touched is unusable, in which case the +// transaction must be aborted. It returns the abort category alongside the error because its exits +// are unrelated causes, not degrees of one cause: only a group mismatch and a blocked tablet are +// predicate moves. A malformed entry in preds, or a predicate that no group serves, are different +// failures needing different remedies — retrying a malformed predicate key can never succeed — so +// those return abortReasonUncategorized rather than claiming a move that did not happen. Every exit +// still aborts the transaction, exactly as before; only the reported category changes. +// +// Each entry in preds is a "-" string built by the Alpha +// (LocalCache.fillPreds). +func (s *Server) checkPreds(preds []string) (string, error) { + for _, pkey := range preds { + splits := strings.SplitN(pkey, "-", 2) + if len(splits) < 2 { + return abortReasonUncategorized, errors.Errorf(abortDetailMissingGroupIDFmt, pkey) + } + // ParseUint with bitSize 32 rather than Atoi: group ids are uint32 everywhere, so this + // bounds the value to the type it is compared against below and rejects negatives, which + // Atoi accepted and the uint32 conversion silently wrapped (CodeQL go/incorrect-integer-conversion). + gid, err := strconv.ParseUint(splits[0], 10, 32) + if err != nil { + return abortReasonUncategorized, errors.Wrapf(err, abortDetailBadGroupIDFmt, pkey) + } + pred := splits[1] + if strings.Contains(pred, hnsw.VecKeyword) { + pred = pred[0:strings.Index(pred, hnsw.VecKeyword)] + } + // Checked before the tablet lookup because blockTablet holds for the entire duration of a + // move: movePredicate defers its unblock past the tablet reassignment proposal, so this is + // the authoritative signal that a move is in flight. Checking it first means an in-flight + // move can never be reported as anything else, and it lets the tablet == nil case below mean + // only "no group serves this predicate" and never "it is moving". + if s.isBlocked(pred) { + return abortReasonPredicateMove, errors.Errorf(abortDetailPredicateBlockedFmt, + displayPred(pred)) + } + tablet := s.ServingTablet(pred) + if tablet == nil { + return abortReasonUncategorized, errors.Errorf(abortDetailTabletNilFmt, displayPred(pred)) + } + // The predicate finished moving to another group while this transaction was open: the + // mutation was written against group gid, but the predicate now belongs elsewhere. + if tablet.GroupId != uint32(gid) { + return abortReasonPredicateMove, errors.Errorf(abortDetailGroupMismatchFmt, + gid, displayPred(pred), tablet.GroupId) + } + } + return abortReasonUncategorized, nil +} + +// displayPred turns a namespaced attribute ("0-name") into the predicate name the user wrote +// ("name") for use in an abort detail. Preds is client-supplied, so unlike x.ParseAttr this must not +// assume the separator is present: a missing separator or an empty name returns the input as is. +func displayPred(pred string) string { + if _, attr, ok := strings.Cut(pred, x.NsSeparator); ok && attr != "" { + return attr + } + return pred +} + 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 { + // The client discarded this transaction itself (a Discard call), so the server decided + // nothing and there is no reason to report. return s.proposeTxn(ctx, src) } + // abortWithReason marks the transaction 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) + // hasConflict also aborts a transaction whose start timestamp is too old to validate, which is + // not a write-write conflict, so it is reported under its own category. See isStaleStartTs. + stale := s.orc.isStaleStartTs(src) s.orc.RUnlock() if conflict { - span.SetAttributes(attribute.Bool("abort", true)) - src.Aborted = true - return s.proposeTxn(ctx, src) + return abortWithReason(conflictAbortReason(stale)) } - checkPreds := func() error { - // Check if any of these tablets is being moved. If so, abort the transaction. - for _, pkey := range src.Preds { - splits := strings.SplitN(pkey, "-", 2) - if len(splits) < 2 { - return errors.Errorf("Unable to find group id in %s", pkey) - } - gid, err := strconv.Atoi(splits[0]) - if err != nil { - return errors.Wrapf(err, "unable to parse group id from %s", pkey) - } - pred := splits[1] - if strings.Contains(pred, hnsw.VecKeyword) { - pred = pred[0:strings.Index(pred, hnsw.VecKeyword)] - } - tablet := s.ServingTablet(pred) - if tablet == nil { - return errors.Errorf("Tablet for %s is nil", pred) - } - if tablet.GroupId != uint32(gid) { - return errors.Errorf("Mutation done in group: %d. Predicate %s assigned to %d", - gid, pred, tablet.GroupId) - } - if s.isBlocked(pred) { - return errors.Errorf("Commits on predicate %s are blocked due to predicate move", pred) - } - } - return nil - } - if err := checkPreds(); err != nil { - span.SetAttributes(attribute.Bool("abort", true)) - src.Aborted = true - return s.proposeTxn(ctx, src) + if reason, err := s.checkPreds(src.Preds); err != nil { + // checkPreds already builds a specific, human-readable message for each cause. Forward it + // instead of discarding it, tagged with whichever category that cause justifies. + return abortWithReason(abortReason(reason, err.Error())) } num := pb.Num{Val: 1, Type: pb.Num_TXN_TS} @@ -402,16 +627,33 @@ 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)) - if err := s.orc.commit(src); err != nil { + // Past this point the transaction already holds a commit timestamp, so an abort here is a "late" + // abort. See lateAbortReason for how its cause is reported. + reason, aborted := lateAbortReason(s.orc.commit(src), ctx.Err()) + if aborted { span.SetAttributes(attribute.Bool("abort", true)) src.Aborted = true } - if err := ctx.Err(); err != nil { + // Propose txn should be used to set watermark as done. + if err := s.proposeTxn(ctx, src); err != nil { + return err + } + if aborted { + return status.Error(codes.Aborted, reason) + } + // proposeTxn sets src.Aborted when it finds no commit timestamp for this transaction, meaning + // something aborted it out of band while this commit was in flight: a TryAbort from a schema + // update, a drop-predicate, or the idle-transaction reaper, or a concurrent client Discard of the + // same transaction. Nothing above set `aborted`, because + // this commit itself decided to proceed. Returning nil here (as this did previously) loses the + // abort entirely: CommitOverNetwork then sees tctx.Aborted with no error and falls through to a + // bare dgo.ErrAborted, so the caller learns nothing. Report it instead. + if src.Aborted { span.SetAttributes(attribute.Bool("abort", true)) - src.Aborted = true + return status.Error(codes.Aborted, + abortReason(abortReasonUncategorized, abortDetailPreAborted)) } - // Propose txn should be used to set watermark as done. - return s.proposeTxn(ctx, src) + return nil } // CommitOrAbort either commits a transaction or aborts it. diff --git a/dgraph/cmd/zero/oracle_reason_test.go b/dgraph/cmd/zero/oracle_reason_test.go new file mode 100644 index 00000000000..b6e2c488e8d --- /dev/null +++ b/dgraph/cmd/zero/oracle_reason_test.go @@ -0,0 +1,362 @@ +/* + * SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package zero + +import ( + "context" + "fmt" + "strings" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/dgraph-io/dgo/v250/protos/api" + "github.com/dgraph-io/dgraph/v25/protos/pb" + "github.com/dgraph-io/dgraph/v25/x" +) + +// The abort-reason wire format is a contract with gRPC clients (e.g. dgraph4j parses the +// ": " 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) { + const legacy = "Transaction has been aborted. Please retry" + require.Equal(t, legacy, abortLegacyMessage, "the legacy sentence must never change") + require.Equal(t, "conflict: "+legacy+". boom", abortReason(abortReasonConflict, "boom")) + require.Equal(t, "stale-startts: "+legacy+". x", abortReason(abortReasonStaleStartTs, "x")) + require.Equal(t, "predicate-move: "+legacy+". y", abortReason(abortReasonPredicateMove, "y")) + require.Equal(t, legacy+". z", abortReason(abortReasonUncategorized, "z")) +} + +// TestAbortVocabulary pins the entire published surface: every category, and every detail a client +// can receive for a server-decided abort, with the category each detail pairs with. Because the +// details are declared in one block in oracle.go, adding a new abort message without deciding its +// category — or quietly rewording an existing one that dgraph4j, pydgraph or the docs depend on — +// fails here. The wording assertions are the pre-abort-reason text verbatim, so hoisting these to +// constants is provably a move and not an edit. +func TestAbortVocabulary(t *testing.T) { + // The three published codes, plus the withheld sentinel. Clients switch on exactly these. + require.Equal(t, "", abortReasonUncategorized) + require.Equal(t, "conflict", abortReasonConflict) + require.Equal(t, "stale-startts", abortReasonStaleStartTs) + require.Equal(t, "predicate-move", abortReasonPredicateMove) + + for _, tc := range []struct { + name string + category string + rendered string + }{ + {"write-write conflict", abortReasonConflict, + abortDetailConflict}, + {"stale start timestamp", abortReasonStaleStartTs, + abortDetailStaleStartTs}, + {"move in flight", abortReasonPredicateMove, + fmt.Sprintf(abortDetailPredicateBlockedFmt, "friend")}, + {"move completed", abortReasonPredicateMove, + fmt.Sprintf(abortDetailGroupMismatchFmt, 1, "friend", 2)}, + {"predicate served by no group", abortReasonUncategorized, + fmt.Sprintf(abortDetailTabletNilFmt, "friend")}, + {"malformed key, no separator", abortReasonUncategorized, + fmt.Sprintf(abortDetailMissingGroupIDFmt, "1foo")}, + {"malformed key, bad group id", abortReasonUncategorized, + fmt.Sprintf(abortDetailBadGroupIDFmt, "xfoo")}, + {"aborted out of band before commit", abortReasonUncategorized, + abortDetailPreAborted}, + } { + t.Run(tc.name, func(t *testing.T) { + wire := abortReason(tc.category, tc.rendered) + // Every abort, categorized or not, must still contain the legacy sentence. Clients that + // spot aborts by message text (dgraph-js-http requires "abort" and "retry") would + // otherwise treat it as a hard failure instead of retrying. + lower := strings.ToLower(wire) + require.Contains(t, lower, "abort") + require.Contains(t, lower, "retry") + legacy := abortLegacyMessage + ". " + tc.rendered + if tc.category == abortReasonUncategorized { + require.Equal(t, legacy, wire, "a withheld cause must carry no prefix") + return + } + require.Equal(t, tc.category+": "+legacy, wire) + }) + } + + // The conflict detail names what it cannot narrow — a conflict key is a one-way fingerprint by + // the time Zero matches it, so data, index, count and @upsert conflicts are indistinguishable. + // Callers who set one scalar value do not expect the keys derived from it to conflict, so the + // possibilities are spelled out. + for _, want := range []string{"index", "count", "@upsert"} { + require.Contains(t, abortDetailConflict, want, + "conflict detail should name the key kinds it cannot distinguish between") + } + + // The out-of-band abort names its known causes rather than guessing a category. None of them is + // a predicate move, and the abort proposal that delivers each records no cause, so labelling + // this predicate-move would be wrong. Before it was handled, commit() returned nil and the abort + // reached the client as a bare dgo.ErrAborted with no reason at all. + for _, want := range []string{"schema update", "drop-predicate", "txn-abort-after", "discarded"} { + require.Contains(t, abortDetailPreAborted, want, + "pre-abort detail should name the causes it cannot distinguish between") + } + + // Backwards-compatibility guard: the conflict abort on the wire is byte-identical to what it was + // before abortReason took over the legacy sentence. dgraph/cmd/alpha/upsert_test.go loops while + // the live server error contains that sentence, and query/query4_test.go and + // systest/unique_test.go assert Contains on it. + require.Equal(t, "conflict: Transaction has been aborted. Please retry. Another transaction "+ + "committed to one of the same keys. The conflicting key cannot be identified: it may be the "+ + "data key written directly, or an index or count key derived from it. On an @upsert "+ + "predicate the uid is excluded, so any two transactions writing the same value conflict", + conflictAbortReason(false)) + + // Wording of the checkPreds details, verbatim from before the abort-reason work. + require.Equal(t, "Unable to find group id in 1foo", + fmt.Sprintf(abortDetailMissingGroupIDFmt, "1foo")) + require.Equal(t, "unable to parse group id from xfoo", + fmt.Sprintf(abortDetailBadGroupIDFmt, "xfoo")) + require.Equal(t, "Tablet for friend is nil", + fmt.Sprintf(abortDetailTabletNilFmt, "friend")) + require.Equal(t, "Commits on predicate friend are blocked due to predicate move", + fmt.Sprintf(abortDetailPredicateBlockedFmt, "friend")) + require.Equal(t, "Mutation done in group: 1. Predicate friend assigned to 2", + fmt.Sprintf(abortDetailGroupMismatchFmt, 1, "friend", 2)) +} + +// TestAbortReasonUncategorized pins the withholding contract: when the server cannot substantiate +// one of the published categories it emits the bare detail with no prefix, rather than borrowing a +// category that would imply the wrong remedy. It starts with the same sentence a pre-feature server +// sends for every abort, and the proposed dgraph4j, pydgraph and dgo parsers map a message without +// a category prefix to UNKNOWN. The absence of a colon-delimited prefix is the whole contract, so it +// must not regress. +func TestAbortReasonUncategorized(t *testing.T) { + require.Equal(t, "Transaction has been aborted. Please retry. Tablet for foo is nil", + abortReason(abortReasonUncategorized, "Tablet for foo is nil")) + require.Equal(t, "Transaction has been aborted. Please retry. context canceled", + abortReason(abortReasonUncategorized, "context canceled")) + + // No withheld detail may begin with a token a client would parse as a real category, otherwise + // withholding would be silently reinterpreted as a category. + for _, detail := range []string{ + "Unable to find group id in 1foo", + "Tablet for foo is nil", + "context canceled", + "context deadline exceeded", + } { + got := abortReason(abortReasonUncategorized, detail) + prefix := got + if i := strings.Index(got, ":"); i >= 0 { + prefix = got[:i] + } + for _, code := range []string{ + abortReasonConflict, abortReasonStaleStartTs, abortReasonPredicateMove, + } { + require.NotEqual(t, code, strings.ToLower(strings.TrimSpace(prefix)), + "withheld detail %q would be parsed as category %q", detail, code) + } + } +} + +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. The detail must name both ways startTxnTs rises — a Zero leader change + // and a conflict-map trim at a snapshot — because asserting only the first is wrong on the + // second path (purgeBelow via applySnapshot, which involves no leader change at all). + 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") + require.Contains(t, r, "snapshot") +} + +// TestCheckPredsCategories is the core of the abort-category fix. checkPreds has five exits and only +// two of them are predicate moves; the other three are unrelated failures with different remedies, +// so they must not be reported as moves. Every case still returns an error (the transaction aborts +// regardless) — this pins only which category each cause claims. +func TestCheckPredsCategories(t *testing.T) { + const pred = "friend" + servingIn := func(gid uint32) *Server { + s := &Server{} + s.blockCommitsOn = new(sync.Map) + s.state = &pb.MembershipState{Groups: map[uint32]*pb.Group{ + gid: {Tablets: map[string]*pb.Tablet{pred: {GroupId: gid, Predicate: pred}}}, + }} + return s + } + + t.Run("in-flight move is reported as a move", func(t *testing.T) { + // The authoritative signal, and it must win even though the tablet still resolves cleanly. + s := servingIn(1) + s.blockCommitsOn.Store(pred, struct{}{}) + reason, err := s.checkPreds([]string{"1-" + pred}) + require.Error(t, err) + require.Equal(t, abortReasonPredicateMove, reason) + require.Contains(t, err.Error(), "blocked due to predicate move") + }) + + t.Run("in-flight move wins over an absent tablet", func(t *testing.T) { + // Regression guard for the check ordering. blockTablet is held for the whole move, so if the + // tablet lookup ran first a genuine in-flight move could be reported as uncategorized. + s := &Server{state: &pb.MembershipState{Groups: map[uint32]*pb.Group{}}} + s.blockCommitsOn = new(sync.Map) + s.blockCommitsOn.Store(pred, struct{}{}) + reason, err := s.checkPreds([]string{"1-" + pred}) + require.Error(t, err) + require.Equal(t, abortReasonPredicateMove, reason, + "isBlocked must be consulted before the tablet lookup") + }) + + t.Run("completed move is reported as a move", func(t *testing.T) { + // Written against group 1, but the predicate now belongs to group 2. + reason, err := servingIn(2).checkPreds([]string{"1-" + pred}) + require.Error(t, err) + require.Equal(t, abortReasonPredicateMove, reason) + require.Contains(t, err.Error(), "assigned to 2") + }) + + t.Run("predicate served by no group is not a move", func(t *testing.T) { + s := &Server{state: &pb.MembershipState{Groups: map[uint32]*pb.Group{}}} + s.blockCommitsOn = new(sync.Map) + reason, err := s.checkPreds([]string{"1-" + pred}) + require.Error(t, err) + require.Equal(t, abortReasonUncategorized, reason, + "an unserved predicate is not a move; retrying may never succeed") + }) + + t.Run("malformed predicate key is not a move", func(t *testing.T) { + s := servingIn(1) + for _, pkey := range []string{pred, "x-" + pred} { + reason, err := s.checkPreds([]string{pkey}) + require.Error(t, err, "pkey %q must abort", pkey) + require.Equal(t, abortReasonUncategorized, reason, + "a malformed predicate key %q is not a move and can never succeed on retry", pkey) + } + }) + + t.Run("details name the predicate without its namespace", func(t *testing.T) { + // The Alpha sends "-", so after the group id is split off the predicate + // is still "-name". Lookups must use that form, but the client-visible detail must + // show the name the user wrote. + for _, ns := range []uint64{x.RootNamespace, 0x2a} { + attr := x.NamespaceAttr(ns, pred) + s := &Server{state: &pb.MembershipState{Groups: map[uint32]*pb.Group{ + 2: {Tablets: map[string]*pb.Tablet{attr: {GroupId: 2, Predicate: attr}}}, + }}} + s.blockCommitsOn = new(sync.Map) + + _, err := s.checkPreds([]string{"1-" + attr}) + require.EqualError(t, err, fmt.Sprintf(abortDetailGroupMismatchFmt, 1, pred, 2)) + + s.blockCommitsOn.Store(attr, struct{}{}) + _, err = s.checkPreds([]string{"2-" + attr}) + require.EqualError(t, err, fmt.Sprintf(abortDetailPredicateBlockedFmt, pred)) + + _, err = s.checkPreds([]string{"2-" + x.NamespaceAttr(ns, "other")}) + require.EqualError(t, err, fmt.Sprintf(abortDetailTabletNilFmt, "other")) + } + }) + + t.Run("healthy predicate does not abort", func(t *testing.T) { + reason, err := servingIn(1).checkPreds([]string{"1-" + pred}) + require.NoError(t, err) + require.Equal(t, abortReasonUncategorized, reason) + }) +} + +// TestHasConflictStaleStartTs pins the exact discriminator commit() uses to choose the +// stale-startts reason: a transaction whose startTs is below the leader's startTxnTs floor 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 floor: hasConflict true, and the stale discriminator true. + stale := &api.TxnContext{StartTs: 42} + require.True(t, o.hasConflict(stale), "transaction below startTxnTs must conflict") + require.True(t, o.isStaleStartTs(stale), "must be flagged stale") + require.Equal(t, conflictAbortReason(true), conflictAbortReason(o.isStaleStartTs(stale))) + + // startTs at/above the floor with no keys: not a conflict, not stale. + fresh := &api.TxnContext{StartTs: 100} + require.False(t, o.hasConflict(fresh), "fresh transaction with no keys must not conflict") + require.False(t, o.isStaleStartTs(fresh), "must not be flagged stale") +} + +// TestLateAbortStaleStartTs guards the late-abort path. Oracle.commit must tell a stale start +// timestamp apart from a genuine keyCommit conflict itself, under its own lock, and lateAbortReason +// must report each under its own category. Re-reading startTxnTs after Oracle.commit returned could +// mislabel the abort, since a snapshot purge or lease renewal can move it in between. +func TestLateAbortStaleStartTs(t *testing.T) { + o := &Oracle{} + o.Init() + defer o.close() + o.updateStartTxnTs(100) + + stale := o.commit(&api.TxnContext{StartTs: 42, CommitTs: 200}) + require.ErrorIs(t, stale, errStaleStartTs) + + first := &api.TxnContext{StartTs: 100, CommitTs: 150, Keys: []string{"a"}} + require.NoError(t, o.commit(first)) + conflict := o.commit(&api.TxnContext{StartTs: 120, CommitTs: 200, Keys: []string{"a"}}) + require.ErrorIs(t, conflict, x.ErrConflict, "second writer of key a must conflict") + + for _, tc := range []struct { + name string + orcErr, ctxErr error + want string + }{ + {"stale start ts", stale, nil, + abortReason(abortReasonStaleStartTs, abortDetailStaleStartTs)}, + {"write-write conflict", conflict, nil, + abortReason(abortReasonConflict, abortDetailConflict)}, + {"cancelled context", nil, context.Canceled, + abortReason(abortReasonUncategorized, "context canceled")}, + {"timed out context", nil, context.DeadlineExceeded, + abortReason(abortReasonUncategorized, "context deadline exceeded")}, + // The oracle would have aborted regardless of the context, so its cause wins. + {"conflict wins over context", conflict, context.Canceled, + abortReason(abortReasonConflict, abortDetailConflict)}, + {"stale wins over context", stale, context.DeadlineExceeded, + abortReason(abortReasonStaleStartTs, abortDetailStaleStartTs)}, + } { + t.Run(tc.name, func(t *testing.T) { + got, aborted := lateAbortReason(tc.orcErr, tc.ctxErr) + require.True(t, aborted) + require.Equal(t, tc.want, got) + }) + } + + got, aborted := lateAbortReason(nil, nil) + require.False(t, aborted, "nothing rejected the commit") + require.Empty(t, got) +} + +// TestDisplayPred pins that stripping the namespace for display can never panic on a hand-crafted +// Preds entry, which is client-supplied. x.ParseAttr would index past the end of a separator-less +// split. +func TestDisplayPred(t *testing.T) { + for in, want := range map[string]string{ + "0-name": "name", + "2a-name": "name", + "0-a-b": "a-b", + "nodash": "nodash", + "0-": "0-", + "": "", + "0-name@en": "name@en", + "-leadingns": "leadingns", + } { + require.Equal(t, want, displayPred(in), "displayPred(%q)", in) + } +} diff --git a/edgraph/server.go b/edgraph/server.go index 3a03282fcd4..7963328ee37 100644 --- a/edgraph/server.go +++ b/edgraph/server.go @@ -726,6 +726,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 @@ -2209,6 +2213,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 diff --git a/systest/integration2/txn_abort_reason_test.go b/systest/integration2/txn_abort_reason_test.go new file mode 100644 index 00000000000..0c6d772bf4b --- /dev/null +++ b/systest/integration2/txn_abort_reason_test.go @@ -0,0 +1,88 @@ +//go:build integration2 + +/* + * SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc. + * SPDX-License-Identifier: Apache-2.0 + */ + +package main + +import ( + "context" + "strings" + "testing" + "time" + + "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/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 timestamp, so committing the now-old +// transaction 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 (the dgo version this repo builds against replaces +// 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 timestamp 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 timestamp leased above — making our open + // transaction stale. + require.NoError(t, c.StopZero(0)) + require.NoError(t, c.StartZero(0)) + require.NoError(t, c.HealthCheck(false)) + + 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) + + // Commit the stale transaction via the raw stub to observe the categorized status. Zero having + // a leader again is not enough: until the Alpha's own membership view catches up, its + // CommitOrAbort hook has no Zero connection and fails with codes.Unknown. Poll the commit + // itself until it reaches Zero and comes back Aborted. Retrying is safe because the stale + // verdict does not change once startTxnTs has moved past this start ts. + var st *status.Status + require.EventuallyWithT(t, func(collect *assert.CollectT) { + _, err := raw.CommitOrAbort(ctx, tc) + st = status.Convert(err) + assert.Equal(collect, codes.Aborted, st.Code(), + "committing a txn whose start ts predates the new leader must abort; got %v", st) + }, 60*time.Second, time.Second) + + require.True(t, strings.HasPrefix(st.Message(), "stale-startts: "), + "abort reason should be categorized as stale-startts; got %q", st.Message()) +} diff --git a/systest/mutations-and-queries/mutations_test.go b/systest/mutations-and-queries/mutations_test.go index 76e18ec4010..3a10d06a5a4 100644 --- a/systest/mutations-and-queries/mutations_test.go +++ b/systest/mutations-and-queries/mutations_test.go @@ -11,6 +11,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "math/rand" "sort" @@ -2055,7 +2056,7 @@ func (ssuite *SystestTestSuite) TestCountIndexConcurrentSetDelUIDList() { mu.SetNquads = []byte(fmt.Sprintf("<0x1> <%#x> .", id)) _, err := dg.NewTxn().Mutate(context.Background(), mu) - if err != nil && err != dgo.ErrAborted { + if err != nil && !errors.Is(err, dgo.ErrAborted) { require.Failf(t, "unable to inserted uid with err: %s", err.Error()) } if err == nil { // Successful insertion. @@ -2106,7 +2107,7 @@ func (ssuite *SystestTestSuite) TestCountIndexConcurrentSetDelUIDList() { mu.DelNquads = []byte(fmt.Sprintf("<0x1> <%#x> .", id)) _, err := dg.NewTxn().Mutate(context.Background(), mu) - if err != nil && err != dgo.ErrAborted { + if err != nil && !errors.Is(err, dgo.ErrAborted) { require.Failf(t, "unable to delete uid with err: %s", err.Error()) } if err == nil { // Successful deletion. @@ -2181,7 +2182,7 @@ func (ssuite *SystestTestSuite) TestCountIndexConcurrentSetDelScalarPredicate() mu.SetNquads = []byte(fmt.Sprintf("<0x1> \"name%d\" .", id)) _, err := dg.NewTxn().Mutate(context.Background(), mu) - if err != nil && err != dgo.ErrAborted { + if err != nil && !errors.Is(err, dgo.ErrAborted) { require.Fail(t, "unable to inserted uid with err: %s", err) } } @@ -2374,7 +2375,7 @@ func (ssuite *SystestTestSuite) TestReverseCountIndex() { mu.SetJson = []byte(`{"uid": "_:b", "friend": [{"uid": "` + id + `"}]}`) for i := 0; i < 10; i++ { _, err := dg.NewTxn().Mutate(context.Background(), mu) - if err == nil || err != dgo.ErrAborted { + if !errors.Is(err, dgo.ErrAborted) { break } } diff --git a/worker/mutation.go b/worker/mutation.go index 5efc6b2d1d7..200fb4367db 100644 --- a/worker/mutation.go +++ b/worker/mutation.go @@ -20,6 +20,8 @@ import ( "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" "github.com/dgraph-io/badger/v4" @@ -854,6 +856,21 @@ func CommitOverNetwork(ctx context.Context, tc *api.TxnContext) (uint64, error) h := hooks.GetHooks() tctx, err := h.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. Emit the same "Commit status" event as a + // non-error abort, plus the reason, so server-decided aborts stay visible in traces. + if st, ok := status.FromError(err); ok && st.Code() == codes.Aborted { + span.AddEvent("Commit status", trace.WithAttributes( + attribute.Int64("commitTs", 0), + attribute.Bool("committed", false), + attribute.String("abortReason", st.Message()))) + 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