Vihren Atlas

Atlas › Test

TestFailureCompletion

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

Package
go.temporal.io/server/common/nexus/nexusrpc
Suite / test hierarchy
TestFailureCompletion
Test
TestFailureCompletion
Introduced at
completion.go ×1 Joint frontier
Covered ranges
105
Covered lines
237
Covered files
4

Covered source

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

go.temporal.io/server/common/nexus/nexusrpc/completion.go 107 covered LOC · 49 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. completion.go
71 > // Just in case something unexpected happens while discarding or closing the body,
72 > // propagate errors to the machine.
73 > if _, err = io.Copy(io.Discard, response.Body); err == nil {
74 > if err = response.Body.Close(); err != nil {
75 return err
76 }
77 }
78 > return nil completion.go
79 }
80
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) completion.go
121 > if err != nil {
122 return err
123 }
124 // Backwards compatibility: if the failure has a cause, unwrap it to ensure behavior remains compatible with older servers.
125 > if failure.Cause != nil { completion.go
126 failure = *failure.Cause
127 }
128 > b, err := json.Marshal(failure) completion.go
129 > if err != nil {
130 return err
131 }
132 > request.Body = io.NopCloser(bytes.NewReader(b)) completion.go
133 > // Set the operation state header for backwards compatibility.
134 > request.Header.Set(headerOperationState, string(c.Error.State))
135 > request.Header.Set("Content-Type", contentTypeJSON)
136 } else {
137 reader, ok := c.Result.(*nexus.Reader)
158 }
159
160 > if c.Header != nil { completion.go
161 > addNexusHeaderToHTTPHeader(c.Header, request.Header) completion.go
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) completion.go
168 > }
169 > if c.Header.Get(headerOperationStartTime) == "" && !c.StartTime.IsZero() { completion.go
170 > request.Header.Set(headerOperationStartTime, c.StartTime.Format(http.TimeFormat)) completion.go
171 > }
172 > if c.Header.Get(headerOperationCloseTime) == "" && !c.CloseTime.IsZero() { completion.go
173 > request.Header.Set(headerOperationCloseTime, marshalTimestamp(c.CloseTime)) completion.go
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 completion.go
239 > if completion.StartTime, parseTimeErr = http.ParseTime(startTimeHeader); parseTimeErr != nil {
240 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to parse operation start time header"))
241 return
242 }
243 }
244 > if closeTimeHeader := request.Header.Get(headerOperationCloseTime); closeTimeHeader != "" { completion.go
245 > var parseTimeErr error completion.go
246 > if completion.CloseTime, parseTimeErr = unmarshalTimestamp(closeTimeHeader); parseTimeErr != nil {
247 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to parse operation close time header"))
248 return
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: completion.go
258 > if !isMediaTypeJSON(request.Header.Get("Content-Type")) {
259 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid request content type: %q", request.Header.Get("Content-Type")))
260 return
261 }
262 > var failure nexus.Failure completion.go
263 > b, err := io.ReadAll(request.Body)
264 > if err != nil {
265 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to read Failure from request body"))
266 return
267 }
268 > if err := json.Unmarshal(b, &failure); err != nil { completion.go
269 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to decode Failure from request body"))
270 return
271 }
272 > completionErr, err := h.FailureConverter.FailureToError(failure) completion.go
273 > if err != nil {
274 h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to decode Failure from request body"))
275 return
276 }
277 > opErr, ok := completionErr.(*nexus.OperationError) completion.go
278 > if !ok {
279 // Backwards compatibility: wrap non-OperationError errors in an OperationError with the appropriate state.
280 completion.Error = &nexus.OperationError{
287 return
288 }
289 > } else { completion.go
290 > completion.Error = opErr
291 > }
292 case nexus.OperationStateSucceeded:
293 completion.Result = nexus.NewLazyValue(
302 return
303 }
304 > if err := h.options.Handler.CompleteOperation(ctx, &completion); err != nil { completion.go
305 h.WriteFailure(writer, request, err)
306 }
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/api.go 75 covered LOC · 39 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
77 }
78
79 > func addLinksToHTTPHeader(links []nexus.Link, httpHeader http.Header) error { api.go
80 > for _, link := range links {
81 > encodedLink, err := encodeLink(link) api.go
82 > if err != nil {
83 return err
84 }
85 > httpHeader.Add(headerLink, encodedLink) api.go
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
95 }
96 > for encodedLink := range strings.SplitSeq(strings.Join(headerValues, ","), ",") { api.go
97 > link, err := decodeLink(encodedLink)
98 > if err != nil {
99 return nil, err
100 }
101 > links = append(links, link) api.go
102 }
103 > return links, nil api.go
104 }
105
120 }
121
122 > func addNexusHeaderToHTTPHeader(nexusHeader nexus.Header, httpHeader http.Header) http.Header { api.go
123 > for k, v := range nexusHeader {
124 > httpHeader.Set(k, v) api.go
125 > }
126 > return httpHeader api.go
127 }
128
140 // decodeLink encodes the link to Nexus-Link header value.
141 // It follows the same format of HTTP Link header: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link
142 > func encodeLink(link nexus.Link) (string, error) { api.go
143 > if err := validateLinkURL(link.URL); err != nil {
144 return "", fmt.Errorf("failed to encode link: %w", err)
145 }
146 > if err := validateLinkType(link.Type); err != nil { api.go
147 return "", fmt.Errorf("failed to encode link: %w", err)
148 }
149 > return fmt.Sprintf(`<%s>; %s="%s"`, link.URL.String(), linkTypeKey, link.Type), nil api.go
150 }
151
152 // decodeLink decodes the Nexus-Link header values.
153 // It must have the same format of HTTP Link header: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Link
154 > func decodeLink(encodedLink string) (nexus.Link, error) { api.go
155 > var link nexus.Link
156 > encodedLink = strings.TrimSpace(encodedLink)
157 > if len(encodedLink) == 0 {
158 return link, errors.New("failed to parse link header: value is empty")
159 }
160
161 > if encodedLink[0] != '<' { api.go
162 return link, fmt.Errorf("failed to parse link header: invalid format: %s", encodedLink)
163 }
164 > urlEnd := strings.Index(encodedLink, ">") api.go
165 > if urlEnd == -1 {
166 return link, fmt.Errorf("failed to parse link header: invalid format: %s", encodedLink)
167 }
168 > urlStr := strings.TrimSpace(encodedLink[1:urlEnd]) api.go
169 > if len(urlStr) == 0 {
170 return link, errors.New("failed to parse link header: url is empty")
171 }
172 > u, err := url.Parse(urlStr) api.go
173 > if err != nil {
174 return link, fmt.Errorf("failed to parse link header: invalid url: %s", urlStr)
175 }
176 > if err := validateLinkURL(u); err != nil { api.go
177 return link, fmt.Errorf("failed to parse link header: %w", err)
178 }
179 > link.URL = u api.go
180 >
181 > params := strings.Split(encodedLink[urlEnd+1:], ";")
182 > // must contain at least one semi-colon, and first param must be empty since
183 > // it corresponds to the url part parsed above.
184 > if len(params) < 2 {
185 return link, fmt.Errorf("failed to parse link header: invalid format: %s", encodedLink)
186 }
187 > if strings.TrimSpace(params[0]) != "" { api.go
188 return link, fmt.Errorf("failed to parse link header: invalid format: %s", encodedLink)
189 }
190
191 > typeKeyFound := false api.go
192 > for _, param := range params[1:] {
193 > param = strings.TrimSpace(param)
194 > if len(param) == 0 {
195 return link, fmt.Errorf("failed to parse link header: parameter is empty: %s", encodedLink)
196 }
197 > kv := strings.SplitN(param, "=", 2) api.go
198 > if len(kv) != 2 {
199 return link, fmt.Errorf("failed to parse link header: invalid parameter format: %s", param)
200 }
201 > key := strings.TrimSpace(kv[0]) api.go
202 > val := strings.TrimSpace(kv[1])
203 > if strings.HasPrefix(val, `"`) != strings.HasSuffix(val, `"`) {
204 return link, fmt.Errorf(
205 "failed to parse link header: parameter value missing double-quote: %s",
207 )
208 }
209 > if strings.HasPrefix(val, `"`) { api.go
210 > val = val[1 : len(val)-1] api.go
211 > }
212 > if key == linkTypeKey { api.go
213 > if err := validateLinkType(val); err != nil {
214 return link, fmt.Errorf("failed to parse link header: %w", err)
215 }
216 > link.Type = val api.go
217 > typeKeyFound = true
218 }
219 }
220 > if !typeKeyFound { api.go
221 return link, fmt.Errorf(
222 "failed to parse link header: %q key not found: %s",
226 }
227
228 > return link, nil api.go
229 }
230
231 > func validateLinkURL(value *url.URL) error { api.go
232 > if value == nil || value.String() == "" {
233 return errors.New("url is empty")
234 }
235 > _, err := url.ParseQuery(value.RawQuery) api.go
236 > if err != nil {
237 return fmt.Errorf("url query not percent-encoded: %s", value)
238 }
239 > return nil api.go
240 }
241
242 > func validateLinkType(value string) error { api.go
243 > if len(value) == 0 {
244 return errors.New("link type is empty")
245 }
246 > for _, c := range value { api.go
247 > if (c < 'a' || c > 'z') && (c < 'A' || c > 'Z') && (c < '0' || c > '9') && c != '_' && c != '.' && c != '/' {
248 return errors.New("link type contains invalid char (valid chars: alphanumeric, '_', '.', '/')")
249 }
250 }
251 > return nil api.go
252 }
253
go.temporal.io/server/common/nexus/nexusrpc/failure_converter.go 39 covered LOC · 12 ranges

Open complete file

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
95 }
96 return f, nil
97 > case *nexus.OperationError: failure_converter.go
98 > if typedErr.OriginalFailure != nil {
99 return *typedErr.OriginalFailure, nil
100 }
101 > data := serializedOperationError{ failure_converter.go
102 > State: string(typedErr.State),
103 > }
104 > details, err := json.Marshal(data)
105 > if err != nil {
106 return nexus.Failure{}, err
107 }
108 > f := nexus.Failure{ failure_converter.go
109 > Message: typedErr.Message,
110 > StackTrace: typedErr.StackTrace,
111 > Metadata: map[string]string{
112 > "type": "nexus.OperationError",
113 > },
114 > Details: details,
115 > }
116 >
117 > if typedErr.Cause != nil {
118 c, err := e.ErrorToFailure(typedErr.Cause)
119 if err != nil {
122 f.Cause = &c
123 }
124 > return f, nil failure_converter.go
125 default:
126 return nexus.Failure{
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":
138 var se serializedHandlerError
155 }
156 return he, nil
157 > case "nexus.OperationError": failure_converter.go
158 > var se serializedOperationError
159 > err := json.Unmarshal(f.Details, &se)
160 > if err != nil {
161 return nil, fmt.Errorf("failed to deserialize OperationError: %w", err)
162 }
163 > oe := &nexus.OperationError{ failure_converter.go
164 > Message: f.Message,
165 > StackTrace: f.StackTrace,
166 > State: nexus.OperationState(se.State),
167 > OriginalFailure: &f,
168 > }
169 > if f.Cause != nil {
170 oe.Cause, err = e.FailureToError(*f.Cause)
171 if err != nil {
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 {
go.temporal.io/server/common/nexus/nexusrpc/timestamp.go 16 covered LOC · 5 ranges

Open complete file

15 // uses 0, 3, 6 or 9 fractional digits.
16 // Copied from https://github.com/protocolbuffers/protobuf-go/blob/0b2c87d84c27802dae7248480444e22421ba577d/encoding/protojson/well_known_types.go#L749C1-L826C2
17 > func marshalTimestamp(t time.Time) string { timestamp.go
18 > x := t.UTC().Format("2006-01-02T15:04:05.000000000")
19 > x = strings.TrimSuffix(x, "000")
20 > x = strings.TrimSuffix(x, "000")
21 > x = strings.TrimSuffix(x, ".000")
22 > return x + "Z"
23 > }
24
25 // unmarshalTimestamp unmarshals a string into a Time instance. Uses RFC 3339, with some extra validation to ensure that
26 // seconds and subseconds are with an expected range.
27 // Copied from https://github.com/protocolbuffers/protobuf-go/blob/0b2c87d84c27802dae7248480444e22421ba577d/encoding/protojson/well_known_types.go#L749C1-L826C2
28 > func unmarshalTimestamp(s string) (time.Time, error) { timestamp.go
29 > t, err := time.Parse(time.RFC3339Nano, s)
30 > if err != nil {
31 return t, err
32 }
33 // Validate seconds.
34 > secs := t.Unix() timestamp.go
35 > if secs < minTimestampSeconds || secs > maxTimestampSeconds {
36 return t, fmt.Errorf("second value out of range: %v", secs)
37 }
38 // Validate subseconds.
39 > i := strings.LastIndexByte(s, '.') // start of subsecond field timestamp.go
40 > j := strings.LastIndexAny(s, "Z-+") // start of timezone field
41 > if i >= 0 && j >= i && j-i > len(".999999999") {
42 return t, fmt.Errorf("invalid subsecond value %v", s)
43 }
44 > return t, nil timestamp.go
45 }