go.temporal.io/server/common/nexus/failure.go
441 LOC · 185 covered · 256 uncovered · 53 ranges · 86 concepts · 28 introducers · 44 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package nexus
2
3
import (
4
"context"
5
"encoding/base64"
6
"encoding/json"
7
"fmt"
8
"net/http"
9
"sync/atomic"
10
11
"github.com/nexus-rpc/sdk-go/nexus"
12
commonpb "go.temporal.io/api/common/v1"
13
enumspb "go.temporal.io/api/enums/v1"
14
failurepb "go.temporal.io/api/failure/v1"
15
nexuspb "go.temporal.io/api/nexus/v1"
16
"go.temporal.io/api/serviceerror"
17
"google.golang.org/grpc/codes"
18
"google.golang.org/grpc/status"
19
"google.golang.org/protobuf/encoding/protojson"
20
"google.golang.org/protobuf/proto"
21
)
22
23
const (
24
// FailureSourceHeaderName is the header used to indicate from where the Nexus failure originated.
25
FailureSourceHeaderName = "Temporal-Nexus-Failure-Source"
26
// FailureSourceWorker indicates the failure originated from outside the server (e.g. bad request or on the Nexus worker).
27
FailureSourceWorker = "worker"
28
)
29
30
type failureSourceContextKeyType struct{}
31
32
var FailureSourceContextKey = failureSourceContextKeyType{}
33
34
func SetFailureSourceOnContext(ctx context.Context, response *http.Response) {
35
if response == nil || response.Header == nil {
36
return
37
}
38
39
failureSourceHeader := response.Header.Get(FailureSourceHeaderName)
40
if failureSourceHeader == "" {
41
return
42
}
43
44
failureSourceContext := ctx.Value(FailureSourceContextKey)
45
if failureSourceContext == nil {
46
return
47
}
48
49
if val, ok := failureSourceContext.(*atomic.Value); ok {
50
val.Store(failureSourceHeader)
51
}
52
}
53
54
var failureTypeString = string((&failurepb.Failure{}).ProtoReflect().Descriptor().FullName())
55
56
// ProtoFailureToNexusFailure converts a proto Nexus Failure to a Nexus SDK Failure.
57
func ProtoFailureToNexusFailure(failure *nexuspb.Failure) nexus.Failure {
58
nf := nexus.Failure{
59
Message: failure.GetMessage(),
60
StackTrace: failure.GetStackTrace(),
61
Metadata: failure.GetMetadata(),
62
Details: failure.GetDetails(),
63
}
64
if failure.GetCause() != nil {
65
cause := ProtoFailureToNexusFailure(failure.GetCause())
66
nf.Cause = &cause
67
}
68
return nf
69
}
70
71
// NexusFailureToProtoFailure converts a Nexus SDK Failure to a proto Nexus Failure.
72
// Always returns a non-nil value.
73
func NexusFailureToProtoFailure(failure nexus.Failure) *nexuspb.Failure {
74
pf := &nexuspb.Failure{
75
Message: failure.Message,
76
Metadata: failure.Metadata,
77
Details: failure.Details,
78
StackTrace: failure.StackTrace,
79
}
80
if failure.Cause != nil {
81
pf.Cause = NexusFailureToProtoFailure(*failure.Cause)
82
}
83
return pf
84
}
85
86
type serializedHandlerError struct {
87
Type string `json:"type,omitempty"`
88
RetryableOverride *bool `json:"retryableOverride,omitempty"`
89
// Bytes as base64 encoded string.
90
EncodedAttributes string `json:"encodedAttributes,omitempty"`
91
}
92
93
// TemporalFailureToNexusFailure converts an API proto Failure to a Nexus SDK Failure setting the metadata "type" field to
94
// the proto fullname of the temporal API Failure message or the standard Nexus SDK failure types.
95
// Returns an error if the failure cannot be converted.
96
//
97
// Compared to TemporalFailureToNexusFailureInPlace, this function clones the failure before converting it to avoid
98
// mutating a shared proto.
99
>
func TemporalFailureToNexusFailure(failure *failurepb.Failure) (nexus.Failure, error) {
failure.go ×1
100
>
return TemporalFailureToNexusFailureInPlace(proto.Clone(failure).(*failurepb.Failure))
101
>
}
102
103
// TemporalFailureToNexusFailureInPlace converts an API proto Failure to a Nexus SDK Failure setting the metadata
104
// "type" field to the proto fullname of the temporal API Failure message or the standard Nexus SDK failure types.
105
// Returns an error if the failure cannot be converted.
106
//
107
// Compared to TemporalFailureToNexusFailure, this function converts the failure in place to avoid cloning. This
108
// is only safe for callers who exclusively own the failure.
109
>
func TemporalFailureToNexusFailureInPlace(failure *failurepb.Failure) (nexus.Failure, error) {
failure.go ×2
110
>
var causep *nexus.Failure
111
>
if failure.GetCause() != nil {
113
>
var err error
114
>
cause, err = TemporalFailureToNexusFailureInPlace(failure.GetCause())
115
>
if err != nil {
116
return nexus.Failure{}, err
117
}
119
}
120
123
>
var encodedAttributes string
124
>
if failure.EncodedAttributes != nil {
126
>
if err != nil {
127
return nexus.Failure{}, fmt.Errorf("failed to deserialize HandlerError attributes: %w", err)
128
}
130
}
132
>
// nolint:exhaustive,revive // There are only two valid values other than unspecified.
133
>
switch info.NexusHandlerFailureInfo.GetRetryBehavior() {
135
>
val := true
136
>
retryableOverride = &val
138
>
val := false
139
>
retryableOverride = &val
140
}
141
143
>
Type: info.NexusHandlerFailureInfo.GetType(),
144
>
RetryableOverride: retryableOverride,
145
>
EncodedAttributes: encodedAttributes,
146
>
}
147
>
148
>
details, err := json.Marshal(handlerError)
149
>
if err != nil {
150
return nexus.Failure{}, err
151
}
153
>
Message: failure.GetMessage(),
154
>
StackTrace: failure.GetStackTrace(),
155
>
Metadata: map[string]string{
156
>
"type": "nexus.HandlerError",
157
>
},
158
>
Details: details,
159
>
Cause: causep,
160
>
}, nil
161
}
162
// Unset message and stack trace so it's not serialized in the details.
164
>
message, failure.Message = failure.Message, ""
165
>
var stackTrace string
166
>
stackTrace, failure.StackTrace = failure.StackTrace, ""
167
>
168
>
data, err := protojson.Marshal(failure)
169
>
failure.Message = message
170
>
failure.StackTrace = stackTrace
171
>
if err != nil {
172
return nexus.Failure{}, err
173
}
174
176
>
Message: failure.GetMessage(),
177
>
StackTrace: failure.GetStackTrace(),
178
>
Metadata: map[string]string{
179
>
"type": failureTypeString,
180
>
},
181
>
Details: data,
182
>
Cause: causep,
183
>
}, nil
184
}
185
186
// NexusFailureToTemporalFailure converts a Nexus Failure to an API proto Failure.
187
// If the failure metadata "type" field is set to the fullname of the temporal API Failure message, the failure is
188
// reconstructed using protojson.Unmarshal on the failure details field. Otherwise, the failure is reconstructed
189
// based on the known Nexus SDK failure types.
190
// Returns an error if the failure cannot be converted.
191
// nolint:revive // cognitive-complexity is high but justified to keep each case together
192
>
func NexusFailureToTemporalFailure(f nexus.Failure) (*failurepb.Failure, error) {
failure.go ×3
193
>
apiFailure := &failurepb.Failure{
194
>
Message: f.Message,
195
>
StackTrace: f.StackTrace,
196
>
}
197
>
198
>
if f.Metadata != nil {
201
>
opts := protojson.UnmarshalOptions{DiscardUnknown: true}
202
>
if err := opts.Unmarshal(f.Details, apiFailure); err != nil {
203
return nil, err
204
}
205
// Restore these fields as they are not included in the marshalled failure.
207
>
apiFailure.StackTrace = f.StackTrace
209
>
// Special case for OperationError that adapts from Nexus semantics to Temporal semantics.
210
>
// Note that Temporal -> Temporal doesn't go through this code path, operation errors are always used as empty
211
>
// wrappers for an underlying causes.
212
>
var operationError *nexus.OperationError
213
>
err := json.Unmarshal(f.Details, &operationError)
214
>
if err != nil {
215
return nil, fmt.Errorf("failed to deserialize OperationError: %w", err)
216
}
219
>
apiFailure.FailureInfo = &failurepb.Failure_CanceledFailureInfo{
220
>
CanceledFailureInfo: &failurepb.CanceledFailureInfo{},
221
>
}
223
>
// Failed operation errors are represented as non-retryable ApplicationFailure in Temporal.
failure.go ×1
224
>
apiFailure.FailureInfo = &failurepb.Failure_ApplicationFailureInfo{
225
>
ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
226
>
NonRetryable: true,
227
>
Type: "OperationError",
228
>
},
229
>
}
230
>
}
232
>
var se serializedHandlerError
233
>
err := json.Unmarshal(f.Details, &se)
234
>
if err != nil {
235
return nil, fmt.Errorf("failed to deserialize HandlerError: %w", err)
236
}
238
>
if se.RetryableOverride == nil {
244
>
}
246
>
NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{
247
>
Type: se.Type,
248
>
RetryBehavior: retryBehavior,
249
>
},
250
>
}
251
>
if len(se.EncodedAttributes) > 0 {
253
>
if err != nil {
254
return nil, fmt.Errorf("failed to decode base64 HandlerError attributes: %w", err)
255
}
257
>
if err := protojson.Unmarshal(decoded, apiFailure.EncodedAttributes); err != nil {
258
return nil, fmt.Errorf("failed to deserialize HandlerError attributes: %w", err)
259
}
260
}
262
>
// We don't recognize this type, convert to a generic ApplicationFailure and preserve the original Nexus failure
263
>
// as serialized details.
264
>
applicationFailureInfo, err := nexusFailureMetadataToApplicationFailureInfo(f)
265
>
if err != nil {
266
return nil, fmt.Errorf("failed to serialize Nexus failure: %w", err)
267
}
269
}
271
>
// We don't recognize this type, convert to a generic ApplicationFailure and preserve the original Nexus failure as
failure.go ×2
272
>
// serialized details.
273
>
applicationFailureInfo, err := nexusFailureMetadataToApplicationFailureInfo(f)
274
>
if err != nil {
275
return nil, fmt.Errorf("failed to serialize Nexus failure: %w", err)
276
}
278
}
279
282
>
apiFailure.Cause, err = NexusFailureToTemporalFailure(*f.Cause)
283
>
if err != nil {
284
return nil, err
285
}
286
}
288
}
289
290
>
func nexusFailureMetadataToApplicationFailureInfo(failure nexus.Failure) (*failurepb.Failure_ApplicationFailureInfo, error) {
failure.go ×3
291
>
var payloads *commonpb.Payloads
292
>
if len(failure.Metadata) > 0 || len(failure.Details) > 0 {
293
>
// Delete before serializing (note the failure here is passed by value).
294
>
failure.Message = ""
295
>
failure.StackTrace = ""
296
>
data, err := json.Marshal(failure)
297
>
if err != nil {
298
return nil, err
299
}
301
>
Payloads: []*commonpb.Payload{
302
>
{
303
>
Metadata: map[string][]byte{
304
>
"encoding": []byte("json/plain"),
305
>
},
306
>
Data: data,
307
>
},
308
>
},
309
>
}
310
}
312
>
ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
313
>
Type: "NexusFailure",
314
>
Details: payloads,
315
>
},
316
>
}, nil
317
}
318
319
// ConvertGRPCError converts either a serviceerror or a gRPC status error into a Nexus HandlerError if possible.
320
// If exposeDetails is true, the error message from the given error is exposed in the converted HandlerError, otherwise,
321
// a default message with minimal information is attached to the returned error.
322
// Roughly taken from https://github.com/googleapis/googleapis/blob/master/google/rpc/code.proto
323
// and
324
// https://github.com/grpc-ecosystem/grpc-gateway/blob/a7cf811e6ffabeaddcfb4ff65602c12671ff326e/runtime/errors.go#L56.
326
>
var st *status.Status
327
>
stGetter, ok := err.(interface{ Status() *status.Status })
328
>
if ok {
329
>
st = stGetter.Status()
330
>
} else {
331
st, ok = status.FromError(err)
332
if !ok {
333
// The Nexus SDK will translate this into an internal server error and will not expose the error details.
334
return err
335
}
336
}
337
339
>
340
>
switch st.Code() {
341
>
case codes.AlreadyExists, codes.InvalidArgument, codes.FailedPrecondition, codes.OutOfRange:
failure.go ×2
342
>
if !exposeDetails {
344
>
}
346
>
Type: nexus.HandlerErrorTypeBadRequest,
347
>
Message: errMessage,
348
>
}
349
case codes.Aborted, codes.Unavailable:
350
if !exposeDetails {
351
errMessage = "service unavailable"
352
}
353
return &nexus.HandlerError{
354
Type: nexus.HandlerErrorTypeUnavailable,
355
Message: errMessage,
356
}
357
case codes.Canceled:
358
// TODO: This should have a different status code (e.g. 499 which is semi standard but not supported by nexus).
359
// The important thing is that the request is retryable, internal serves that purpose.
360
if !exposeDetails {
361
errMessage = "canceled"
362
}
363
return &nexus.HandlerError{
364
Type: nexus.HandlerErrorTypeInternal,
365
Message: errMessage,
366
}
367
case codes.DataLoss, codes.Internal, codes.Unknown:
368
if !exposeDetails {
369
errMessage = "internal error"
370
}
371
return &nexus.HandlerError{
372
Type: nexus.HandlerErrorTypeInternal,
373
Message: errMessage,
374
}
375
case codes.Unauthenticated:
376
if !exposeDetails {
377
errMessage = "authentication failed"
378
}
379
return &nexus.HandlerError{
380
Type: nexus.HandlerErrorTypeUnauthenticated,
381
Message: errMessage,
382
}
383
case codes.PermissionDenied:
384
if !exposeDetails {
385
errMessage = "permission denied"
386
}
387
return &nexus.HandlerError{
388
Type: nexus.HandlerErrorTypeUnauthorized,
389
Message: errMessage,
390
}
391
case codes.NotFound:
392
if !exposeDetails {
393
errMessage = "not found"
394
}
395
return &nexus.HandlerError{
396
Type: nexus.HandlerErrorTypeNotFound,
397
Message: errMessage,
398
}
400
>
if !exposeDetails {
402
>
}
404
>
Type: nexus.HandlerErrorTypeResourceExhausted,
405
>
Message: errMessage,
406
>
}
407
case codes.Unimplemented:
408
if !exposeDetails {
409
errMessage = "not implemented"
410
}
411
return &nexus.HandlerError{
412
Type: nexus.HandlerErrorTypeNotImplemented,
413
Message: errMessage,
414
}
415
case codes.DeadlineExceeded:
416
if !exposeDetails {
417
errMessage = "request timeout"
418
}
419
return &nexus.HandlerError{
420
Type: nexus.HandlerErrorTypeUpstreamTimeout,
421
Message: errMessage,
422
}
423
case codes.OK:
424
return nil
425
}
426
if !exposeDetails {
427
return &nexus.HandlerError{
428
Type: nexus.HandlerErrorTypeInternal,
429
Message: "internal error",
430
}
431
}
432
// Let the nexus SDK handle this for us (log and convert to an internal error).
433
return err
434
}
435
436
func AdaptAuthorizeError(permissionDeniedError *serviceerror.PermissionDenied) error {
437
if permissionDeniedError.Reason != "" {
438
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeUnauthorized, "permission denied: %s", permissionDeniedError.Reason)
439
}
440
return nexus.NewHandlerErrorf(nexus.HandlerErrorTypeUnauthorized, "permission denied")
441
}