Skip to content

Commit f514cfa

Browse files
authored
feat(errs): classify YARPC status errors (#641)
## Summary Intent: - Retry transient YARPC failures instead of dead-lettering them as unknown errors. - Preserve dependency attribution while distinguishing caller cancellation. Changes: - Classify typed YARPC statuses by retryability and origin. - Register the classifier with Stovepipe and document the mapping.
1 parent 6ed0c5a commit f514cfa

6 files changed

Lines changed: 233 additions & 2 deletions

File tree

platform/errs/README.md

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@ One operational consequence worth knowing before relying on any of this: **retry
8989

9090
## Adding a Backend-Specific Classifier
9191

92-
Backend classifiers live alongside the extension they classify, under `platform/errs/<backend>/`. The canonical examples are `platform/errs/mysql` (MySQL driver errors), `platform/errs/http` (rejected status codes and transport failures from clients built on `platform/http`), and `platform/errs/generic` (transport-agnostic concerns such as `context.Canceled`).
92+
Backend classifiers live alongside the extension they classify, under `platform/errs/<backend>/`. The canonical examples are `platform/errs/mysql` (MySQL driver errors), `platform/errs/http` (rejected status codes and transport failures from clients built on `platform/http`), `platform/errs/yarpc` (YARPC status codes), and `platform/errs/generic` (transport-agnostic concerns such as `context.Canceled`).
9393

9494
A classifier:
9595

@@ -124,20 +124,26 @@ import (
124124
genericerrs "github.com/uber/submitqueue/platform/errs/generic"
125125
httperrs "github.com/uber/submitqueue/platform/errs/http"
126126
mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
127+
yarpcerrs "github.com/uber/submitqueue/platform/errs/yarpc"
127128
)
128129

129130
c := consumer.New(logger, scope, registry,
130131
errs.NewClassifierProcessor(
131132
genericerrs.Classifier,
132133
httperrs.Classifier,
134+
yarpcerrs.Classifier,
133135
mysqlerrs.Classifier,
134136
),
135137
)
136138
```
137139

140+
Classifiers are not installed globally. A host that wants YARPC statuses classified adds `yarpcerrs.Classifier` to the `ErrorProcessor` at the boundary that consumes those errors, as above. This wiring covers outbound YARPC failures returned into that processor; inbound RPC handlers do not pass through it automatically and need their own transport middleware or mapper if they require the same classification.
141+
138142
`httperrs` precedes `mysqlerrs` for a reason worth knowing before reordering the list: the MySQL classifier treats any `net.Error` as retryable infra, and the `*url.Error` an HTTP client returns satisfies `net.Error`. Whichever runs first claims that node, so with the order reversed an HTTP transport failure is classified as a MySQL one — retryable either way, but no longer attributed to the dependency it came from. This is the cross-extension ambiguity `NewClassifierProcessor` documents as deferred; registration order is the workaround.
139143

140-
Tests follow the same shape: assert per-node behaviour against `Classifier.Classify(node)` directly, and assert end-to-end behaviour by running `errs.NewClassifierProcessor(Classifier).Process(err)` and checking the helpers (`IsRetryable`, `IsUserError`, …) on the result. See `platform/errs/mysql/mysql_test.go` and `platform/errs/generic/generic_test.go`.
144+
The YARPC classifier reads the typed status code rather than matching its rendered message. Cancellation is retryable caller-side infrastructure; transient or ambiguous server codes (`Unknown`, `DeadlineExceeded`, `ResourceExhausted`, `Aborted`, `Internal`, and `Unavailable`) are retryable dependency failures; request verdicts and permanent server failures are non-retryable dependency failures. A deadline may expire after a mutating RPC succeeded, so this classification relies on the repository-wide requirement that queue-driven operations are idempotent.
145+
146+
Tests follow the same shape: assert per-node behaviour against `Classifier.Classify(node)` directly, and assert end-to-end behaviour by running `errs.NewClassifierProcessor(Classifier).Process(err)` and checking the helpers (`IsRetryable`, `IsUserError`, …) on the result. See `platform/errs/mysql/mysql_test.go`, `platform/errs/yarpc/yarpc_test.go`, and `platform/errs/generic/generic_test.go`.
141147

142148
## Overriding Classification from a Controller
143149

platform/errs/yarpc/BUILD.bazel

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["yarpc.go"],
6+
importpath = "github.com/uber/submitqueue/platform/errs/yarpc",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//platform/errs:go_default_library",
10+
"@org_uber_go_yarpc//yarpcerrors:go_default_library",
11+
],
12+
)
13+
14+
go_test(
15+
name = "go_default_test",
16+
srcs = ["yarpc_test.go"],
17+
embed = [":go_default_library"],
18+
deps = [
19+
"//platform/errs:go_default_library",
20+
"@com_github_stretchr_testify//assert:go_default_library",
21+
"@org_uber_go_yarpc//yarpcerrors:go_default_library",
22+
],
23+
)

platform/errs/yarpc/yarpc.go

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Package yarpc classifies YARPC status errors by code.
16+
package yarpc
17+
18+
import (
19+
"github.com/uber/submitqueue/platform/errs"
20+
"go.uber.org/yarpc/yarpcerrors"
21+
)
22+
23+
// Classifier is the canonical YARPC error classifier.
24+
var Classifier errs.Classifier = classifier{}
25+
26+
type classifier struct{}
27+
28+
// yarpcError is YARPC's single-node status contract for transport-specific
29+
// errors. The classifier processor still owns traversal to reach this node.
30+
type yarpcError interface {
31+
YARPCError() *yarpcerrors.Status
32+
}
33+
34+
func (classifier) Classify(err error) errs.Verdict {
35+
var status *yarpcerrors.Status
36+
switch e := err.(type) {
37+
case *yarpcerrors.Status:
38+
status = e
39+
case yarpcError:
40+
status = e.YARPCError()
41+
default:
42+
return errs.Unknown
43+
}
44+
if status == nil {
45+
return errs.Unknown
46+
}
47+
48+
switch status.Code() {
49+
case yarpcerrors.CodeCancelled:
50+
// Cancellation belongs to the caller's operation rather than to the
51+
// downstream service, matching generic's context.Canceled verdict.
52+
return errs.InfraRetryable
53+
54+
case yarpcerrors.CodeUnknown,
55+
yarpcerrors.CodeDeadlineExceeded,
56+
yarpcerrors.CodeResourceExhausted,
57+
yarpcerrors.CodeAborted,
58+
yarpcerrors.CodeInternal,
59+
yarpcerrors.CodeUnavailable:
60+
return errs.InfraDependencyRetryable
61+
62+
case yarpcerrors.CodeInvalidArgument,
63+
yarpcerrors.CodeNotFound,
64+
yarpcerrors.CodeAlreadyExists,
65+
yarpcerrors.CodePermissionDenied,
66+
yarpcerrors.CodeFailedPrecondition,
67+
yarpcerrors.CodeOutOfRange,
68+
yarpcerrors.CodeUnimplemented,
69+
yarpcerrors.CodeDataLoss,
70+
yarpcerrors.CodeUnauthenticated:
71+
return errs.InfraDependency
72+
73+
default:
74+
return errs.Unknown
75+
}
76+
}

platform/errs/yarpc/yarpc_test.go

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
// Copyright (c) 2026 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package yarpc
16+
17+
import (
18+
"errors"
19+
"fmt"
20+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
"github.com/uber/submitqueue/platform/errs"
24+
"go.uber.org/yarpc/yarpcerrors"
25+
)
26+
27+
type customYARPCError struct {
28+
status *yarpcerrors.Status
29+
}
30+
31+
func (e customYARPCError) Error() string {
32+
return e.status.Error()
33+
}
34+
35+
func (e customYARPCError) YARPCError() *yarpcerrors.Status {
36+
return e.status
37+
}
38+
39+
func TestClassifier_StatusCodes(t *testing.T) {
40+
tests := []struct {
41+
name string
42+
code yarpcerrors.Code
43+
want errs.Verdict
44+
}{
45+
{name: "cancelled", code: yarpcerrors.CodeCancelled, want: errs.InfraRetryable},
46+
{name: "unknown", code: yarpcerrors.CodeUnknown, want: errs.InfraDependencyRetryable},
47+
{name: "deadline exceeded", code: yarpcerrors.CodeDeadlineExceeded, want: errs.InfraDependencyRetryable},
48+
{name: "resource exhausted", code: yarpcerrors.CodeResourceExhausted, want: errs.InfraDependencyRetryable},
49+
{name: "aborted", code: yarpcerrors.CodeAborted, want: errs.InfraDependencyRetryable},
50+
{name: "internal", code: yarpcerrors.CodeInternal, want: errs.InfraDependencyRetryable},
51+
{name: "unavailable", code: yarpcerrors.CodeUnavailable, want: errs.InfraDependencyRetryable},
52+
{name: "invalid argument", code: yarpcerrors.CodeInvalidArgument, want: errs.InfraDependency},
53+
{name: "not found", code: yarpcerrors.CodeNotFound, want: errs.InfraDependency},
54+
{name: "already exists", code: yarpcerrors.CodeAlreadyExists, want: errs.InfraDependency},
55+
{name: "permission denied", code: yarpcerrors.CodePermissionDenied, want: errs.InfraDependency},
56+
{name: "failed precondition", code: yarpcerrors.CodeFailedPrecondition, want: errs.InfraDependency},
57+
{name: "out of range", code: yarpcerrors.CodeOutOfRange, want: errs.InfraDependency},
58+
{name: "unimplemented", code: yarpcerrors.CodeUnimplemented, want: errs.InfraDependency},
59+
{name: "data loss", code: yarpcerrors.CodeDataLoss, want: errs.InfraDependency},
60+
{name: "unauthenticated", code: yarpcerrors.CodeUnauthenticated, want: errs.InfraDependency},
61+
{name: "ok", code: yarpcerrors.CodeOK, want: errs.Unknown},
62+
{name: "unrecognized code", code: yarpcerrors.Code(99), want: errs.Unknown},
63+
}
64+
65+
for _, tt := range tests {
66+
t.Run(tt.name, func(t *testing.T) {
67+
assert.Equal(t, tt.want, Classifier.Classify(yarpcerrors.Newf(tt.code, "rpc failed")))
68+
})
69+
}
70+
}
71+
72+
func TestClassifier_Unknown(t *testing.T) {
73+
tests := []struct {
74+
name string
75+
err error
76+
}{
77+
{name: "wrapped status", err: fmt.Errorf("call failed: %w", yarpcerrors.DeadlineExceededErrorf("late"))},
78+
{name: "plain error", err: errors.New("anything")},
79+
{name: "nil", err: nil},
80+
}
81+
82+
for _, tt := range tests {
83+
t.Run(tt.name, func(t *testing.T) {
84+
assert.Equal(t, errs.Unknown, Classifier.Classify(tt.err))
85+
})
86+
}
87+
}
88+
89+
func TestClassifier_TransportSpecificYARPCError(t *testing.T) {
90+
err := customYARPCError{status: yarpcerrors.Newf(yarpcerrors.CodeUnavailable, "down")}
91+
assert.Equal(t, errs.InfraDependencyRetryable, Classifier.Classify(err))
92+
}
93+
94+
func TestClassifier_AppliedViaProcessor(t *testing.T) {
95+
processor := errs.NewClassifierProcessor(Classifier)
96+
97+
t.Run("wrapped deadline is a retryable dependency error", func(t *testing.T) {
98+
err := fmt.Errorf("set ref: %w", yarpcerrors.DeadlineExceededErrorf("context deadline exceeded"))
99+
out := processor.Process(err)
100+
assert.True(t, errs.IsRetryable(out))
101+
assert.True(t, errs.IsDependencyError(out))
102+
})
103+
104+
t.Run("wrapped invalid argument is a non-retryable dependency error", func(t *testing.T) {
105+
err := fmt.Errorf("set ref: %w", yarpcerrors.InvalidArgumentErrorf("bad ref"))
106+
out := processor.Process(err)
107+
assert.False(t, errs.IsRetryable(out))
108+
assert.True(t, errs.IsDependencyError(out))
109+
})
110+
111+
t.Run("cancelled is retryable without dependency attribution", func(t *testing.T) {
112+
out := processor.Process(yarpcerrors.CancelledErrorf("caller cancelled"))
113+
assert.True(t, errs.IsRetryable(out))
114+
assert.False(t, errs.IsDependencyError(out))
115+
})
116+
117+
t.Run("a controller verdict wins over the classifier", func(t *testing.T) {
118+
err := errs.NewDependencyError(yarpcerrors.UnavailableErrorf("down"))
119+
out := processor.Process(err)
120+
assert.Same(t, err, out)
121+
assert.False(t, errs.IsRetryable(out))
122+
})
123+
}

service/stovepipe/server/BUILD.bazel

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ go_library(
1313
"//platform/errs/generic:go_default_library",
1414
"//platform/errs/http:go_default_library",
1515
"//platform/errs/mysql:go_default_library",
16+
"//platform/errs/yarpc:go_default_library",
1617
"//platform/extension/consumergate/noop:go_default_library",
1718
"//platform/extension/counter:go_default_library",
1819
"//platform/extension/hook:go_default_library",

service/stovepipe/server/main.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ import (
3535
genericerrs "github.com/uber/submitqueue/platform/errs/generic"
3636
httperrs "github.com/uber/submitqueue/platform/errs/http"
3737
mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
38+
yarpcerrs "github.com/uber/submitqueue/platform/errs/yarpc"
3839
consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
3940
"github.com/uber/submitqueue/platform/extension/counter"
4041
hookext "github.com/uber/submitqueue/platform/extension/hook"
@@ -284,6 +285,7 @@ func run() error {
284285
errs.NewClassifierProcessor(
285286
genericerrs.Classifier,
286287
httperrs.Classifier,
288+
yarpcerrs.Classifier,
287289
mysqlerrs.Classifier,
288290
),
289291
consumergatenoop.New(),

0 commit comments

Comments
 (0)