diff --git a/go.mod b/go.mod index 0f20520..09d35d3 100644 --- a/go.mod +++ b/go.mod @@ -28,7 +28,8 @@ require ( go.temporal.io/sdk/contrib/tally v0.2.0 go.uber.org/zap v1.28.0 go.uber.org/zap/exp v0.3.0 - golang.org/x/sync v0.22.0 + golang.org/x/net v0.61.0 + golang.org/x/sync v0.24.0 ) require ( @@ -93,10 +94,9 @@ require ( go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.10.0 // indirect go.yaml.in/yaml/v3 v3.0.5 // indirect - golang.org/x/net v0.58.0 // indirect - golang.org/x/sys v0.47.0 // indirect - golang.org/x/term v0.45.0 // indirect - golang.org/x/text v0.41.0 // indirect + golang.org/x/sys v0.49.0 // indirect + golang.org/x/term v0.47.0 // indirect + golang.org/x/text v0.43.0 // indirect golang.org/x/time v0.15.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260831171406-18b4a7587f8a // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260831171406-18b4a7587f8a // indirect diff --git a/go.sum b/go.sum index 18d3105..f4c4ee1 100644 --- a/go.sum +++ b/go.sum @@ -397,8 +397,8 @@ golang.org/x/net v0.0.0-20200822124328-c89045814202/go.mod h1:/O7V0waA8r7cgGh81R golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/net v0.0.0-20210913180222-943fd674d43e/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= -golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= -golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/net v0.61.0 h1:RR+6j/BTBCrggOLugxa+66R8FvoN6MHlzDlZIbN9YFw= +golang.org/x/net v0.61.0/go.mod h1:WgDCOTH8iwtB66T4/GfI94AHo/hhdb7Hah9NDseQfiI= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20200107190931-bf48bf16ab8d/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -411,8 +411,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= -golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sync v0.24.0 h1:iqiDrGoDgyuvAKrhLQthUEzU/R378N78gXbBjxed7oA= +golang.org/x/sync v0.24.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -431,19 +431,19 @@ golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20210910150752-751e447fb3d0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA= +golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= -golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= +golang.org/x/term v0.47.0 h1:cpyOwQkr2Nzxe2p6KnrTG7Ei2wtzjsuSjhpXXepZ/BM= +golang.org/x/term v0.47.0/go.mod h1:efAfAuG6RKABvg1ArGebRkvX6LXul14SM67o2e4mUCY= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= -golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= -golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/text v0.43.0 h1:1QivrlhwAsnMOcqgOdl9my7s9OyZl8O5/UGUTJVXoWc= +golang.org/x/text v0.43.0/go.mod h1:hbSIYA/amXcRXTFSZqh8tWHEoqRvvbKPJ2uz/KU+a+s= golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= @@ -458,8 +458,8 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= -golang.org/x/tools v0.48.0 h1:3+hClM1aLL5mjMKm5ovokw9epgRXPuu2tILgismM6RE= -golang.org/x/tools v0.48.0/go.mod h1:08xX0orndb/F7jJxGDicx061tyd5pcMto75YMAXr6lk= +golang.org/x/tools v0.51.0 h1:k4Xc/1Om9jwkBJBo4NVLMSARBoWtK10mx+W5BnXCeAI= +golang.org/x/tools v0.51.0/go.mod h1:9eEncMayCV6zRMGhR5eZEC2iBx98qWcF1HZ9Z7wJOoA= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/http.go b/http.go new file mode 100644 index 0000000..8d9073c --- /dev/null +++ b/http.go @@ -0,0 +1,515 @@ +/* + * Copyright 2026 Zigflow authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package temporal + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "math" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + temporalSDK "go.temporal.io/sdk/temporal" + "golang.org/x/net/http/httpguts" +) + +// Application error types returned by HTTPActivity.Call. +const ( + // ErrorTypeHTTPResponseTooLarge is the type of the non-retryable error + // returned when a 2xx response body exceeds the maximum body size. + ErrorTypeHTTPResponseTooLarge = "HTTPResponseTooLarge" + + // ErrorTypeHTTPRequest is the type of errors returned when a request is + // invalid or cannot be executed. Validation errors are non-retryable. + // Errors executing the request are retryable. + ErrorTypeHTTPRequest = "HTTPRequestError" + + // ErrorTypeHTTPStatus is the type of errors returned for non-2xx + // responses. Their details contain an HTTPStatusErrorDetails value. + ErrorTypeHTTPStatus = "HTTPStatusError" +) + +var nonRetryableStatusCodes = map[int]struct{}{ + http.StatusNotImplemented: {}, + http.StatusHTTPVersionNotSupported: {}, + http.StatusVariantAlsoNegotiates: {}, + http.StatusLoopDetected: {}, + http.StatusNetworkAuthenticationRequired: {}, +} + +var retryableStatusCodes = map[int]struct{}{ + http.StatusRequestTimeout: {}, + http.StatusTooManyRequests: {}, +} + +// HTTPActivity makes HTTP requests from Temporal activities. Register its Call +// method with a worker as an activity. +// +// HTTPActivity must be constructed with NewHTTPActivity. The zero value is not +// usable. An HTTPActivity may be used by concurrent activity executions, +// provided any transport or cookie jar supplied through options supports +// concurrent use and is not reconfigured while calls are running. +type HTTPActivity struct { + client *http.Client + maxBodyBytes int64 + maxPreviewLimit int64 +} + +// HTTPRequest describes the HTTP request made by HTTPActivity.Call. It is +// serialised as activity input, so Body and Headers count towards Temporal +// payload size. +type HTTPRequest struct { + // Method is the HTTP method. An empty method means GET. + Method string `json:"method"` + + // URL is the request URL. It must be an absolute HTTP or HTTPS URL with + // a host. + URL string `json:"url"` + + // Headers are the request headers. Header names and values are validated + // and copied before the request is sent. + Headers http.Header `json:"header,omitempty"` + + // Body is the raw request body. The helper does not encode its content + // or choose a Content-Type; callers supply the appropriate header. + Body []byte `json:"body,omitempty"` + + // Query holds values appended to existing URL query parameters. + // When non-empty, the combined query is re-encoded, which can change + // parameter order and escaping. + Query map[string][]string `json:"query,omitempty"` +} + +// HTTPResponse is the result of a 2xx response returned by HTTPActivity.Call. +// It is serialised as the activity result. +type HTTPResponse struct { + // StatusCode is the 2xx HTTP status code. + StatusCode int `json:"statusCode"` + + // Header is a copy of the response headers. + Header http.Header `json:"header"` + + // Body is the complete response body, no larger than the configured + // maximum body size. + Body []byte `json:"body,omitempty"` +} + +// HTTPStatusErrorDetails describes a non-2xx response. It is attached as the +// details of an ErrorTypeHTTPStatus application error. +type HTTPStatusErrorDetails struct { + // StatusCode is the HTTP status code. + StatusCode int `json:"statusCode"` + + // Status is the status code followed by the standard text returned by + // http.StatusText, which is empty for unrecognised codes. For example, + // "404 Not Found". + Status string `json:"status"` + + // Body is a bounded preview of the response body. The preview limit is + // fixed at 4 KiB and can only be reduced indirectly by a smaller maximum + // body size. A read failure may leave a partial preview, with the read + // error attached as the application error's cause. + Body []byte `json:"body,omitempty"` + + // BodyTruncated reports whether Body was truncated because the response + // body exceeded the preview limit. A false value does not guarantee a + // complete preview if reading failed. + BodyTruncated bool `json:"bodyTruncated,omitempty"` +} + +// Call makes the HTTP request described by input and returns the response. +// +// Call runs inside a Temporal activity and makes one HTTP client call per +// activity attempt; redirects and transport behaviour may result in additional +// requests. Temporal controls retries according to the activity's retry policy +// and the retryability of the returned error. Temporal may execute an activity +// more than once, so requests with side effects are only safe to retry when +// the server supports idempotency, for example through an idempotency key +// supplied in Headers that remains stable across activity retries. +// +// Invalid URLs, header names or header values return a non-retryable +// ErrorTypeHTTPRequest error, as do malformed existing query strings when +// input.Query is non-empty. Errors executing the request return a retryable +// ErrorTypeHTTPRequest error. Cancellation observed during request execution +// or response reading returns a Temporal cancelled error. Deadline expiry is +// handled as a failure, not cancellation. +// +// A 2xx response is returned as an HTTPResponse. If its body exceeds the +// maximum body size, Call returns a non-retryable +// ErrorTypeHTTPResponseTooLarge error. Failures reading a 2xx response body +// return a wrapped error without a helper-defined application error type. +// +// Any other status returns an ErrorTypeHTTPStatus error whose details contain +// an HTTPStatusErrorDetails value. The helper's current retry policy is: +// +// - 408 and 429 are retryable +// - 413 is retryable only when the parsed Retry-After delay is positive +// - 501, 505, 506, 508 and 511 are non-retryable +// - other 5xx statuses are retryable +// - all other statuses, including 3xx, are non-retryable +// +// This policy is not a universal guarantee and may not match the semantics of +// every API. For retryable responses, a valid positive +// Retry-After value, in seconds or as an HTTP date, overrides the next retry +// delay on Temporal servers that support it. +func (h *HTTPActivity) Call(ctx context.Context, input HTTPRequest) (*HTTPResponse, error) { + method := input.Method + if method == "" { + method = http.MethodGet + } + + req, err := http.NewRequestWithContext(ctx, method, input.URL, bytes.NewReader(input.Body)) + if err != nil { + return nil, temporalSDK.NewNonRetryableApplicationError( + "error creating http request", + ErrorTypeHTTPRequest, + err, + ) + } + + if err := h.validateURL(req); err != nil { + return nil, err + } + + if input.Headers != nil { + if err := h.validateHeaders(input.Headers); err != nil { + return nil, err + } + + req.Header = input.Headers.Clone() + } + + if len(input.Query) > 0 { + q, err := url.ParseQuery(req.URL.RawQuery) + if err != nil { + return nil, temporalSDK.NewNonRetryableApplicationError( + "error parsing HTTP request query", + ErrorTypeHTTPRequest, + err, + ) + } + + for k, vs := range input.Query { + for _, v := range vs { + q.Add(k, v) + } + } + req.URL.RawQuery = q.Encode() + } + + resp, err := h.client.Do(req) + if err != nil { + if errors.Is(ctx.Err(), context.Canceled) { + return nil, temporalSDK.NewCanceledError() + } + + return nil, temporalSDK.NewApplicationErrorWithCause( + "error executing HTTP request", + ErrorTypeHTTPRequest, + err, + ) + } + defer func() { + _ = resp.Body.Close() + }() + + result, err := h.parseResponse(resp) + + if errors.Is(ctx.Err(), context.Canceled) { + return nil, temporalSDK.NewCanceledError() + } + + return result, err +} + +// isRetryable applies the helper's default HTTP status retry policy. +// Most 5xx responses are retryable. Most 4xx responses are non-retryable. +// API-specific semantics can require a different policy. +func (h *HTTPActivity) isRetryable(statusCode int, retryDelay time.Duration) bool { + // Check it doesn't overlap with the non-retryable status codes + if _, ok := nonRetryableStatusCodes[statusCode]; ok { + return false + } + + // Check for custom status codes + if _, ok := retryableStatusCodes[statusCode]; ok { + return true + } + + if statusCode == http.StatusRequestEntityTooLarge && retryDelay > 0 { + return true + } + + if statusCode >= 500 && statusCode < 600 { + return true + } + + return false +} + +func (h *HTTPActivity) parseResponse(resp *http.Response) (*HTTPResponse, error) { + statusCode := resp.StatusCode + + if statusCode < 200 || statusCode >= 300 { + previewLimit := min(h.maxBodyBytes, h.maxPreviewLimit) + body, readErr := io.ReadAll( + io.LimitReader(resp.Body, previewLimit+1), + ) + + truncated := int64(len(body)) > previewLimit + if truncated { + body = body[:previewLimit] + } + + retryDelay := h.parseRetryAfter(resp) + isRetryable := h.isRetryable(resp.StatusCode, retryDelay) + + status := fmt.Sprintf( + "%d %s", + statusCode, + http.StatusText(statusCode), + ) + + options := temporalSDK.ApplicationErrorOptions{ + NonRetryable: !isRetryable, + Cause: readErr, + Details: []any{ + HTTPStatusErrorDetails{ + StatusCode: statusCode, + Status: status, + Body: body, + BodyTruncated: truncated, + }, + }, + } + + if isRetryable { + options.NextRetryDelay = retryDelay + } + + return nil, temporalSDK.NewApplicationErrorWithOptions( + "HTTP call returned "+status, + ErrorTypeHTTPStatus, + options, + ) + } + + body, err := io.ReadAll(io.LimitReader(resp.Body, h.maxBodyBytes+1)) + if int64(len(body)) > h.maxBodyBytes { + return nil, temporalSDK.NewNonRetryableApplicationError( + fmt.Sprintf("response body exceeds %d bytes", h.maxBodyBytes), + ErrorTypeHTTPResponseTooLarge, + nil, + ) + } + if err != nil { + return nil, fmt.Errorf("error reading http response: %w", err) + } + + return &HTTPResponse{ + StatusCode: statusCode, + Header: resp.Header.Clone(), + Body: body, + }, nil +} + +func (h *HTTPActivity) parseRetryAfter(resp *http.Response) time.Duration { + retryDelay := resp.Header.Get("Retry-After") + + // Check a value set + retryDelay = strings.TrimSpace(retryDelay) + if retryDelay != "" { + const maxRetryAfterSeconds = math.MaxInt64 / int64(time.Second) + + seconds, err := strconv.ParseInt(retryDelay, 10, 64) + if err == nil { + if seconds <= 0 || seconds > maxRetryAfterSeconds { + return 0 + } + + return time.Duration(seconds) * time.Second + } + + if retryAt, err := http.ParseTime(retryDelay); err == nil { + delay := time.Until(retryAt) + + if delay > 0 { + return delay + } + } + } + + return 0 +} + +func (h *HTTPActivity) validateHeaders(headers http.Header) error { + for name, values := range headers { + if !httpguts.ValidHeaderFieldName(name) { + return temporalSDK.NewNonRetryableApplicationError( + "invalid HTTP header name", + ErrorTypeHTTPRequest, + nil, + ) + } + + for _, value := range values { + if !httpguts.ValidHeaderFieldValue(value) { + return temporalSDK.NewNonRetryableApplicationError( + fmt.Sprintf("invalid value for HTTP header %q", name), + ErrorTypeHTTPRequest, + nil, + ) + } + } + } + + return nil +} + +func (h *HTTPActivity) validateURL(req *http.Request) error { + if (req.URL.Scheme != "http" && req.URL.Scheme != "https") || + req.URL.Hostname() == "" || + req.URL.Opaque != "" { + return temporalSDK.NewNonRetryableApplicationError( + "URL must be an absolute HTTP or HTTPS URL with a host", + ErrorTypeHTTPRequest, + nil, + ) + } + + return nil +} + +// NewHTTPActivity returns an HTTPActivity configured with opts. +// +// The defaults are a five-second client timeout, Go's default transport and +// redirect policy, no cookie jar, a 1 MiB response-body limit and a 4 KiB +// error-body preview. +// +// Options apply in order, so a later option overrides an earlier option that +// sets the same value. NewHTTPActivity returns an error if any option is nil +// or returns an error. +func NewHTTPActivity(opts ...HTTPOption) (*HTTPActivity, error) { + // Create the HTTPActivity with sensible defaults + h := &HTTPActivity{ + client: &http.Client{ + Timeout: 5 * time.Second, + Transport: http.DefaultTransport, + }, + maxBodyBytes: 1 << 20, // 1 MiB default; configurable for larger payloads. + maxPreviewLimit: 4 << 10, // 4 KiB + } + + for i, opt := range opts { + if opt == nil { + return nil, fmt.Errorf("http option %d is nil", i) + } + + if err := opt(h); err != nil { + return nil, fmt.Errorf("error setting http option: %w", err) + } + } + + return h, nil +} + +// HTTPOption configures an HTTPActivity during construction. +type HTTPOption func(*HTTPActivity) error + +// WithCookieJar sets the cookie jar used by the HTTP client. A nil jar +// disables the cookie jar, which is the default. +// +// The jar is shared by every call made through the HTTPActivity. It is not +// persisted by Temporal, and activity attempts may run on different workers. +// The jar must support concurrent use and must not be reconfigured while calls +// are running. +func WithCookieJar(jar http.CookieJar) HTTPOption { + return func(h *HTTPActivity) error { + h.client.Jar = jar + return nil + } +} + +// WithMaxBodySize sets the maximum size, in bytes, of a 2xx response body. The +// default is 1 MiB. Larger bodies fail with a non-retryable +// ErrorTypeHTTPResponseTooLarge error. The error-body preview is limited to +// 4 KiB, or to this size if it is smaller. +// +// The limit excludes headers and serialisation overhead and does not guarantee +// that the result fits Temporal's payload limits. Larger limits are supported, +// including when external storage is configured separately. +// +// The option returns an error unless maxBodyBytes is greater than zero and +// less than math.MaxInt64. +func WithMaxBodySize(maxBodyBytes int64) HTTPOption { + return func(h *HTTPActivity) error { + if maxBodyBytes <= 0 || maxBodyBytes == math.MaxInt64 { + return fmt.Errorf( + "maximum body size must be positive and below %d: %d", + math.MaxInt64, + maxBodyBytes, + ) + } + + h.maxBodyBytes = maxBodyBytes + return nil + } +} + +// WithNoRedirect stops the HTTP client following redirects. 3xx responses are +// then returned to Call, which the current status policy treats as +// non-retryable ErrorTypeHTTPStatus errors. +func WithNoRedirect() HTTPOption { + return func(h *HTTPActivity) error { + h.client.CheckRedirect = func(_ *http.Request, _ []*http.Request) error { + return http.ErrUseLastResponse + } + return nil + } +} + +// WithTimeout sets the HTTP client timeout, which covers the whole request +// including reading the response body. The default is five seconds. A zero +// timeout disables the client timeout. The request context, including any +// Temporal activity timeouts, still applies. The option returns an error if +// timeout is negative. +func WithTimeout(timeout time.Duration) HTTPOption { + return func(h *HTTPActivity) error { + if timeout < 0 { + return fmt.Errorf("timeout cannot be below 0: %d", timeout) + } + + h.client.Timeout = timeout + return nil + } +} + +// WithTransport sets the transport used to make requests. A nil transport +// selects Go's default transport. The transport must support concurrent use +// and must not be reconfigured while calls are running. +func WithTransport(transport http.RoundTripper) HTTPOption { + return func(h *HTTPActivity) error { + h.client.Transport = transport + return nil + } +} diff --git a/http_test.go b/http_test.go new file mode 100644 index 0000000..47869e7 --- /dev/null +++ b/http_test.go @@ -0,0 +1,1002 @@ +/* + * Copyright 2026 Zigflow authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package temporal + +import ( + "bytes" + "context" + "errors" + "io" + "math" + "net/http" + "net/http/cookiejar" + "net/url" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + temporalSDK "go.temporal.io/sdk/temporal" +) + +// fakeTransport is an http.RoundTripper backed by a function. It closes the +// request body after the function returns, as the RoundTripper contract +// requires. +type fakeTransport struct { + fn transportFunc +} + +func (f *fakeTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if req.Body != nil { + defer func() { _ = req.Body.Close() }() + } + + return f.fn(req) +} + +// fakeBody is a controllable response body. It returns data, then err (or +// io.EOF when err is nil), and records whether it was closed. +type fakeBody struct { + data []byte + err error + onRead func() + closed atomic.Bool +} + +func (b *fakeBody) Read(p []byte) (int, error) { + if b.onRead != nil { + b.onRead() + } + + if len(b.data) > 0 { + n := copy(p, b.data) + b.data = b.data[n:] + return n, nil + } + + if b.err != nil { + return 0, b.err + } + + return 0, io.EOF +} + +func (b *fakeBody) Close() error { + b.closed.Store(true) + return nil +} + +const testURL = "https://example.com/" + +// transportFunc is the function behind a fakeTransport. +type transportFunc func(*http.Request) (*http.Response, error) + +var ( + errTestRead = errors.New("read failed") + errTestTransport = errors.New("transport failed") +) + +func newResponse(req *http.Request, status int, header http.Header, body io.ReadCloser) *http.Response { + if header == nil { + header = http.Header{} + } + + return &http.Response{ + StatusCode: status, + Status: strconv.Itoa(status) + " " + http.StatusText(status), + Proto: "HTTP/1.1", + ProtoMajor: 1, + ProtoMinor: 1, + Header: header, + Body: body, + ContentLength: -1, + Request: req, + } +} + +// respond returns a transport function that replies with a fresh body on +// every call. +func respond(status int, header http.Header, body []byte) transportFunc { + return func(req *http.Request) (*http.Response, error) { + return newResponse(req, status, header.Clone(), io.NopCloser(bytes.NewReader(body))), nil + } +} + +func respondWithBody(status int, body io.ReadCloser) transportFunc { + return func(req *http.Request) (*http.Response, error) { + return newResponse(req, status, nil, body), nil + } +} + +func unexpectedTransport(t *testing.T) transportFunc { + return func(req *http.Request) (*http.Response, error) { + t.Errorf("unexpected request to %s", req.URL) + return nil, errors.New("unexpected request") + } +} + +// waitForRequestContext blocks until the request context ends, failing the +// test rather than hanging if it never does. +func waitForRequestContext(t *testing.T, req *http.Request) error { + select { + case <-req.Context().Done(): + return req.Context().Err() + case <-time.After(5 * time.Second): + t.Error("request context was not done") + return errors.New("request context was not done") + } +} + +func newTestHTTPActivity( + t *testing.T, + fn transportFunc, + opts ...HTTPOption, +) *HTTPActivity { + t.Helper() + + h, err := NewHTTPActivity(append([]HTTPOption{WithTransport(&fakeTransport{fn: fn})}, opts...)...) + require.NoError(t, err) + + return h +} + +func requireApplicationError( + t *testing.T, + err error, + errType string, + nonRetryable bool, +) *temporalSDK.ApplicationError { + t.Helper() + + var appErr *temporalSDK.ApplicationError + require.ErrorAs(t, err, &appErr) + assert.Equal(t, errType, appErr.Type()) + assert.Equal(t, nonRetryable, appErr.NonRetryable()) + + return appErr +} + +func requireStatusErrorDetails(t *testing.T, appErr *temporalSDK.ApplicationError) HTTPStatusErrorDetails { + t.Helper() + + var details HTTPStatusErrorDetails + require.NoError(t, appErr.Details(&details)) + + return details +} + +func TestHTTPActivityCallRequest(t *testing.T) { + binaryBody := []byte{0x00, 0x01, 0xfe, 0xff, '\r', '\n'} + + tests := []struct { + name string + input HTTPRequest + wantMethod string + wantBody []byte + }{ + { + name: "empty method defaults to GET", + input: HTTPRequest{URL: testURL}, + wantMethod: http.MethodGet, + wantBody: []byte{}, + }, + { + name: "explicit method and binary body are preserved", + input: HTTPRequest{Method: http.MethodPut, URL: testURL, Body: binaryBody}, + wantMethod: http.MethodPut, + wantBody: binaryBody, + }, + { + name: "custom method is preserved", + input: HTTPRequest{Method: "PROPFIND", URL: "http://example.com/"}, + wantMethod: "PROPFIND", + wantBody: []byte{}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + assert.Equal(t, test.wantMethod, req.Method) + + body, err := io.ReadAll(req.Body) + assert.NoError(t, err) + assert.Equal(t, test.wantBody, body) + + return respond(http.StatusOK, nil, nil)(req) + }) + + _, err := h.Call(context.Background(), test.input) + require.NoError(t, err) + }) + } +} + +func TestHTTPActivityCallCopiesHeaders(t *testing.T) { + input := http.Header{ + "Content-Type": {"application/octet-stream"}, + "X-Multi": {"a", "b"}, + } + want := input.Clone() + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + assert.Equal(t, "application/octet-stream", req.Header.Get("Content-Type")) + assert.Equal(t, []string{"a", "b"}, req.Header.Values("X-Multi")) + + req.Header.Set("X-Multi", "mutated") + req.Header.Add("X-Added", "1") + + return respond(http.StatusOK, nil, nil)(req) + }) + + _, err := h.Call(context.Background(), HTTPRequest{URL: testURL, Headers: input}) + require.NoError(t, err) + assert.Equal(t, want, input) +} + +func TestHTTPActivityCallRejectsInvalidRequests(t *testing.T) { + tests := []struct { + name string + input HTTPRequest + }{ + {name: "invalid method", input: HTTPRequest{Method: "BAD METHOD", URL: testURL}}, + {name: "unparseable URL", input: HTTPRequest{URL: "http://[::1"}}, + {name: "empty URL", input: HTTPRequest{URL: ""}}, + {name: "relative URL", input: HTTPRequest{URL: "/relative"}}, + {name: "unsupported scheme", input: HTTPRequest{URL: "ftp://example.com/"}}, + {name: "missing host", input: HTTPRequest{URL: "http:///path"}}, + {name: "opaque URL", input: HTTPRequest{URL: "http:opaque"}}, + { + name: "invalid header name", + input: HTTPRequest{URL: testURL, Headers: http.Header{"Bad Name": {"value"}}}, + }, + { + name: "empty header name", + input: HTTPRequest{URL: testURL, Headers: http.Header{"": {"value"}}}, + }, + { + name: "header value with line break", + input: HTTPRequest{URL: testURL, Headers: http.Header{"X-Test": {"a\r\nInjected: b"}}}, + }, + { + name: "header value with control character", + input: HTTPRequest{URL: testURL, Headers: http.Header{"X-Test": {"a\x00b"}}}, + }, + { + name: "malformed existing query with additional values", + input: HTTPRequest{ + URL: "https://example.com/?a=1;b=2", + Query: map[string][]string{"c": {"3"}}, + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + h := newTestHTTPActivity(t, unexpectedTransport(t)) + + res, err := h.Call(context.Background(), test.input) + assert.Nil(t, res) + _ = requireApplicationError(t, err, ErrorTypeHTTPRequest, true) + }) + } +} + +func TestHTTPActivityCallQuery(t *testing.T) { + t.Run("values are appended and escaped", func(t *testing.T) { + const special = "hello world&=?/" + + query := map[string][]string{ + "a": {"2", "3"}, + "c": {special}, + } + want := map[string][]string{ + "a": {"2", "3"}, + "c": {special}, + } + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + got, err := url.ParseQuery(req.URL.RawQuery) + assert.NoError(t, err) + assert.Equal(t, url.Values{ + "a": {"1", "2", "3"}, + "b": {"x/y"}, + "c": {special}, + }, got) + + return respond(http.StatusOK, nil, nil)(req) + }) + + _, err := h.Call(context.Background(), HTTPRequest{ + URL: "https://example.com/search?a=1&b=x%2Fy", + Query: query, + }) + require.NoError(t, err) + assert.Equal(t, want, query) + }) + + rawQueryTests := []struct { + name string + query map[string][]string + }{ + {name: "nil query", query: nil}, + {name: "empty query", query: map[string][]string{}}, + } + + for _, test := range rawQueryTests { + t.Run(test.name+" preserves the raw query", func(t *testing.T) { + const rawQuery = "z=1&a=%7e&b=a+b&c=1;d=2" + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + assert.Equal(t, rawQuery, req.URL.RawQuery) + return respond(http.StatusOK, nil, nil)(req) + }) + + _, err := h.Call(context.Background(), HTTPRequest{ + URL: "https://example.com/p?" + rawQuery, + Query: test.query, + }) + require.NoError(t, err) + }) + } +} + +func TestHTTPActivityCallSuccess(t *testing.T) { + body := []byte{0x00, 0xff, 'o', 'k', '\n'} + + for _, status := range []int{http.StatusOK, http.StatusCreated, http.StatusAccepted, http.StatusPartialContent} { + t.Run(strconv.Itoa(status), func(t *testing.T) { + header := http.Header{ + "Content-Type": {"application/octet-stream"}, + "X-Multi": {"a", "b"}, + } + + h := newTestHTTPActivity(t, respond(status, header, body)) + + res, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + require.NoError(t, err) + assert.Equal(t, status, res.StatusCode) + assert.Equal(t, header, res.Header) + assert.Equal(t, body, res.Body) + }) + } + + t.Run("empty body", func(t *testing.T) { + h := newTestHTTPActivity(t, respond(http.StatusNoContent, nil, nil)) + + res, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + require.NoError(t, err) + assert.Equal(t, http.StatusNoContent, res.StatusCode) + assert.Empty(t, res.Body) + }) +} + +func TestHTTPActivityCallBodyLimit(t *testing.T) { + tests := []struct { + name string + opts []HTTPOption + size int + wantErr bool + }{ + {name: "configured limit exactly", opts: []HTTPOption{WithMaxBodySize(16)}, size: 16}, + {name: "configured limit plus one", opts: []HTTPOption{WithMaxBodySize(16)}, size: 17, wantErr: true}, + {name: "default limit exactly", size: 1 << 20}, + {name: "default limit plus one", size: 1<<20 + 1, wantErr: true}, + {name: "limit above 2 MiB", opts: []HTTPOption{WithMaxBodySize(3 << 20)}, size: 5 << 19}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + body := bytes.Repeat([]byte{'x'}, test.size) + h := newTestHTTPActivity(t, respond(http.StatusOK, nil, body), test.opts...) + + res, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + if test.wantErr { + assert.Nil(t, res) + _ = requireApplicationError(t, err, ErrorTypeHTTPResponseTooLarge, true) + return + } + + require.NoError(t, err) + assert.Equal(t, body, res.Body) + }) + } +} + +func TestHTTPActivityCallStatusPolicy(t *testing.T) { + tests := []struct { + status int + wantRetryable bool + }{ + {status: http.StatusRequestTimeout, wantRetryable: true}, + {status: http.StatusTooManyRequests, wantRetryable: true}, + {status: http.StatusNotImplemented}, + {status: http.StatusHTTPVersionNotSupported}, + {status: http.StatusVariantAlsoNegotiates}, + {status: http.StatusLoopDetected}, + {status: http.StatusNetworkAuthenticationRequired}, + {status: http.StatusInternalServerError, wantRetryable: true}, + {status: http.StatusBadGateway, wantRetryable: true}, + {status: http.StatusServiceUnavailable, wantRetryable: true}, + {status: http.StatusGatewayTimeout, wantRetryable: true}, + {status: http.StatusInsufficientStorage, wantRetryable: true}, + {status: 599, wantRetryable: true}, + {status: http.StatusMultipleChoices}, + {status: http.StatusNotModified}, + {status: http.StatusBadRequest}, + {status: http.StatusUnauthorized}, + {status: http.StatusForbidden}, + {status: http.StatusNotFound}, + {status: http.StatusConflict}, + {status: http.StatusMisdirectedRequest}, + {status: http.StatusUnprocessableEntity}, + {status: http.StatusTooEarly}, + } + + for _, test := range tests { + t.Run(strconv.Itoa(test.status), func(t *testing.T) { + h := newTestHTTPActivity(t, respond(test.status, nil, nil)) + + res, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + assert.Nil(t, res) + + appErr := requireApplicationError(t, err, ErrorTypeHTTPStatus, !test.wantRetryable) + assert.Equal(t, test.status, requireStatusErrorDetails(t, appErr).StatusCode) + }) + } +} + +func TestHTTPActivityCallRetryAfter(t *testing.T) { + future := time.Now().Add(time.Hour).UTC().Format(http.TimeFormat) + past := time.Now().Add(-time.Hour).UTC().Format(http.TimeFormat) + maxSeconds := math.MaxInt64 / int64(time.Second) + + tests := []struct { + name string + status int + retryAfter *string + wantRetryable bool + wantMin time.Duration + wantMax time.Duration + }{ + {name: "413 absent", status: http.StatusRequestEntityTooLarge}, + {name: "413 invalid", status: http.StatusRequestEntityTooLarge, retryAfter: new("soon")}, + {name: "413 zero", status: http.StatusRequestEntityTooLarge, retryAfter: new("0")}, + {name: "413 negative", status: http.StatusRequestEntityTooLarge, retryAfter: new("-10")}, + {name: "413 past date", status: http.StatusRequestEntityTooLarge, retryAfter: &past}, + { + name: "413 positive seconds", + status: http.StatusRequestEntityTooLarge, + retryAfter: new("120"), + wantRetryable: true, + wantMin: 120 * time.Second, + wantMax: 120 * time.Second, + }, + { + name: "413 future date", + status: http.StatusRequestEntityTooLarge, + retryAfter: &future, + wantRetryable: true, + wantMin: 58 * time.Minute, + wantMax: 61 * time.Minute, + }, + { + name: "503 positive seconds", + status: http.StatusServiceUnavailable, + retryAfter: new("120"), + wantRetryable: true, + wantMin: 120 * time.Second, + wantMax: 120 * time.Second, + }, + { + name: "503 surrounding whitespace", + status: http.StatusServiceUnavailable, + retryAfter: new(" 30\t"), + wantRetryable: true, + wantMin: 30 * time.Second, + wantMax: 30 * time.Second, + }, + { + name: "503 future date", + status: http.StatusServiceUnavailable, + retryAfter: &future, + wantRetryable: true, + wantMin: 58 * time.Minute, + wantMax: 61 * time.Minute, + }, + { + name: "429 positive seconds", + status: http.StatusTooManyRequests, + retryAfter: new("5"), + wantRetryable: true, + wantMin: 5 * time.Second, + wantMax: 5 * time.Second, + }, + {name: "503 absent", status: http.StatusServiceUnavailable, wantRetryable: true}, + {name: "503 invalid", status: http.StatusServiceUnavailable, retryAfter: new("soon"), wantRetryable: true}, + {name: "503 past date", status: http.StatusServiceUnavailable, retryAfter: &past, wantRetryable: true}, + { + name: "503 overflows int64", + status: http.StatusServiceUnavailable, + retryAfter: new("99999999999999999999"), + wantRetryable: true, + }, + { + name: "503 overflows duration", + status: http.StatusServiceUnavailable, + retryAfter: new(strconv.FormatInt(maxSeconds+1, 10)), + wantRetryable: true, + }, + {name: "404 positive seconds", status: http.StatusNotFound, retryAfter: new("120")}, + {name: "501 positive seconds", status: http.StatusNotImplemented, retryAfter: new("120")}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + header := http.Header{} + if test.retryAfter != nil { + header["Retry-After"] = []string{*test.retryAfter} + } + + h := newTestHTTPActivity(t, respond(test.status, header, nil)) + + _, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + appErr := requireApplicationError(t, err, ErrorTypeHTTPStatus, !test.wantRetryable) + + delay := appErr.NextRetryDelay() + assert.GreaterOrEqual(t, delay, test.wantMin) + assert.LessOrEqual(t, delay, test.wantMax) + }) + } +} + +func TestHTTPActivityCallStatusErrorDetails(t *testing.T) { + tests := []struct { + name string + opts []HTTPOption + status int + body []byte + wantRetryable bool + wantStatus string + wantBody []byte + wantTruncated bool + }{ + { + name: "retryable", + status: http.StatusServiceUnavailable, + body: []byte("unavailable"), + wantRetryable: true, + wantStatus: "503 Service Unavailable", + wantBody: []byte("unavailable"), + }, + { + name: "non-retryable", + status: http.StatusNotFound, + body: []byte{0x00, 0xff}, + wantStatus: "404 Not Found", + wantBody: []byte{0x00, 0xff}, + }, + { + name: "preview at 4 KiB limit", + status: http.StatusNotFound, + body: bytes.Repeat([]byte{'a'}, 4<<10), + wantBody: bytes.Repeat([]byte{'a'}, 4<<10), + }, + { + name: "preview over 4 KiB limit", + status: http.StatusNotFound, + body: append(bytes.Repeat([]byte{'a'}, 4<<10), 'b'), + wantBody: bytes.Repeat([]byte{'a'}, 4<<10), + wantTruncated: true, + }, + { + name: "preview at smaller body limit", + opts: []HTTPOption{WithMaxBodySize(100)}, + status: http.StatusNotFound, + body: bytes.Repeat([]byte{'a'}, 100), + wantBody: bytes.Repeat([]byte{'a'}, 100), + }, + { + name: "preview over smaller body limit", + opts: []HTTPOption{WithMaxBodySize(100)}, + status: http.StatusNotFound, + body: append(bytes.Repeat([]byte{'a'}, 100), 'b'), + wantBody: bytes.Repeat([]byte{'a'}, 100), + wantTruncated: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + h := newTestHTTPActivity(t, respond(test.status, nil, test.body), test.opts...) + + _, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + appErr := requireApplicationError(t, err, ErrorTypeHTTPStatus, !test.wantRetryable) + + details := requireStatusErrorDetails(t, appErr) + assert.Equal(t, test.status, details.StatusCode) + if test.wantStatus != "" { + assert.Equal(t, test.wantStatus, details.Status) + } + assert.Equal(t, test.wantBody, details.Body) + assert.Equal(t, test.wantTruncated, details.BodyTruncated) + }) + } +} + +func TestHTTPActivityCallReadFailures(t *testing.T) { + t.Run("2xx body", func(t *testing.T) { + h := newTestHTTPActivity(t, respondWithBody(http.StatusOK, &fakeBody{data: []byte("partial"), err: errTestRead})) + + res, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + assert.Nil(t, res) + require.ErrorIs(t, err, errTestRead) + }) + + previewTests := []struct { + status int + wantRetryable bool + }{ + {status: http.StatusServiceUnavailable, wantRetryable: true}, + {status: http.StatusNotFound}, + } + + for _, test := range previewTests { + t.Run("error preview "+strconv.Itoa(test.status), func(t *testing.T) { + h := newTestHTTPActivity(t, respondWithBody(test.status, &fakeBody{data: []byte("partial"), err: errTestRead})) + + _, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + appErr := requireApplicationError(t, err, ErrorTypeHTTPStatus, !test.wantRetryable) + require.ErrorIs(t, err, errTestRead) + + details := requireStatusErrorDetails(t, appErr) + assert.Equal(t, test.status, details.StatusCode) + assert.Equal(t, []byte("partial"), details.Body) + assert.False(t, details.BodyTruncated) + }) + } +} + +func TestHTTPActivityCallTransportFailure(t *testing.T) { + h := newTestHTTPActivity(t, func(*http.Request) (*http.Response, error) { + return nil, errTestTransport + }) + + res, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + assert.Nil(t, res) + _ = requireApplicationError(t, err, ErrorTypeHTTPRequest, false) + require.ErrorIs(t, err, errTestTransport) + assert.False(t, temporalSDK.IsCanceledError(err)) +} + +func TestHTTPActivityCallCancellation(t *testing.T) { + t.Run("during request execution", func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + cancel() + return nil, waitForRequestContext(t, req) + }) + + res, err := h.Call(ctx, HTTPRequest{URL: testURL}) + assert.Nil(t, res) + assert.True(t, temporalSDK.IsCanceledError(err), "got %v", err) + }) + + for _, status := range []int{http.StatusOK, http.StatusServiceUnavailable} { + t.Run("during response reading "+strconv.Itoa(status), func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + + body := &fakeBody{onRead: cancel, err: context.Canceled} + h := newTestHTTPActivity(t, respondWithBody(status, body)) + + res, err := h.Call(ctx, HTTPRequest{URL: testURL}) + assert.Nil(t, res) + assert.True(t, temporalSDK.IsCanceledError(err), "got %v", err) + }) + } +} + +func TestHTTPActivityCallDeadlines(t *testing.T) { + t.Run("client timeout", func(t *testing.T) { + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + return nil, waitForRequestContext(t, req) + }, WithTimeout(20*time.Millisecond)) + + res, err := h.Call(context.Background(), HTTPRequest{URL: testURL}) + assert.Nil(t, res) + _ = requireApplicationError(t, err, ErrorTypeHTTPRequest, false) + assert.False(t, temporalSDK.IsCanceledError(err)) + }) + + t.Run("context deadline", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + t.Cleanup(cancel) + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + return nil, waitForRequestContext(t, req) + }, WithTimeout(0)) + + res, err := h.Call(ctx, HTTPRequest{URL: testURL}) + assert.Nil(t, res) + _ = requireApplicationError(t, err, ErrorTypeHTTPRequest, false) + require.ErrorIs(t, err, context.DeadlineExceeded) + assert.False(t, temporalSDK.IsCanceledError(err)) + }) +} + +func TestHTTPActivityCallClosesResponseBody(t *testing.T) { + tests := []struct { + name string + opts []HTTPOption + status int + body *fakeBody + }{ + {name: "success", status: http.StatusOK, body: &fakeBody{data: []byte("ok")}}, + {name: "status failure", status: http.StatusNotFound, body: &fakeBody{data: []byte("missing")}}, + { + name: "oversized", + opts: []HTTPOption{WithMaxBodySize(4)}, + status: http.StatusOK, + body: &fakeBody{data: []byte("12345")}, + }, + {name: "read failure", status: http.StatusOK, body: &fakeBody{err: errTestRead}}, + {name: "preview read failure", status: http.StatusServiceUnavailable, body: &fakeBody{err: errTestRead}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + h := newTestHTTPActivity(t, respondWithBody(test.status, test.body), test.opts...) + + _, _ = h.Call(context.Background(), HTTPRequest{URL: testURL}) + assert.True(t, test.body.closed.Load()) + }) + } +} + +func TestNewHTTPActivityDefaults(t *testing.T) { + h, err := NewHTTPActivity() + require.NoError(t, err) + + assert.Equal(t, 5*time.Second, h.client.Timeout) + assert.Equal(t, int64(1<<20), h.maxBodyBytes) + assert.Equal(t, int64(4<<10), h.maxPreviewLimit) + assert.Equal(t, http.DefaultTransport, h.client.Transport) + assert.Nil(t, h.client.Jar) + assert.Nil(t, h.client.CheckRedirect) +} + +func TestNewHTTPActivityOptions(t *testing.T) { + transportA := &fakeTransport{} + transportB := &fakeTransport{} + jar, err := cookiejar.New(nil) + require.NoError(t, err) + + tests := []struct { + name string + opts []HTTPOption + check func(t *testing.T, h *HTTPActivity) + }{ + { + name: "timeout override", + opts: []HTTPOption{WithTimeout(2 * time.Second)}, + check: func(t *testing.T, h *HTTPActivity) { assert.Equal(t, 2*time.Second, h.client.Timeout) }, + }, + { + name: "zero timeout", + opts: []HTTPOption{WithTimeout(0)}, + check: func(t *testing.T, h *HTTPActivity) { assert.Zero(t, h.client.Timeout) }, + }, + { + name: "body size override", + opts: []HTTPOption{WithMaxBodySize(10)}, + check: func(t *testing.T, h *HTTPActivity) { assert.Equal(t, int64(10), h.maxBodyBytes) }, + }, + { + name: "transport override", + opts: []HTTPOption{WithTransport(transportA)}, + check: func(t *testing.T, h *HTTPActivity) { assert.Same(t, transportA, h.client.Transport) }, + }, + { + name: "nil transport", + opts: []HTTPOption{WithTransport(nil)}, + check: func(t *testing.T, h *HTTPActivity) { assert.Nil(t, h.client.Transport) }, + }, + { + name: "cookie jar", + opts: []HTTPOption{WithCookieJar(jar)}, + check: func(t *testing.T, h *HTTPActivity) { assert.Same(t, jar, h.client.Jar) }, + }, + { + name: "nil cookie jar", + opts: []HTTPOption{WithCookieJar(nil)}, + check: func(t *testing.T, h *HTTPActivity) { assert.Nil(t, h.client.Jar) }, + }, + { + name: "later options take precedence", + opts: []HTTPOption{ + WithTimeout(time.Second), + WithMaxBodySize(1), + WithTransport(transportA), + WithCookieJar(jar), + WithTimeout(2 * time.Second), + WithMaxBodySize(2), + WithTransport(transportB), + WithCookieJar(nil), + }, + check: func(t *testing.T, h *HTTPActivity) { + assert.Equal(t, 2*time.Second, h.client.Timeout) + assert.Equal(t, int64(2), h.maxBodyBytes) + assert.Same(t, transportB, h.client.Transport) + assert.Nil(t, h.client.Jar) + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + h, err := NewHTTPActivity(test.opts...) + require.NoError(t, err) + test.check(t, h) + }) + } +} + +func TestNewHTTPActivityOptionErrors(t *testing.T) { + tests := []struct { + name string + opts []HTTPOption + }{ + {name: "nil option", opts: []HTTPOption{WithTimeout(time.Second), nil}}, + {name: "negative timeout", opts: []HTTPOption{WithTimeout(-time.Nanosecond)}}, + {name: "zero body size", opts: []HTTPOption{WithMaxBodySize(0)}}, + {name: "negative body size", opts: []HTTPOption{WithMaxBodySize(-1)}}, + {name: "maximum int64 body size", opts: []HTTPOption{WithMaxBodySize(math.MaxInt64)}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + h, err := NewHTTPActivity(test.opts...) + require.Error(t, err) + assert.Nil(t, h) + }) + } +} + +func TestHTTPActivityCallRedirects(t *testing.T) { + const ( + startPath = "/start" + finalPath = "/final" + ) + + tests := []struct { + name string + opts []HTTPOption + wantPaths []string + }{ + {name: "followed by default", wantPaths: []string{startPath, finalPath}}, + {name: "disabled", opts: []HTTPOption{WithNoRedirect()}, wantPaths: []string{startPath}}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var paths []string + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + paths = append(paths, req.URL.Path) + + switch req.URL.Path { + case startPath: + return respond(http.StatusFound, http.Header{"Location": {finalPath}}, nil)(req) + case finalPath: + return respond(http.StatusOK, nil, []byte("done"))(req) + default: + return unexpectedTransport(t)(req) + } + }, test.opts...) + + res, err := h.Call(context.Background(), HTTPRequest{URL: "https://example.com" + startPath}) + assert.Equal(t, test.wantPaths, paths) + + if len(test.wantPaths) == 2 { + require.NoError(t, err) + assert.Equal(t, []byte("done"), res.Body) + return + } + + assert.Nil(t, res) + appErr := requireApplicationError(t, err, ErrorTypeHTTPStatus, true) + assert.Equal(t, http.StatusFound, requireStatusErrorDetails(t, appErr).StatusCode) + }) + } +} + +func TestHTTPActivityCallCookies(t *testing.T) { + tests := []struct { + name string + opts func(jar http.CookieJar) []HTTPOption + wantCookies []string + }{ + { + name: "reused within the same activity", + opts: func(jar http.CookieJar) []HTTPOption { return []HTTPOption{WithCookieJar(jar)} }, + wantCookies: []string{"", "session=abc"}, + }, + { + name: "disabled by a later nil jar", + opts: func(jar http.CookieJar) []HTTPOption { + return []HTTPOption{WithCookieJar(jar), WithCookieJar(nil)} + }, + wantCookies: []string{"", ""}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + jar, err := cookiejar.New(nil) + require.NoError(t, err) + + var cookies []string + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + cookies = append(cookies, req.Header.Get("Cookie")) + return respond(http.StatusOK, http.Header{"Set-Cookie": {"session=abc; Path=/"}}, nil)(req) + }, test.opts(jar)...) + + for _, path := range []string{"/login", "/data"} { + _, err := h.Call(context.Background(), HTTPRequest{URL: "https://example.com" + path}) + require.NoError(t, err) + } + + assert.Equal(t, test.wantCookies, cookies) + }) + } +} + +func TestHTTPActivityCallConcurrent(t *testing.T) { + const calls = 32 + + var count atomic.Int64 + + h := newTestHTTPActivity(t, func(req *http.Request) (*http.Response, error) { + count.Add(1) + return respond(http.StatusOK, nil, []byte(req.URL.Query().Get("id")))(req) + }) + + var wg sync.WaitGroup + for i := range calls { + wg.Go(func() { + id := strconv.Itoa(i) + + res, err := h.Call(context.Background(), HTTPRequest{ + URL: testURL, + Query: map[string][]string{"id": {id}}, + }) + if assert.NoError(t, err) { + assert.Equal(t, []byte(id), res.Body) + } + }) + } + wg.Wait() + + assert.Equal(t, int64(calls), count.Load()) +}