diff --git a/chasm/lib/nexusoperation/task_handler_helpers.go b/chasm/lib/nexusoperation/task_handler_helpers.go index 06a3d04019..66c783483d 100644 --- a/chasm/lib/nexusoperation/task_handler_helpers.go +++ b/chasm/lib/nexusoperation/task_handler_helpers.go @@ -141,7 +141,7 @@ func cancelCallOutcomeTag(callCtx context.Context, callErr error) string { // Always returns a non-nil failure. func callErrorToFailure(callErr error) (*failurepb.Failure, bool, error) { if serviceErr, ok := errors.AsType[serviceerror.ServiceError](callErr); ok { - retryable := common.IsRetryableRPCError(callErr) + retryable := common.IsRetryableRPCError(serviceErr) failure := &failurepb.Failure{ Message: fmt.Sprintf("%s: %s", strings.Replace(fmt.Sprintf("%T", serviceErr), "*serviceerror.", "", 1), serviceErr.Error()), FailureInfo: &failurepb.Failure_ServerFailureInfo{ @@ -233,7 +233,7 @@ func newInvocationResult( } if serviceErr, ok := errors.AsType[serviceerror.ServiceError](callErr); ok { - retryable := common.IsRetryableRPCError(callErr) + retryable := common.IsRetryableRPCError(serviceErr) failure := &failurepb.Failure{ Message: fmt.Sprintf("%s: %s", strings.Replace(fmt.Sprintf("%T", serviceErr), "*serviceerror.", "", 1), serviceErr.Error()), FailureInfo: &failurepb.Failure_ServerFailureInfo{ diff --git a/chasm/lib/nexusoperation/task_handler_helpers_test.go b/chasm/lib/nexusoperation/task_handler_helpers_test.go new file mode 100644 index 0000000000..e7a6788d4f --- /dev/null +++ b/chasm/lib/nexusoperation/task_handler_helpers_test.go @@ -0,0 +1,37 @@ +package nexusoperation + +import ( + "net/url" + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/serviceerror" +) + +// wrappedUnavailable returns a transient serviceerror reachable only via Unwrap, +// as an HTTP client wraps transport errors. +func wrappedUnavailable() error { + return &url.Error{ + Op: "Post", + URL: "https://internal", + Err: serviceerror.NewUnavailable("no frontend host to route request to"), + } +} + +func TestCallErrorToFailure_RetriesTransientServiceErrorEvenWhenWrapped(t *testing.T) { + t.Parallel() + + failure, retryable, err := callErrorToFailure(wrappedUnavailable()) + require.NoError(t, err) + require.True(t, retryable, "wrapped Unavailable must be classified as retryable") + require.False(t, failure.GetServerFailureInfo().GetNonRetryable()) +} + +func TestNewInvocationResult_RetriesTransientServiceErrorEvenWhenWrapped(t *testing.T) { + t.Parallel() + + result, err := newInvocationResult(nil, wrappedUnavailable()) + require.NoError(t, err) + require.IsType(t, invocationResultRetry{}, result, + "wrapped Unavailable must produce a retry result, not a terminal failure") +} diff --git a/common/util.go b/common/util.go index 750dabac74..87e02bbc63 100644 --- a/common/util.go +++ b/common/util.go @@ -763,6 +763,9 @@ func getFieldNameFromStruct(structPtr any, fieldPtr any) (string, error) { } // IsRetryableRPCError checks if the error is a retryable gRPC error. +// +// This does not unwrap err: If err may be wrapped (e.g. by an HTTP client, which wraps +// RoundTripper errors in *url.Error), unwrap it yourself and pass the unwrapped error here. func IsRetryableRPCError(err error) bool { var st *status.Status stGetter, ok := err.(interface{ Status() *status.Status }) diff --git a/components/nexusoperations/executors.go b/components/nexusoperations/executors.go index 2713c65747..b7eb7bb272 100644 --- a/components/nexusoperations/executors.go +++ b/components/nexusoperations/executors.go @@ -535,7 +535,7 @@ func (e taskExecutor) handleStartOperationError(env hsm.Environment, node *hsm.N switch { case errors.As(callErr, &serviceErr): - if !common.IsRetryableRPCError(callErr) { + if !common.IsRetryableRPCError(serviceErr) { return handleNonRetryableStartOperationError(node, operation, callErr) } // Fall through all uncaught errors to retryable diff --git a/components/nexusoperations/executors_test.go b/components/nexusoperations/executors_test.go index b5faee1b0a..41d5bc55e3 100644 --- a/components/nexusoperations/executors_test.go +++ b/components/nexusoperations/executors_test.go @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "net/http" + "net/url" "testing" "time" @@ -99,6 +101,9 @@ func TestProcessInvocationTask(t *testing.T) { startToCloseTimeout time.Duration schedToStartTimeout time.Duration destinationDown bool + // httpCallerErr, when set, fails the outbound HTTP call with this error instead of + // performing a real request. + httpCallerErr error }{ { name: "async start", @@ -379,6 +384,22 @@ func TestProcessInvocationTask(t *testing.T) { require.Empty(t, events) }, }, + { + name: "transient service error wrapped by HTTP caller", + requestTimeout: time.Hour, + httpCallerErr: &url.Error{ + Op: "Post", + URL: "http://unavailable", + Err: serviceerror.NewUnavailable("no frontend host to route request to"), + }, + expectedMetricOutcome: "service-error:Unavailable", + checkOutcome: func(t *testing.T, op nexusoperations.Operation, events []*historypb.HistoryEvent) { + require.Equal(t, enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF, op.State()) + require.False(t, op.LastAttemptFailure.GetServerFailureInfo().GetNonRetryable()) + require.Contains(t, op.LastAttemptFailure.Message, "Unavailable") + require.Empty(t, events) + }, + }, { name: "invocation timeout by request timeout", requestTimeout: 2 * time.Millisecond, @@ -655,11 +676,17 @@ func TestProcessInvocationTask(t *testing.T) { Logger: log.NewNoopLogger(), EndpointRegistry: endpointReg, ClientProvider: func(ctx context.Context, namespaceID string, entry *persistencespb.NexusEndpointEntry, service string) (*nexusrpc.HTTPClient, error) { - return nexusrpc.NewHTTPClient(nexusrpc.HTTPClientOptions{ + options := nexusrpc.HTTPClientOptions{ BaseURL: "http://" + listenAddr, Service: service, Serializer: commonnexus.PayloadSerializer, - }) + } + if tc.httpCallerErr != nil { + options.HTTPCaller = func(*http.Request) (*http.Response, error) { + return nil, tc.httpCallerErr + } + } + return nexusrpc.NewHTTPClient(options) }, }))