Atlas › Test
TestCancel_ContextDeadlinePropagated
Exact test identity: go.temporal.io/server/common/nexus/nexusrpc/TestCancel_ContextDeadlinePropagated
- Package
go.temporal.io/server/common/nexus/nexusrpc- Suite / test hierarchy
TestCancel_ContextDeadlinePropagated- Test
TestCancel_ContextDeadlinePropagated- Introduced at
- TestCancel_HandleFromClient, TestCancel_ContextDeadlinePropagated Frontier kind: Test frontier
- Covered ranges
- 59
- Covered lines
- 152
- Covered files
- 5
Co-introduced tests
1 other test enter at the same concept.
Covered source
Expand a file to inspect source; the > gutter marks covered lines.
go.temporal.io/server/common/nexus/nexusrpc/server.go 66 covered LOC · 27 ranges
Open complete file
245
}
246
247
>
func (h *httpHandler) cancelOperation(service, operation, token string, writer http.ResponseWriter, request *http.Request) {
server.go
248
>
options := nexus.CancelOperationOptions{Header: httpHeaderToNexusHeader(request.Header)}
249
>
250
>
ctx, cancel, ok := h.contextWithTimeoutFromHTTPRequest(writer, request)
251
>
if !ok {
252
return
253
}
255
>
256
>
ctx = nexus.WithHandlerContext(ctx, nexus.HandlerInfo{
257
>
Service: service,
258
>
Operation: operation,
259
>
Header: options.Header,
260
>
})
261
>
if err := h.options.Handler.CancelOperation(ctx, service, operation, token, options); err != nil {
262
h.WriteFailure(writer, request, err)
263
return
264
}
265
266
>
writer.WriteHeader(http.StatusAccepted)
server.go
267
}
268
270
// Returns (0, true) if unset. Returns ({parsedDuration}, true) if set. If set and there is an error parsing the
271
// duration, it writes a failure response and returns (0, false).
272
>
func (h *httpHandler) parseRequestTimeoutHeader(writer http.ResponseWriter, request *http.Request) (time.Duration, bool) {
server.go
273
>
timeoutStr := request.Header.Get(nexus.HeaderRequestTimeout)
274
>
if timeoutStr != "" {
275
>
timeoutDuration, err := ParseDuration(timeoutStr)
server.go
276
>
if err != nil {
277
h.Logger.Warn("invalid request timeout header", "timeout", timeoutStr)
278
h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid request timeout header"))
279
return 0, false
280
}
282
}
283
return 0, true
286
// contextWithTimeoutFromHTTPRequest extracts the context from the HTTP request and applies the timeout indicated by
287
// the Request-Timeout header, if set.
288
>
func (h *httpHandler) contextWithTimeoutFromHTTPRequest(writer http.ResponseWriter, request *http.Request) (context.Context, context.CancelFunc, bool) {
server.go
289
>
requestTimeout, ok := h.parseRequestTimeoutHeader(writer, request)
290
>
if !ok {
291
return nil, nil, false
292
}
294
>
ctx, cancel := context.WithTimeout(request.Context(), requestTimeout)
server.go
295
>
return ctx, cancel, true
296
>
}
297
return request.Context(), func() {}, true
298
}
318
}
319
320
>
func (h *httpHandler) handleRequest(writer http.ResponseWriter, request *http.Request) {
server.go
321
>
if request.Method != "POST" {
322
h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "invalid request method: expected POST, got %q", request.Method))
323
return
324
}
325
>
parts := strings.Split(request.URL.EscapedPath(), "/")
server.go
326
>
// First part is empty (due to leading /)
327
>
if len(parts) < 3 {
328
h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeNotFound, "not found"))
329
return
330
}
331
>
service, err := url.PathUnescape(parts[1])
server.go
332
>
if err != nil {
333
h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to parse URL path"))
334
return
335
}
336
>
operation, err := url.PathUnescape(parts[2])
server.go
337
>
if err != nil {
338
h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeBadRequest, "failed to parse URL path"))
339
return
341
342
// First handle StartOperation at /{service}/{operation}
344
h.startOperation(service, operation, writer, request)
345
return
348
// Handle deprecated /{service}/{operation}/{operation_token}/cancel
349
// TODO(bergundy): remove in server release v1.31.0
350
>
if len(parts) == 5 && parts[4] == "cancel" {
server.go
351
token, err := url.PathUnescape(parts[3])
352
if err != nil {
359
}
360
361
>
if len(parts) != 4 || parts[3] != "cancel" {
server.go
362
h.WriteFailure(writer, request, nexus.NewHandlerErrorf(nexus.HandlerErrorTypeNotFound, "not found"))
363
return
364
}
365
366
>
token := request.Header.Get(nexus.HeaderOperationToken)
server.go
367
>
if token == "" {
368
token = request.URL.Query().Get("token")
369
if token == "" {
371
return
372
}
374
>
// Sanitize this header as it is explicitly passed in as an argument.
375
>
request.Header.Del(nexus.HeaderOperationToken)
376
>
}
377
378
>
h.cancelOperation(service, operation, token, writer, request)
server.go
379
}
380
381
// NewHTTPHandler constructs an [http.Handler] from given options for handling Nexus service requests.
382
>
func NewHTTPHandler(options HandlerOptions) http.Handler {
server.go
383
>
if options.Logger == nil {
384
>
options.Logger = slog.Default()
server.go
385
>
}
386
>
if options.GetResultTimeout == 0 {
server.go
387
options.GetResultTimeout = time.Minute
388
}
389
>
if options.Serializer == nil {
server.go
390
>
options.Serializer = nexus.DefaultSerializer()
server.go
391
>
}
392
>
if options.FailureConverter == nil {
server.go
393
>
options.FailureConverter = DefaultFailureConverter()
server.go
394
>
}
396
>
BaseHTTPHandler: BaseHTTPHandler{
397
>
Logger: options.Logger,
398
>
FailureConverter: options.FailureConverter,
399
>
},
400
>
options: options,
401
>
}
402
>
403
>
return http.HandlerFunc(handler.handleRequest)
404
}
go.temporal.io/server/common/nexus/nexusrpc/client.go 42 covered LOC · 15 ranges
Open complete file
152
// NewHTTPClient creates a new [HTTPClient] from provided [HTTPClientOptions].
153
// BaseURL and Service are required.
154
>
func NewHTTPClient(options HTTPClientOptions) (*HTTPClient, error) {
client.go
155
>
if options.HTTPCaller == nil {
156
>
options.HTTPCaller = http.DefaultClient.Do
157
>
}
158
>
if options.BaseURL == "" {
159
return nil, errors.New("empty BaseURL")
160
}
162
return nil, errors.New("empty Service")
163
}
165
>
var err error
166
>
baseURL, err = url.Parse(options.BaseURL)
167
>
if err != nil {
168
return nil, err
169
}
170
>
if baseURL.Scheme != "http" && baseURL.Scheme != "https" {
client.go
171
return nil, fmt.Errorf("invalid URL scheme: %s", baseURL.Scheme)
172
}
173
>
if options.Serializer == nil {
client.go
174
>
options.Serializer = nexus.DefaultSerializer()
client.go
175
>
}
176
>
if options.FailureConverter == nil {
client.go
177
>
options.FailureConverter = DefaultFailureConverter()
client.go
178
>
}
180
>
baseHTTPClient: baseHTTPClient{
181
>
serializer: options.Serializer,
182
>
failureConverter: options.FailureConverter,
183
>
httpCaller: options.HTTPCaller,
184
>
},
185
>
serviceBaseURL: baseURL,
186
>
service: options.Service,
187
>
}, nil
188
}
189
381
// Does not incur a trip to the server.
382
// Fails if provided an empty operation or token.
383
>
func (c *HTTPClient) NewOperationHandle(operation string, token string) (*OperationHandle[*nexus.LazyValue], error) {
client.go
384
>
var es []error
385
>
if operation == "" {
386
es = append(es, errEmptyOperationName)
387
}
389
es = append(es, errEmptyOperationToken)
390
}
392
return nil, errors.Join(es...)
393
}
394
>
return &OperationHandle[*nexus.LazyValue]{
client.go
395
>
client: c,
396
>
Operation: operation,
397
>
Token: token,
398
>
}, nil
399
}
400
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
go.temporal.io/server/common/nexus/nexusrpc/api.go 27 covered LOC · 11 ranges
Open complete file
104
}
105
106
>
func httpHeaderToNexusHeader(httpHeader http.Header, excludePrefixes ...string) nexus.Header {
api.go
107
>
header := nexus.Header{}
108
>
headerLoop:
109
>
for k, v := range httpHeader {
110
>
lowerK := strings.ToLower(k)
111
>
for _, prefix := range excludePrefixes {
112
if strings.HasPrefix(lowerK, prefix) {
113
continue headerLoop
115
}
116
// Nexus headers can only have single values, ignore multiple values.
117
>
header[lowerK] = v[0]
api.go
118
}
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)
125
}
126
>
return httpHeader
api.go
127
}
128
129
>
func addContextTimeoutToHTTPHeader(ctx context.Context, httpHeader http.Header) http.Header {
api.go
130
>
deadline, ok := ctx.Deadline()
131
>
if !ok {
132
return httpHeader
133
}
134
>
httpHeader.Set(nexus.HeaderRequestTimeout, FormatDuration(time.Until(deadline)))
api.go
135
>
return httpHeader
136
}
137
254
var durationRegexp = regexp.MustCompile(`^(\d+(?:\.\d+)?)(ms|s|m)$`)
255
256
>
func ParseDuration(value string) (time.Duration, error) {
api.go
257
>
m := durationRegexp.FindStringSubmatch(value)
258
>
if len(m) == 0 {
259
return 0, fmt.Errorf("invalid duration: %q", value)
260
}
261
>
v, err := strconv.ParseFloat(m[1], 64)
api.go
262
>
if err != nil {
263
return 0, err
264
}
265
267
>
case "ms":
268
>
return time.Millisecond * time.Duration(v), nil
269
case "s":
270
return time.Millisecond * time.Duration(v*1e3), nil
277
278
// FormatDuration converts a duration into a string representation in millisecond resolution.
279
>
func FormatDuration(d time.Duration) string {
api.go
280
>
return strconv.FormatInt(d.Milliseconds(), 10) + "ms"
281
>
}
282
283
// MarkAsWrapperError adds the "unwrap-error" metadata to the original failure of the given OperationError, which
go.temporal.io/server/common/nexus/nexusrpc/handle.go 14 covered LOC · 5 ranges
Open complete file
22
//
23
// Cancelation is asynchronous and may be not be respected by the operation's implementation.
24
>
func (h *OperationHandle[T]) Cancel(ctx context.Context, options nexus.CancelOperationOptions) error {
handle.go
25
>
u := h.client.serviceBaseURL.JoinPath(url.PathEscape(h.client.service), url.PathEscape(h.Operation), "cancel")
26
>
request, err := http.NewRequestWithContext(ctx, "POST", u.String(), nil)
27
>
if err != nil {
28
return err
29
}
30
>
request.Header.Set(nexus.HeaderOperationToken, h.Token)
handle.go
31
>
addContextTimeoutToHTTPHeader(ctx, request.Header)
32
>
request.Header.Set(headerUserAgent, userAgent)
33
>
addNexusHeaderToHTTPHeader(options.Header, request.Header)
34
>
response, err := h.client.httpCaller(request)
35
>
if err != nil {
36
return err
37
}
38
39
// Do this once here and make sure it doesn't leak.
40
>
body, err := readAndReplaceBody(response)
handle.go
41
>
if err != nil {
42
return err
43
}
44
45
>
if response.StatusCode != http.StatusAccepted {
handle.go
46
return h.client.bestEffortHandlerErrorFromResponse(response, body)
47
}
49
}
go.temporal.io/server/common/nexus/nexusrpc/failure_converter.go 3 covered LOC · 1 range
Open complete file
196
// [Failure] instances are converted to [FailureError] to allow access to the full failure metadata and details if
197
// available.
199
>
return defaultFailureConverter
200
>
}
201
202
func retryBehaviorAsOptionalBool(e *nexus.HandlerError) *bool {