Vihren Atlas

Atlas › Test

TestBadRequestCompletion

go.temporal.io/server/common/nexus/nexusrpc/TestBadRequestCompletion

Package
go.temporal.io/server/common/nexus/nexusrpc
Suite / test hierarchy
TestBadRequestCompletion
Test
TestBadRequestCompletion
Introduced at
completion.go ×1 Joint frontier
Covered ranges
84
Covered lines
222
Covered files
5

Covered source

Expand a file to inspect source; the > gutter marks covered lines.

go.temporal.io/server/common/nexus/nexusrpc/completion.go 98 covered LOC · 38 ranges

Open complete file

33 // NewCompletionHTTPClient constructs a [CompletionHTTPClient] from given options for sending Nexus operation completion
34 // callbacks via HTTP.
35 > func NewCompletionHTTPClient(options CompletionHTTPClientOptions) *CompletionHTTPClient { completion.go
36 > if options.HTTPCaller == nil {
37 > options.HTTPCaller = http.DefaultClient.Do completion.go
38 > }
39 > if options.Serializer == nil { completion.go
40 > options.Serializer = nexus.DefaultSerializer() completion.go
41 > }
42 > if options.FailureConverter == nil { completion.go
43 > options.FailureConverter = DefaultFailureConverter() completion.go
44 > }
45 > return &CompletionHTTPClient{ completion.go
46 > baseHTTPClient: baseHTTPClient{
47 > httpCaller: options.HTTPCaller,
48 > serializer: options.Serializer,
49 > failureConverter: options.FailureConverter,
50 > },
51 > }
52 }
53
54 // CompleteOperation sends a completion callback for a Nexus operation to the given URL with the given completion details.
55 > func (c *CompletionHTTPClient) CompleteOperation(ctx context.Context, url string, completion CompleteOperationOptions) error { completion.go
56 > httpReq, err := http.NewRequestWithContext(ctx, "POST", url, nil)
57 > if err != nil {
58 return err
59 }
60 > if err := completion.applyToHTTPRequest(c, httpReq); err != nil { completion.go
61 return err
62 }
63
64 > response, err := c.httpCaller(httpReq) completion.go
65 > if err != nil {
66 return err
67 }
68
69 > if response.StatusCode >= 200 && response.StatusCode < 300 { completion.go
70 // Body is not read but should be discarded to keep the underlying TCP connection alive.
71 // Just in case something unexpected happens while discarding or closing the body,
79 }
80
81 > body, err := readAndReplaceBody(response) completion.go
82 > if err != nil {
83 return err
84 }
85
86 > return c.bestEffortHandlerErrorFromResponse(response, body) completion.go
87 }
88
111 // nolint:revive // This method is long but it's more readable to keep the logic in one place since it's all related to
112 // constructing the completion request.
113 > func (c CompleteOperationOptions) applyToHTTPRequest(cc *CompletionHTTPClient, request *http.Request) error { completion.go
114 > if request.Header == nil {
115 request.Header = make(http.Header)
116 }
117
118 // Set the body and operation state based on whether the completion is successful or not.
119 > if c.Error != nil { completion.go
120 failure, err := cc.failureConverter.ErrorToFailure(c.Error)
121 if err != nil {
134 request.Header.Set(headerOperationState, string(c.Error.State))
135 request.Header.Set("Content-Type", contentTypeJSON)
136 > } else { completion.go
137 > reader, ok := c.Result.(*nexus.Reader)
138 > if !ok {
139 > content, ok := c.Result.(*nexus.Content)
140 > if !ok {
141 > var err error
142 > content, err = cc.serializer.Serialize(c.Result)
143 > if err != nil {
144 return err
145 }
146 }
147 > request.ContentLength = int64(len(content.Data)) completion.go
148 > reader = &nexus.Reader{
149 > Header: content.Header,
150 > ReadCloser: io.NopCloser(bytes.NewReader(content.Data)),
151 > }
152 }
153 > if reader.Header != nil { completion.go
154 > addContentHeaderToHTTPHeader(reader.Header, request.Header)
155 > }
156 > request.Body = reader.ReadCloser
157 > request.Header.Set(headerOperationState, string(nexus.OperationStateSucceeded))
158 }
159
160 > if c.Header != nil { completion.go
161 addNexusHeaderToHTTPHeader(c.Header, request.Header)
162 }
163 > if c.Header.Get(headerUserAgent) == "" { completion.go
164 > request.Header.Set(headerUserAgent, userAgent)
165 > }
166 > if c.Header.Get(nexus.HeaderOperationToken) == "" && c.OperationToken != "" {
167 request.Header.Set(nexus.HeaderOperationToken, c.OperationToken)
168 }
169 > if c.Header.Get(headerOperationStartTime) == "" && !c.StartTime.IsZero() { completion.go
170 request.Header.Set(headerOperationStartTime, c.StartTime.Format(http.TimeFormat))
171 }
172 > if c.Header.Get(headerOperationCloseTime) == "" && !c.CloseTime.IsZero() { completion.go
173 request.Header.Set(headerOperationCloseTime, marshalTimestamp(c.CloseTime))
174 }
175 > if c.Header.Get(headerLink) == "" { completion.go
176 > if err := addLinksToHTTPHeader(c.Links, request.Header); err != nil {
177 return err
178 }
179 }
180 > return nil completion.go
181 }
182
228 }
229
230 > func (h *completionHTTPHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) { completion.go
231 > ctx := request.Context()
232 > completion := CompletionRequest{
233 > State: nexus.OperationState(request.Header.Get(headerOperationState)),
234 > OperationToken: request.Header.Get(nexus.HeaderOperationToken),
235 > HTTPRequest: request,
236 > }
237 > if startTimeHeader := request.Header.Get(headerOperationStartTime); startTimeHeader != "" {
238 var parseTimeErr error
239 if completion.StartTime, parseTimeErr = http.ParseTime(startTimeHeader); parseTimeErr != nil {
242 }
243 }
244 > if closeTimeHeader := request.Header.Get(headerOperationCloseTime); closeTimeHeader != "" { completion.go
245 var parseTimeErr error
246 if completion.CloseTime, parseTimeErr = unmarshalTimestamp(closeTimeHeader); parseTimeErr != nil {
249 }
250 }
251 > var decodeErr error completion.go
252 > if completion.Links, decodeErr = getLinksFromHeader(request.Header); decodeErr != nil {
253 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to decode links from request headers"))
254 return
255 }
256 > switch completion.State { completion.go
257 case nexus.OperationStateFailed, nexus.OperationStateCanceled:
258 if !isMediaTypeJSON(request.Header.Get("Content-Type")) {
290 completion.Error = opErr
291 }
292 > case nexus.OperationStateSucceeded: completion.go
293 > completion.Result = nexus.NewLazyValue(
294 > h.options.Serializer,
295 > &nexus.Reader{
296 > ReadCloser: request.Body,
297 > Header: prefixStrippedHTTPHeaderToNexusHeader(request.Header, "content-"),
298 > },
299 > )
300 default:
301 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid request operation state: %q", completion.State))
302 return
303 }
304 > if err := h.options.Handler.CompleteOperation(ctx, &completion); err != nil { completion.go
305 > h.WriteFailure(writer, request, err) completion.go
306 > }
307 }
308
309 // NewCompletionHTTPHandler constructs an [http.Handler] from given options for handling operation completion requests.
310 > func NewCompletionHTTPHandler(options CompletionHandlerOptions) http.Handler { completion.go
311 > if options.Logger == nil {
312 > options.Logger = slog.Default() completion.go
313 > }
314 > if options.Serializer == nil { completion.go
315 > options.Serializer = nexus.DefaultSerializer() completion.go
316 > }
317 > if options.FailureConverter == nil { completion.go
318 > options.FailureConverter = DefaultFailureConverter() completion.go
319 > }
320 > return &completionHTTPHandler{ completion.go
321 > options: options,
322 > BaseHTTPHandler: BaseHTTPHandler{
323 > Logger: options.Logger,
324 > FailureConverter: options.FailureConverter,
325 > },
326 > }
327 }
go.temporal.io/server/common/nexus/nexusrpc/failure_converter.go 50 covered LOC · 16 ranges

Open complete file

30 }
31
32 > func (e serializedHandlerError) RetryBehavior() nexus.HandlerErrorRetryBehavior { failure_converter.go
33 > if e.RetryableOverride == nil {
34 > return nexus.HandlerErrorRetryBehaviorUnspecified failure_converter.go
35 > }
36 if *e.RetryableOverride {
37 return nexus.HandlerErrorRetryBehaviorRetryable
46 // ErrorToFailure implements FailureConverter.
47 // nolint:revive // Keeping all of the logic together for readability, even if it means the function is long.
48 > func (e knownErrorFailureConverter) ErrorToFailure(err error) (nexus.Failure, error) { failure_converter.go
49 > if err == nil {
50 return nexus.Failure{}, nil
51 }
52 // NOTE: not using errors.Unwrap here we are intentionally only supporting unwrapping known errors.
53 > switch typedErr := err.(type) { failure_converter.go
54 case *nexus.FailureError:
55 f := typedErr.Failure
65 }
66 return f, nil
67 > case *nexus.HandlerError: failure_converter.go
68 > if typedErr.OriginalFailure != nil {
69 return *typedErr.OriginalFailure, nil
70 }
71 > data := serializedHandlerError{ failure_converter.go
72 > Type: string(typedErr.Type),
73 > RetryableOverride: retryBehaviorAsOptionalBool(typedErr),
74 > }
75 > var details []byte
76 > details, err := json.Marshal(data)
77 > if err != nil {
78 return nexus.Failure{}, err
79 }
80 > f := nexus.Failure{ failure_converter.go
81 > Message: typedErr.Message,
82 > StackTrace: typedErr.StackTrace,
83 > Metadata: map[string]string{
84 > "type": "nexus.HandlerError",
85 > },
86 > Details: details,
87 > }
88 >
89 > if typedErr.Cause != nil {
90 c, err := e.ErrorToFailure(typedErr.Cause)
91 if err != nil {
94 f.Cause = &c
95 }
96 > return f, nil failure_converter.go
97 case *nexus.OperationError:
98 if typedErr.OriginalFailure != nil {
132 // FailureToError implements FailureConverter.
133 // nolint:revive // Keeping all of the logic together for readability, even if it means the function is long.
134 > func (e knownErrorFailureConverter) FailureToError(f nexus.Failure) (error, error) { failure_converter.go
135 > if f.Metadata != nil {
136 > switch f.Metadata["type"] { failure_converter.go
137 > case "nexus.HandlerError": failure_converter.go
138 > var se serializedHandlerError
139 > err := json.Unmarshal(f.Details, &se)
140 > if err != nil {
141 return nil, fmt.Errorf("failed to deserialize HandlerError: %w", err)
142 }
143 > he := &nexus.HandlerError{ failure_converter.go
144 > Message: f.Message,
145 > StackTrace: f.StackTrace,
146 > Type: nexus.HandlerErrorType(se.Type),
147 > RetryBehavior: se.RetryBehavior(),
148 > OriginalFailure: &f,
149 > }
150 > if f.Cause != nil {
151 he.Cause, err = e.FailureToError(*f.Cause)
152 if err != nil {
154 }
155 }
156 > return he, nil failure_converter.go
157 case "nexus.OperationError":
158 var se serializedOperationError
196 // [Failure] instances are converted to [FailureError] to allow access to the full failure metadata and details if
197 // available.
198 > func DefaultFailureConverter() FailureConverter { failure_converter.go
199 > return defaultFailureConverter
200 > }
201
202 > func retryBehaviorAsOptionalBool(e *nexus.HandlerError) *bool { failure_converter.go
203 > // nolint:exhaustive // this is a simple optional boolean.
204 > switch e.RetryBehavior {
205 case nexus.HandlerErrorRetryBehaviorRetryable:
206 ret := true
go.temporal.io/server/common/nexus/nexusrpc/server.go 29 covered LOC · 10 ranges

Open complete file

93 // status code based on the type of error.
94 // nolint:revive // Keeping all of the logic together for readability, even if it means the function is long.
95 > func (h *BaseHTTPHandler) WriteFailure(writer http.ResponseWriter, r *http.Request, err error) { server.go
96 > var failure nexus.Failure
97 > var failureError *nexus.FailureError
98 > var opError *nexus.OperationError
99 > var handlerError *nexus.HandlerError
100 > var operationState nexus.OperationState
101 > statusCode := http.StatusInternalServerError
102 >
103 > if errors.As(err, &opError) {
104 operationState = opError.State
105 var convErr error
123 }
124 writer.Header().Set(headerOperationState, string(operationState))
125 > } else if errors.As(err, &handlerError) { server.go
126 > var convErr error
127 > failure, convErr = h.FailureConverter.ErrorToFailure(handlerError)
128 > if convErr != nil {
129 h.Logger.Error("failed to convert handler error to failure", "error", convErr)
130 writer.WriteHeader(http.StatusInternalServerError)
132 }
133 // Backward compatibility, unwrap the failure cause.
134 > if r.Header.Get(HeaderTemporalNexusFailureSupport) != "true" && failure.Cause != nil { server.go
135 failure = *failure.Cause
136 }
137 > switch handlerError.Type { server.go
138 > case nexus.HandlerErrorTypeBadRequest: server.go
139 > statusCode = http.StatusBadRequest
140 case nexus.HandlerErrorTypeRequestTimeout:
141 statusCode = http.StatusRequestTimeout
170 }
171
172 > b, err := json.Marshal(failure) server.go
173 > if err != nil {
174 h.Logger.Error("failed to marshal failure", "error", err)
175 writer.WriteHeader(http.StatusInternalServerError)
176 return
177 }
178 > writer.Header().Set("Content-Type", contentTypeJSON) server.go
179 >
180 > // Set the retry header here after ensuring that we don't fail with internal error due to failed marshaling to
181 > // preserve the user's intent.
182 > if handlerError != nil {
183 > switch handlerError.RetryBehavior { server.go
184 case nexus.HandlerErrorRetryBehaviorNonRetryable:
185 writer.Header().Set(headerRetryable, "false")
186 case nexus.HandlerErrorRetryBehaviorRetryable:
187 writer.Header().Set(headerRetryable, "true")
188 > default: server.go
189 // don't set the header
190 }
191 }
192
193 > writer.WriteHeader(statusCode) server.go
194 >
195 > if _, err := writer.Write(b); err != nil {
196 h.Logger.Error("failed to write response body", "error", err)
197 }
go.temporal.io/server/common/nexus/nexusrpc/api.go 27 covered LOC · 12 ranges

Open complete file

43 const statusOperationUnsuccessful = http.StatusFailedDependency
44
45 > func isMediaTypeJSON(contentType string) bool { api.go
46 > if contentType == "" {
47 return false
48 }
49 > mediaType, _, err := mime.ParseMediaType(contentType) api.go
50 > return err == nil && mediaType == "application/json"
51 }
52
53 > func prefixStrippedHTTPHeaderToNexusHeader(httpHeader http.Header, prefix string) nexus.Header { api.go
54 > header := nexus.Header{}
55 > for k, v := range httpHeader {
56 > lowerK := strings.ToLower(k)
57 > if strings.HasPrefix(lowerK, prefix) {
58 > // Nexus headers can only have single values, ignore multiple values. api.go
59 > header[lowerK[len(prefix):]] = v[0]
60 > }
61 }
62 > return header api.go
63 }
64
65 > func addContentHeaderToHTTPHeader(nexusHeader nexus.Header, httpHeader http.Header) http.Header { api.go
66 > for k, v := range nexusHeader {
67 > httpHeader.Set("Content-"+k, v) api.go
68 > }
69 > return httpHeader api.go
70 }
71
77 }
78
79 > func addLinksToHTTPHeader(links []nexus.Link, httpHeader http.Header) error { api.go
80 > for _, link := range links {
81 encodedLink, err := encodeLink(link)
82 if err != nil {
85 httpHeader.Add(headerLink, encodedLink)
86 }
87 > return nil api.go
88 }
89
90 > func getLinksFromHeader(httpHeader http.Header) ([]nexus.Link, error) { api.go
91 > var links []nexus.Link
92 > headerValues := httpHeader.Values(headerLink)
93 > if len(headerValues) == 0 {
94 > return nil, nil api.go
95 > }
96 for encodedLink := range strings.SplitSeq(strings.Join(headerValues, ","), ",") {
97 link, err := decodeLink(encodedLink)
go.temporal.io/server/common/nexus/nexusrpc/client.go 18 covered LOC · 8 ranges

Open complete file

87 }
88
89 > func (c *baseHTTPClient) failureFromResponse(response *http.Response, body []byte) (nexus.Failure, error) { client.go
90 > if !isMediaTypeJSON(response.Header.Get("Content-Type")) {
91 return nexus.Failure{}, newUnexpectedResponseError(fmt.Sprintf("invalid response content type: %q", response.Header.Get("Content-Type")), response, body)
92 }
93 > var failure nexus.Failure client.go
94 > err := json.Unmarshal(body, &failure)
95 > return failure, err
96 }
97
119
120 // bestEffortHandlerErrorFromResponse attempts to read a handler error from the response, but falls back to an unexpected response error.
121 > func (c *baseHTTPClient) bestEffortHandlerErrorFromResponse(response *http.Response, body []byte) error { client.go
122 > failure, err := c.failureFromResponse(response, body)
123 > if err != nil {
124 return c.defaultErrorFromResponse(response, body, nil)
125 }
126 > convErr, err := c.failureConverter.FailureToError(failure) client.go
127 > if err != nil {
128 return newUnexpectedResponseError(fmt.Sprintf("failed to convert Failure to error: %s", err.Error()), response, body)
129 }
130 > if _, ok := convErr.(*nexus.HandlerError); !ok { client.go
131 convErr = c.defaultErrorFromResponse(response, body, convErr)
132 }
133 > return convErr client.go
134 }
135
402 // body with an in-memory buffer.
403 // The body is replaced even when there was an error reading the entire body.
404 > func readAndReplaceBody(response *http.Response) ([]byte, error) { client.go
405 > responseBody := response.Body
406 > body, err := io.ReadAll(responseBody)
407 > if err := responseBody.Close(); err != nil {
408 return nil, err
409 }
410 > response.Body = io.NopCloser(bytes.NewReader(body)) client.go
411 > return body, err
412 }
413