diff --git a/src/dbnode/client/errors.go b/src/dbnode/client/errors.go index 896f1998b4..56ec98e7c2 100644 --- a/src/dbnode/client/errors.go +++ b/src/dbnode/client/errors.go @@ -56,8 +56,13 @@ func IsBadRequestError(err error) bool { // IsConsistencyResultError determines if the error is a consistency result error. func IsConsistencyResultError(err error) bool { - _, ok := err.(consistencyResultErr) - return ok + for err != nil { + if _, ok := err.(consistencyResultErr); ok { //nolint:errorlint + return true + } + err = xerrors.InnerError(err) + } + return false } // NumResponded returns how many nodes responded for a given error @@ -117,8 +122,8 @@ func isHostNotAvailableError(err error) bool { type consistencyResultError interface { error + xerrors.ContainedError - InnerError() error numResponded() int numSuccess() int } diff --git a/src/dbnode/client/fetch_tagged_results_accumulator.go b/src/dbnode/client/fetch_tagged_results_accumulator.go index 32f3b3aa2d..19c8aecf28 100644 --- a/src/dbnode/client/fetch_tagged_results_accumulator.go +++ b/src/dbnode/client/fetch_tagged_results_accumulator.go @@ -205,8 +205,13 @@ func (accum *fetchTaggedResultAccumulator) accumulatedResult( doneAccumulating := true // NB(r): Use new renamed error to keep the underlying error // (invalid/retryable) type. + enqueued := accum.topoMap.HostsLen() + responded := enqueued + consistencyErr := newConsistencyResultError(accum.consistencyLevel, enqueued, responded, + accum.errors) err := fmt.Errorf("unable to satisfy consistency requirements: shards=%d, err=%v", - accum.numShardsPending, accum.errors) + accum.numShardsPending, consistencyErr) + err = xerrors.NewRenamedError(consistencyErr, err) for i := range accum.errors { if IsBadRequestError(accum.errors[i]) { err = xerrors.NewInvalidParamsError(err) diff --git a/src/dbnode/client/fetch_tagged_results_accumulator_merge_test.go b/src/dbnode/client/fetch_tagged_results_accumulator_merge_test.go index 0981bf9685..92f0ceb138 100644 --- a/src/dbnode/client/fetch_tagged_results_accumulator_merge_test.go +++ b/src/dbnode/client/fetch_tagged_results_accumulator_merge_test.go @@ -343,6 +343,10 @@ func (tm testFetchStateWorkflow) run() fetchTaggedResultAccumulator { } assert.Equal(tm.t, s.expectedDone, done, fmt.Sprintf("i=%d, step=%+v", i, s)) assert.Equal(tm.t, s.expectedErr, err != nil, fmt.Sprintf("i=%d, step=%+v, err=%v", i, s, err)) + if err != nil { + assert.True(tm.t, IsConsistencyResultError(err), + fmt.Sprintf("i=%d, step=%+v, expected consistency result error", i, s)) + } } return accum }