diff --git a/http_action/README.md b/http_action/README.md index c314de545..0cd47bb97 100644 --- a/http_action/README.md +++ b/http_action/README.md @@ -40,6 +40,9 @@ The HTTP Action Capability enables Chainlink Runtime Environment (CRE) workflows #### 2.1.4 Gateway Mode Client - **Purpose**: Gateway-proxied HTTP request execution - **Features**: Rate limiting, request deduplication via consistent hashing, exponential backoff retry +- **Deadlines**: `timeoutMs` bounds delivery to a gateway. The response wait is `timeoutMs + + responseGraceMs`, started after the send, so the gateway (which applies `timeoutMs` to the endpoint + call itself) always reports back before we give up. Exhausting the grace means the gateway went silent. --- @@ -96,6 +99,30 @@ type Response struct { - `maxAgeMs`: Must be non-negative and not exceed configured `maxCacheAgeMs`. - `store`: Can be true or false; +### 3.4 Error Classification + +Failures are attributed to the party responsible for them. Only platform faults reach +`http_action_execution_error_count`, which is what the "Execution Errors in More than F Nodes" alert +reads. A slow or failing customer endpoint must never page us. + +| Condition | Error type | Capability error | Counters (`http_action_` prefix) | +|---|---|---|---| +| Input validation failed | `UserError` | user, `InvalidArgument` / `LimitExceeded` | `validation_failure_count` | +| Gateway reports endpoint send/read failure | `UserError` | user, `InvalidArgument` | `external_endpoint_error_count` | +| Gateway reports a blocked request | `UserError` | user, `InvalidArgument` | `validation_failure_count` | +| Response exceeds the size limit | `UserError` | user, `LimitExceeded` | `external_endpoint_error_count` | +| No gateway reachable | plain error | system, `Internal` | `capability_gateway_send_error_count`, `execution_error_count` | +| Gateway returned an unclassified error | plain error | system, `Internal` | `execution_error_count` | +| Gateway silent past the response deadline | `TimeoutError` | system, `DeadlineExceeded` | `execution_timeout_count`, `execution_error_count` | +| Caller canceled before a response arrived | `CanceledError` | system, `Canceled` | `request_canceled_count` | + +Direct mode returns `InputValidationError` rather than `UserError` for the first row; both map to the +same capability error. + +Cancellation is not the user's fault, but `caperrors.Origin` has only `System` and `User`, so it is +reported as a system error carrying `Canceled`. Alerts over capability failures should exclude that +code rather than treat it as a platform fault. + --- ## 4. Configuration Specification @@ -144,9 +171,13 @@ type GatewayConnectionConfig struct { InitialIntervalMs uint32 `json:"initialIntervalMs"` // Initial retry interval MaxElapsedTimeMs uint32 `json:"maxElapsedTimeMs"` // Maximum retry duration Multiplier float64 `json:"multiplier"` // Backoff multiplier + ResponseGraceMs uint32 `json:"responseGraceMs"` // Extra wait for a gateway response beyond timeoutMs } ``` +`responseGraceMs` defaults to 5000, mirroring the gateway's own budget for delivering a response back +to the node. It is optional: omitting it keeps the default, so existing deployments need no change. + ### 4.6 Configuration Examples #### 4.6.1 Gateway Mode Configuration @@ -157,7 +188,8 @@ type GatewayConnectionConfig struct { "gatewayConnection": { "initialIntervalMs": 100, "maxElapsedTimeMs": 30000, - "multiplier": 2.0 + "multiplier": 2.0, + "responseGraceMs": 5000 } } ``` diff --git a/http_action/action/action.go b/http_action/action/action.go index bb95faaa0..56010efae 100644 --- a/http_action/action/action.go +++ b/http_action/action/action.go @@ -138,6 +138,20 @@ func (s *service) SendRequest(ctx context.Context, metadata capabilities.Request metadata.WorkflowID, metadata.WorkflowOwner, metadata.WorkflowName, metadata.WorkflowExecutionID, err), common.UserErrorCode(err)) } + var canceledErr gateway.CanceledError + if errors.As(err, &canceledErr) { + return nil, caperrors.NewPublicSystemError( + fmt.Errorf("request canceled for workflowID %s (Owner: %s, Name: %s, ExecutionID: %s): %w", + metadata.WorkflowID, metadata.WorkflowOwner, metadata.WorkflowName, metadata.WorkflowExecutionID, err), + caperrors.Canceled) + } + var gatewayTimeoutErr gateway.TimeoutError + if errors.As(err, &gatewayTimeoutErr) { + return nil, caperrors.NewPublicSystemError( + fmt.Errorf("request failed for workflowID %s (Owner: %s, Name: %s, ExecutionID: %s): %w", + metadata.WorkflowID, metadata.WorkflowOwner, metadata.WorkflowName, metadata.WorkflowExecutionID, err), + caperrors.DeadlineExceeded) + } return nil, caperrors.NewPublicSystemError( fmt.Errorf("request failed for workflowID %s (Owner: %s, Name: %s, ExecutionID: %s): %w", metadata.WorkflowID, metadata.WorkflowOwner, metadata.WorkflowName, metadata.WorkflowExecutionID, err), caperrors.Internal) diff --git a/http_action/action/action_test.go b/http_action/action/action_test.go index 8985a6bc4..711f869a7 100644 --- a/http_action/action/action_test.go +++ b/http_action/action/action_test.go @@ -389,4 +389,46 @@ func TestSendRequest_ErrorHandling(t *testing.T) { assert.Equal(t, caperrors.Internal, capErr.Code()) assert.Equal(t, caperrors.VisibilityPublic, capErr.Visibility()) }) + + t.Run("client returns TimeoutError and service returns DeadlineExceeded system error", func(t *testing.T) { + setup := setupServiceTest(t) + + input := &http.Request{ + Url: "https://example.com", + Method: "GET", + Timeout: durationpb.New(1000 * time.Millisecond), + CacheSettings: &http.CacheSettings{}, + } + + setup.mockClient.Err = gateway.NewTimeoutError(errors.New("gateway did not respond before deadline")) + + _, err := setup.service.SendRequest(t.Context(), setup.metadata, input) + require.Error(t, err) + + var capErr caperrors.Error + assert.True(t, errors.As(err, &capErr)) + assert.Equal(t, caperrors.DeadlineExceeded, capErr.Code()) + assert.Equal(t, caperrors.OriginSystem, capErr.Origin()) + }) + + t.Run("client returns CanceledError and service returns Canceled system error", func(t *testing.T) { + setup := setupServiceTest(t) + + input := &http.Request{ + Url: "https://example.com", + Method: "GET", + Timeout: durationpb.New(1000 * time.Millisecond), + CacheSettings: &http.CacheSettings{}, + } + + setup.mockClient.Err = gateway.NewCanceledError(context.Canceled) + + _, err := setup.service.SendRequest(t.Context(), setup.metadata, input) + require.Error(t, err) + + var capErr caperrors.Error + assert.True(t, errors.As(err, &capErr)) + assert.Equal(t, caperrors.Canceled, capErr.Code()) + assert.Equal(t, caperrors.OriginSystem, capErr.Origin()) + }) } diff --git a/http_action/common/metrics.go b/http_action/common/metrics.go index a5f0b0b88..bb3864c79 100644 --- a/http_action/common/metrics.go +++ b/http_action/common/metrics.go @@ -29,6 +29,7 @@ type Metrics struct { successfulResponse metric.Int64Counter executionError metric.Int64Counter executionTimeout metric.Int64Counter + requestCanceled metric.Int64Counter externalEndpointError metric.Int64Counter requestLatency metric.Int64Histogram requestLatencyExcludingExternal metric.Int64Histogram @@ -99,12 +100,20 @@ func (m *Metrics) init() error { m.executionTimeout, err = meter.Int64Counter( "http_action_execution_timeout_count", - metric.WithDescription("Number of HTTP action execution timeouts"), + metric.WithDescription("Number of HTTP action requests where the gateway did not respond before the response deadline"), ) if err != nil { return fmt.Errorf("failed to create execution timeout metric: %w", err) } + m.requestCanceled, err = meter.Int64Counter( + "http_action_request_canceled_count", + metric.WithDescription("Number of HTTP action requests canceled by the caller before a gateway response arrived"), + ) + if err != nil { + return fmt.Errorf("failed to create request canceled metric: %w", err) + } + m.externalEndpointError, err = meter.Int64Counter( "http_action_external_endpoint_error_count", metric.WithDescription("Number of HTTP action external endpoint errors"), @@ -171,11 +180,17 @@ func (m *Metrics) IncrementExecutionError(ctx context.Context, proxyMode ProxyMo m.executionError.Add(ctx, 1, metric.WithAttributes(attribute.String(AttrProxyMode, proxyMode.String()))) } +// Gateway silence is a platform fault, so it also counts as an execution error. func (m *Metrics) IncrementExecutionTimeout(ctx context.Context, proxyMode ProxyMode, lggr logger.Logger) { m.executionTimeout.Add(ctx, 1, metric.WithAttributes(attribute.String(AttrProxyMode, proxyMode.String()))) m.executionError.Add(ctx, 1, metric.WithAttributes(attribute.String(AttrProxyMode, proxyMode.String()))) } +// Caller cancellation is nobody's fault, so it is deliberately not an execution error. +func (m *Metrics) IncrementRequestCanceled(ctx context.Context, proxyMode ProxyMode, lggr logger.Logger) { + m.requestCanceled.Add(ctx, 1, metric.WithAttributes(attribute.String(AttrProxyMode, proxyMode.String()))) +} + func (m *Metrics) IncrementExternalEndpointError(ctx context.Context, proxyMode ProxyMode, lggr logger.Logger) { m.externalEndpointError.Add(ctx, 1, metric.WithAttributes(attribute.String(AttrProxyMode, proxyMode.String()))) } diff --git a/http_action/common/service_config.go b/http_action/common/service_config.go index 0d356e64e..2c2a51f75 100644 --- a/http_action/common/service_config.go +++ b/http_action/common/service_config.go @@ -54,6 +54,8 @@ type GatewayConnectionConfig struct { MaxElapsedTimeMs uint32 `json:"maxElapsedTimeMs"` // Multiplier is the multiplier for the exponential backoff retry strategy. Multiplier float64 `json:"multiplier"` + // ResponseGraceMs is the extra time the node waits for a gateway response beyond the request's own timeout. + ResponseGraceMs uint32 `json:"responseGraceMs"` } // HTTPClientConfig defines configuration options for the HTTP client used in "direct" mode. diff --git a/http_action/gateway/gateway_outbound_proxy.go b/http_action/gateway/gateway_outbound_proxy.go index dcb5723df..c32f2998f 100644 --- a/http_action/gateway/gateway_outbound_proxy.go +++ b/http_action/gateway/gateway_outbound_proxy.go @@ -28,6 +28,8 @@ const ( defaultGatewayConnectionInitialIntervalMs = 100 // 100 milliseconds defaultGatewayConnectionMaxElapsedTimeMs = 5_000 defaultGatewayConnectionMultiplier = 2.0 + // Mirrors the gateway's own budget for delivering a response back to us. + defaultGatewayResponseGraceMs = 5_000 ) var ( @@ -110,6 +112,9 @@ func applyDefaults(cfg common.GatewayConnectionConfig) common.GatewayConnectionC if cfg.Multiplier == 0 { cfg.Multiplier = defaultGatewayConnectionMultiplier } + if cfg.ResponseGraceMs == 0 { + cfg.ResponseGraceMs = defaultGatewayResponseGraceMs + } return cfg } @@ -136,8 +141,9 @@ func (p *gatewayOutboundProxy) SendRequest(ctx context.Context, metadata capabil } input = validatedInput - ctx, cancel := context.WithTimeout(ctx, input.Timeout.AsDuration()) - defer cancel() + // Bounds delivery to a gateway only; the response wait gets its own deadline after the send. + sendCtx, cancelSend := context.WithTimeout(ctx, input.Timeout.AsDuration()) + defer cancelSend() // Set only one of Headers or MultiHeaders on the outgoing request (MultiHeaders if input has it, else Headers). gatewayHeaders, gatewayMultiHeaders := gatewayHeadersFromInput(input) @@ -190,13 +196,13 @@ func (p *gatewayOutboundProxy) SendRequest(ctx context.Context, metadata capabil p.metrics.IncrementRequestCount(ctx, lggr) - donID, err := p.validator.ResolveGatewayProxyDonID(ctx) + donID, err := p.validator.ResolveGatewayProxyDonID(sendCtx) if err != nil { p.metrics.IncrementExecutionError(ctx, common.ProxyModeGateway, lggr) return nil, 0, fmt.Errorf("failed to resolve gateway proxy DON: %w", err) } - selectedGateway, err := p.awaitConnection(ctx, lggr, donID, gatewayReq.Hash()) + selectedGateway, err := p.awaitConnection(sendCtx, lggr, donID, gatewayReq.Hash()) if err != nil { p.metrics.IncrementGatewaySendError(ctx, selectedGateway, donID, lggr) return nil, 0, fmt.Errorf("failed to establish connection to gateway: %w", err) @@ -205,11 +211,15 @@ func (p *gatewayOutboundProxy) SendRequest(ctx context.Context, metadata capabil lggr.Debugw("sending request to gateway", "donID", donID, "selectedGateway", selectedGateway) p.metrics.IncrementGatewaySendCount(ctx, selectedGateway, donID, lggr) - if err := p.gatewayConnector.SendToGateway(ctx, selectedGateway, &gatewayResp); err != nil { + if err := p.gatewayConnector.SendToGateway(sendCtx, selectedGateway, &gatewayResp); err != nil { p.metrics.IncrementGatewaySendError(ctx, selectedGateway, donID, lggr) return nil, 0, fmt.Errorf("failed to send request to gateway: %w", err) } + responseTimeout := input.Timeout.AsDuration() + time.Duration(p.gatewayConnectionConfig.ResponseGraceMs)*time.Millisecond + waitCtx, cancelWait := context.WithTimeout(ctx, responseTimeout) + defer cancelWait() + select { case resp := <-responseCh: lggr.Debugw("received response from gateway") @@ -241,18 +251,33 @@ func (p *gatewayOutboundProxy) SendRequest(ctx context.Context, metadata capabil } return response, resp.ExternalEndpointLatency, nil - case <-ctx.Done(): - p.metrics.IncrementExecutionTimeout(ctx, common.ProxyModeGateway, lggr) + case <-waitCtx.Done(): elapsedMs := time.Since(startTime).Milliseconds() - timeoutMs := input.Timeout.AsDuration().Milliseconds() - cause := context.Cause(ctx) - lggr.Debugw(ErrMsgGatewayResponseWait, + timeoutMs := responseTimeout.Milliseconds() + + // A live parent means we hit our own deadline; otherwise the caller cancelled us. + if parentErr := ctx.Err(); parentErr != nil { + p.metrics.IncrementRequestCanceled(ctx, common.ProxyModeGateway, lggr) + cause := context.Cause(ctx) + lggr.Debugw(ErrMsgGatewayResponseWait, + "elapsedMs", elapsedMs, + "timeoutMs", timeoutMs, + "cause", cause, + ) + return nil, 0, NewCanceledError( + fmt.Errorf("%s (elapsedMs: %d, timeoutMs: %d): %w", ErrMsgGatewayResponseWait, elapsedMs, timeoutMs, cause), + ) + } + + p.metrics.IncrementExecutionTimeout(ctx, common.ProxyModeGateway, lggr) + lggr.Errorw(ErrMsgGatewayResponseTimeout, "elapsedMs", elapsedMs, "timeoutMs", timeoutMs, - "cause", cause, + "selectedGateway", selectedGateway, + "donID", donID, ) - return nil, 0, NewUserError( - fmt.Errorf("%s (elapsedMs: %d, timeoutMs: %d): %w", ErrMsgGatewayResponseWait, elapsedMs, timeoutMs, cause), + return nil, 0, NewTimeoutError( + fmt.Errorf("%s (elapsedMs: %d, timeoutMs: %d): %w", ErrMsgGatewayResponseTimeout, elapsedMs, timeoutMs, context.Cause(waitCtx)), ) } } diff --git a/http_action/gateway/gateway_outbound_proxy_test.go b/http_action/gateway/gateway_outbound_proxy_test.go index 4e07a611a..90d790678 100644 --- a/http_action/gateway/gateway_outbound_proxy_test.go +++ b/http_action/gateway/gateway_outbound_proxy_test.go @@ -131,6 +131,10 @@ func TestOutgoingConnectorHandler_AwaitConnection(t *testing.T) { // Helper for setting up proxy and mockConnector for SendRequest tests func setupSendRequestTest(t *testing.T) (*gatewayOutboundProxy, *mockGatewayConnector, chan string) { + return setupSendRequestTestWithConfig(t, common.ServiceConfig{}) +} + +func setupSendRequestTestWithConfig(t *testing.T, cfg common.ServiceConfig) (*gatewayOutboundProxy, *mockGatewayConnector, chan string) { readyCh := make(chan string, 1) mockConnector := &mockGatewayConnector{ SourceDonID: "don1", @@ -144,7 +148,7 @@ func setupSendRequestTest(t *testing.T) (*gatewayOutboundProxy, *mockGatewayConn lggr := logger.Test(t) proxy, err := NewGatewayOutboundProxy( mockConnector, - common.ServiceConfig{}, + cfg, lggr, newMetrics(t), newTestValidator(t), @@ -342,9 +346,7 @@ func TestGatewayOutboundProxy_SendRequest_UserErrors(t *testing.T) { assert.True(t, errors.As(err, &userErr)) }) - // Ensure that canceling the SendRequest context before a gateway response - // is a UserError. - t.Run("gateway response timeout returns UserError", func(t *testing.T) { + t.Run("caller cancellation returns CanceledError", func(t *testing.T) { proxy, _, readyCh := setupSendRequestTest(t) metadata := capabilities.RequestMetadata{ @@ -386,8 +388,85 @@ func TestGatewayOutboundProxy_SendRequest_UserErrors(t *testing.T) { assert.Contains(t, err.Error(), ErrMsgGatewayResponseWait) assert.Contains(t, err.Error(), "context canceled") + var canceledErr CanceledError + assert.True(t, errors.As(err, &canceledErr)) var userErr UserError - assert.True(t, errors.As(err, &userErr)) + assert.False(t, errors.As(err, &userErr)) + var timeoutErr TimeoutError + assert.False(t, errors.As(err, &timeoutErr)) + }) + + t.Run("no gateway response returns TimeoutError", func(t *testing.T) { + proxy, _, readyCh := setupSendRequestTestWithConfig(t, common.ServiceConfig{ + GatewayConnectionConfig: common.GatewayConnectionConfig{ResponseGraceMs: 100}, + }) + + metadata := capabilities.RequestMetadata{ + WorkflowID: "wf1", + WorkflowExecutionID: "exec1", + WorkflowOwner: "owner1", + } + input := &http.Request{ + Url: "http://example.com", + Method: "GET", + Body: []byte("test"), + Timeout: durationpb.New(200 * time.Millisecond), + CacheSettings: &http.CacheSettings{}, + } + + // Never respond on behalf of the gateway; readyCh is buffered so the send does not block. + _ = readyCh + + output, _, err := proxy.SendRequest(t.Context(), metadata, input, time.Now()) + require.Error(t, err) + require.Nil(t, output) + assert.Contains(t, err.Error(), ErrMsgGatewayResponseTimeout) + + var timeoutErr TimeoutError + assert.True(t, errors.As(err, &timeoutErr)) + var userErr UserError + assert.False(t, errors.As(err, &userErr)) + }) + + // The regression the grace period exists for: a classified response necessarily lands after the + // request timeout, and must still be honored rather than reported as a platform timeout. + t.Run("gateway response after request timeout is still classified", func(t *testing.T) { + proxy, _, readyCh := setupSendRequestTestWithConfig(t, common.ServiceConfig{ + GatewayConnectionConfig: common.GatewayConnectionConfig{ResponseGraceMs: 5_000}, + }) + + metadata := capabilities.RequestMetadata{ + WorkflowID: "wf1", + WorkflowExecutionID: "exec1", + WorkflowOwner: "owner1", + } + requestTimeout := 200 * time.Millisecond + input := &http.Request{ + Url: "http://example.com", + Method: "GET", + Body: []byte("test"), + Timeout: durationpb.New(requestTimeout), + CacheSettings: &http.CacheSettings{}, + } + + go func() { + id := <-readyCh + // The delay is the subject of the test, not a synchronization device. + time.Sleep(2 * requestTimeout) + simulateGatewayMessageWithFlags(t, proxy, id, 0, "", "endpoint timed out", true, true, false) + }() + + output, _, err := proxy.SendRequest(t.Context(), metadata, input, time.Now()) + require.Error(t, err) + require.Nil(t, output) + assert.Contains(t, err.Error(), "endpoint timed out") + assert.NotContains(t, err.Error(), ErrMsgGatewayResponseWait) + assert.NotContains(t, err.Error(), ErrMsgGatewayResponseTimeout) + + var userErr UserError + assert.True(t, errors.As(err, &userErr), "late gateway response must be classified, not timed out") + var timeoutErr TimeoutError + assert.False(t, errors.As(err, &timeoutErr)) }) } diff --git a/http_action/gateway/user_error.go b/http_action/gateway/user_error.go index de7a810dd..bae02f4ed 100644 --- a/http_action/gateway/user_error.go +++ b/http_action/gateway/user_error.go @@ -1,8 +1,11 @@ package gateway -// ErrMsgGatewayResponseWait is the stable prefix for timeouts and context cancellation while waiting on the gateway. +// ErrMsgGatewayResponseWait is the stable prefix for cancellation by the caller while waiting on the gateway. const ErrMsgGatewayResponseWait = "request canceled before gateway response" +// ErrMsgGatewayResponseTimeout is the stable prefix for the gateway failing to respond before the deadline. +const ErrMsgGatewayResponseTimeout = "gateway did not respond before deadline" + // UserError represents an error caused by user input or user endpoint // These errors should be surfaced to the user as public errors type UserError struct { @@ -20,3 +23,40 @@ func (e UserError) Unwrap() error { func NewUserError(err error) UserError { return UserError{err: err} } + +// TimeoutError means the gateway delivered no response before the deadline. Endpoint failures are +// reported back explicitly, so silence is a platform fault, not a user one. +type TimeoutError struct { + err error +} + +func (e TimeoutError) Error() string { + return e.err.Error() +} + +func (e TimeoutError) Unwrap() error { + return e.err +} + +func NewTimeoutError(err error) TimeoutError { + return TimeoutError{err: err} +} + +// CanceledError means the caller cancelled before a response arrived, e.g. engine shutdown or an +// exhausted execution budget. Not the user's fault, but caperrors.Origin has no "neither side" state, +// so it surfaces as a system error coded Canceled; alerts should exclude that code. +type CanceledError struct { + err error +} + +func (e CanceledError) Error() string { + return e.err.Error() +} + +func (e CanceledError) Unwrap() error { + return e.err +} + +func NewCanceledError(err error) CanceledError { + return CanceledError{err: err} +}