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/rpc/interceptor/mask_internal_error_test.go b/common/rpc/interceptor/mask_internal_error_test.go index 6f0bfd4767..6a7c9d0073 100644 --- a/common/rpc/interceptor/mask_internal_error_test.go +++ b/common/rpc/interceptor/mask_internal_error_test.go @@ -4,7 +4,7 @@ import ( "fmt" "testing" - "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "go.temporal.io/api/serviceerror" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/common" @@ -17,18 +17,20 @@ import ( ) func TestMaskUnknownOrInternalErrors(t *testing.T) { - statusOk := status.New(codes.OK, "OK") testMaskUnknownOrInternalErrors(t, statusOk, false) - statusUnknown := status.New(codes.Unknown, "Unknown") + statusCanceled := status.New(codes.Canceled, "Canceled message") + testMaskUnknownOrInternalErrors(t, statusCanceled, false) + + statusUnknown := status.New(codes.Unknown, "Unknown message") testMaskUnknownOrInternalErrors(t, statusUnknown, true) - statusInternal := status.New(codes.Internal, "Internal") + statusInternal := status.New(codes.Internal, "Internal message") testMaskUnknownOrInternalErrors(t, statusInternal, true) } -func testMaskUnknownOrInternalErrors(t *testing.T, st *status.Status, expectRelpace bool) { +func testMaskUnknownOrInternalErrors(t *testing.T, st *status.Status, expectReplace bool) { controller := gomock.NewController(t) mockRegistry := namespace.NewMockRegistry(controller) mockLogger := log.NewMockLogger(controller) @@ -36,27 +38,27 @@ func testMaskUnknownOrInternalErrors(t *testing.T, st *status.Status, expectRelp errorMaskInterceptor := NewMaskInternalErrorDetailsInterceptor( dynamicconfig.FrontendMaskInternalErrorDetails.Get(dc), mockRegistry, mockLogger) - err := serviceerror.FromStatus(st) - if expectRelpace { + err := st.Err() + if expectReplace { mockLogger.EXPECT().Error(gomock.Any(), gomock.Any()).Times(1) } - errorMessage := errorMaskInterceptor.maskUnknownOrInternalErrors(nil, "test", err) - if expectRelpace { + gotError := errorMaskInterceptor.maskUnknownOrInternalErrors(nil, "test", err) + if expectReplace { errorHash := common.ErrorHash(err) - expectedMessage := fmt.Sprintf("rpc error: code = %s desc = %s (%s)", st.Message(), errorFrontendMasked, errorHash) + expectedMessage := fmt.Sprintf( + "rpc error: code = %s desc = %s (%s)", + st.Code(), + errorFrontendMasked, + errorHash, + ) - assert.Equal(t, expectedMessage, errorMessage.Error()) + require.Equal(t, expectedMessage, gotError.Error()) } else { - if err == nil { - assert.Equal(t, errorMessage, nil) - } else { - assert.Equal(t, errorMessage.Error(), st.Message()) - } + require.Equal(t, err, gotError) } } func TestMaskInternalErrorDetailsInterceptor(t *testing.T) { - controller := gomock.NewController(t) mockRegistry := namespace.NewMockRegistry(controller) dc := dynamicconfig.NewNoopCollection() @@ -68,18 +70,18 @@ func TestMaskInternalErrorDetailsInterceptor(t *testing.T) { test_namespace := "test-namespace" req := &workflowservice.StartWorkflowExecutionRequest{Namespace: test_namespace} mockRegistry.EXPECT().GetNamespace(namespace.Name(test_namespace)).Return(&namespace.Namespace{}, nil).AnyTimes() - assert.True(t, errorMask.shouldMaskErrors(req)) + require.True(t, errorMask.shouldMaskErrors(req)) namespace_not_found := "namespace-not-found" req = &workflowservice.StartWorkflowExecutionRequest{Namespace: namespace_not_found} mockRegistry.EXPECT().GetNamespace(namespace.Name(namespace_not_found)).Return(nil, serviceerror.NewNamespaceNotFound("missing-namespace")) - assert.False(t, errorMask.shouldMaskErrors(req)) + require.False(t, errorMask.shouldMaskErrors(req)) empty_namespace := "" req = &workflowservice.StartWorkflowExecutionRequest{Namespace: empty_namespace} mockRegistry.EXPECT().GetNamespace(namespace.Name(empty_namespace)).Return(nil, serviceerror.NewNamespaceNotFound("missing-namespace")) - assert.False(t, errorMask.shouldMaskErrors(req)) + require.False(t, errorMask.shouldMaskErrors(req)) var ei any - assert.False(t, errorMask.shouldMaskErrors(ei)) + require.False(t, errorMask.shouldMaskErrors(ei)) } diff --git a/common/rpc/interceptor/service_error_interceptor.go b/common/rpc/interceptor/service_error_interceptor.go index 8a6f014d73..67abfc827f 100644 --- a/common/rpc/interceptor/service_error_interceptor.go +++ b/common/rpc/interceptor/service_error_interceptor.go @@ -6,6 +6,8 @@ import ( "go.temporal.io/api/serviceerror" "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/persistence/serialization" "go.temporal.io/server/common/util" "google.golang.org/grpc" @@ -16,13 +18,21 @@ const truncatedSuffix = "... " type ServiceErrorInterceptor struct { maxMessageLength dynamicconfig.IntPropertyFn + + metricsHandler metrics.Handler + logger log.Logger } func NewServiceErrorInterceptor( maxMessageLength dynamicconfig.IntPropertyFn, + metricsHandler metrics.Handler, + logger log.Logger, ) *ServiceErrorInterceptor { return &ServiceErrorInterceptor{ maxMessageLength: maxMessageLength, + + metricsHandler: metricsHandler, + logger: logger, } } @@ -32,7 +42,7 @@ func (i *ServiceErrorInterceptor) Intercept( _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler, ) (any, error) { - resp, err := handler(ctx, req) + resp, err := i.capturePanicHandler(ctx, req, handler) var deserializationError *serialization.DeserializationError var serializationError *serialization.SerializationError @@ -52,3 +62,12 @@ func (i *ServiceErrorInterceptor) Intercept( return resp, st.Err() } + +func (i *ServiceErrorInterceptor) capturePanicHandler( + ctx context.Context, + req any, + handler grpc.UnaryHandler, +) (_ any, retError error) { + defer metrics.CapturePanic(i.logger, i.metricsHandler, &retError) + return handler(ctx, req) +} diff --git a/common/rpc/interceptor/service_error_interceptor_test.go b/common/rpc/interceptor/service_error_interceptor_test.go index 4b86d04d23..1ac62527b4 100644 --- a/common/rpc/interceptor/service_error_interceptor_test.go +++ b/common/rpc/interceptor/service_error_interceptor_test.go @@ -2,6 +2,8 @@ package interceptor import ( "context" + "errors" + "fmt" "strings" "testing" @@ -9,7 +11,11 @@ import ( enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/api/serviceerror" "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/log/tag" + "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/persistence/serialization" + "go.uber.org/mock/gomock" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -31,7 +37,12 @@ func (e *ErrorWithoutStatus) Error() string { // Error returns string message. func TestServiceErrorInterceptorUnknown(t *testing.T) { - interceptor := NewServiceErrorInterceptor(dynamicconfig.GetIntPropertyFn(testMaxMessageLength)) + ctrl := gomock.NewController(t) + interceptor := NewServiceErrorInterceptor( + dynamicconfig.GetIntPropertyFn(testMaxMessageLength), + metrics.NewMockHandler(ctrl), + log.NewTestLogger(), + ) _, err := interceptor.Intercept(t.Context(), nil, nil, func(ctx context.Context, req any) (any, error) { @@ -54,7 +65,12 @@ func TestServiceErrorInterceptorUnknown(t *testing.T) { } func TestServiceErrorInterceptorSer(t *testing.T) { - interceptor := NewServiceErrorInterceptor(dynamicconfig.GetIntPropertyFn(testMaxMessageLength)) + ctrl := gomock.NewController(t) + interceptor := NewServiceErrorInterceptor( + dynamicconfig.GetIntPropertyFn(testMaxMessageLength), + metrics.NewMockHandler(ctrl), + log.NewTestLogger(), + ) serErrors := []error{ serialization.NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, nil), serialization.NewSerializationError(enumspb.ENCODING_TYPE_PROTO3, nil), @@ -69,7 +85,12 @@ func TestServiceErrorInterceptorSer(t *testing.T) { } func TestServiceErrorInterceptorTruncation(t *testing.T) { - interceptor := NewServiceErrorInterceptor(dynamicconfig.GetIntPropertyFn(testMaxMessageLength)) + ctrl := gomock.NewController(t) + interceptor := NewServiceErrorInterceptor( + dynamicconfig.GetIntPropertyFn(testMaxMessageLength), + metrics.NewMockHandler(ctrl), + log.NewTestLogger(), + ) t.Run("nil error is not affected", func(t *testing.T) { _, err := interceptor.Intercept(t.Context(), nil, nil, @@ -146,3 +167,93 @@ func TestServiceErrorInterceptorTruncation(t *testing.T) { } }) } + +func TestServiceErrorInterceptorPanic(t *testing.T) { + testCases := []struct { + name string + panicObj any + errMessage string + }{ + { + name: "panic with error is converted to internal error", + panicObj: errors.New("panic error message"), + errMessage: "panic error message", + }, + { + name: "panic with non-error value is converted to internal error", + panicObj: "something went wrong", + errMessage: "panic: something went wrong", + }, + { + name: "panic with service error is still converted to internal error", + panicObj: serviceerror.NewNotFound("not found message"), + errMessage: "not found message", + }, + { + name: "captured panic message is truncated", + panicObj: errors.New(strings.Repeat("a", testMaxMessageLength+100)), + errMessage: strings.Repeat("a", testMaxMessageLength-len(truncatedSuffix)) + truncatedSuffix, + }, + { + name: "no panic does not log", + panicObj: nil, + errMessage: "", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + metricsHandlerMock := metrics.NewMockHandler(ctrl) + loggerMock := log.NewMockLogger(ctrl) + interceptor := NewServiceErrorInterceptor( + dynamicconfig.GetIntPropertyFn(testMaxMessageLength), + metricsHandlerMock, + loggerMock, + ) + + var loggedTags []tag.Tag + if tc.panicObj != nil { + counterMock := metrics.NewMockCounterIface(ctrl) + counterMock.EXPECT().Record(int64(1)) + metricsHandlerMock.EXPECT().Counter(metrics.ServicePanic.Name()).Return(counterMock) + loggerMock.EXPECT().Error("Panic is captured", gomock.Any(), gomock.Any()). + Do(func(_ string, tags ...tag.Tag) { + loggedTags = tags + }). + Times(1) + } + + resp, err := interceptor.Intercept(t.Context(), nil, nil, + func(_ context.Context, _ any) (any, error) { + if tc.panicObj != nil { + panic(tc.panicObj) + } + return "ok", nil + }) + + if tc.panicObj != nil { + expectedMessage := fmt.Sprintf( + "rpc error: code = Internal desc = %s", + tc.errMessage, + ) + require.Nil(t, resp) + require.Error(t, err) + require.Equal(t, codes.Internal, status.Code(err)) + require.Equal(t, expectedMessage, err.Error()) + + // Logs contains the stack trace + tagsByKey := make(map[string]tag.Tag, len(loggedTags)) + for _, tg := range loggedTags { + tagsByKey[tg.Key()] = tg + } + require.Contains(t, tagsByKey, "sys-stack-trace") + require.Contains(t, tagsByKey["sys-stack-trace"].Value(), "service_error_interceptor_test.go") + require.Contains(t, tagsByKey, "error") + } else { + require.Equal(t, "ok", resp) + require.NoError(t, err) + } + }) + } +} 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) }, })) diff --git a/service/frontend/fx.go b/service/frontend/fx.go index 15b12f1575..651551de21 100644 --- a/service/frontend/fx.go +++ b/service/frontend/fx.go @@ -362,9 +362,13 @@ func ConfigProvider( func ServiceErrorInterceptorProvider( dc *dynamicconfig.Collection, + metricsHandler metrics.Handler, + logger log.Logger, ) *interceptor.ServiceErrorInterceptor { return interceptor.NewServiceErrorInterceptor( dynamicconfig.MaxServiceErrorMessageLength.Get(dc), + metricsHandler, + logger, ) } diff --git a/service/frontend/fx_test.go b/service/frontend/fx_test.go index 0100ceb792..39a7ea121d 100644 --- a/service/frontend/fx_test.go +++ b/service/frontend/fx_test.go @@ -33,6 +33,8 @@ type rateLimitInterceptorTestCase struct { name string // t is the test object t *testing.T + // ctrl is the mock controller + ctrl *gomock.Controller // globalRPSLimit is the global RPS limit for all frontend hosts globalRPSLimit int // perInstanceRPSLimit is the RPS limit for each frontend host @@ -183,7 +185,7 @@ func TestRateLimitInterceptorProvider(t *testing.T) { tc.perInstanceRPSLimit = highPerInstanceRPSLimit tc.operatorRPSRatio = operatorRPSRatio tc.expectRateLimit = false - serviceResolver := membership.NewMockServiceResolver(gomock.NewController(tc.t)) + serviceResolver := membership.NewMockServiceResolver(tc.ctrl) serviceResolver.EXPECT().AvailableMemberCount().Return(0).AnyTimes() tc.serviceResolver = serviceResolver }, @@ -196,11 +198,11 @@ func TestRateLimitInterceptorProvider(t *testing.T) { tc.numRequests = 10 tc.t = t + tc.ctrl = gomock.NewController(t) { // Create a mock service resolver which returns the number of frontend hosts. // This may be overridden by the test case. - ctrl := gomock.NewController(t) - serviceResolver := membership.NewMockServiceResolver(ctrl) + serviceResolver := membership.NewMockServiceResolver(tc.ctrl) serviceResolver.EXPECT().AvailableMemberCount().Return(numHosts).AnyTimes() tc.serviceResolver = serviceResolver } @@ -208,6 +210,8 @@ func TestRateLimitInterceptorProvider(t *testing.T) { serviceErrorInterceptor := interceptor.NewServiceErrorInterceptor( dynamicconfig.GetIntPropertyFn(4000), + metrics.NewMockHandler(tc.ctrl), + log.NewTestLogger(), ) // Create a rate limit interceptor which uses the per-instance and global RPS limits from the test case. @@ -565,17 +569,20 @@ func TestNamespaceRateLimitInterceptorProvider(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { t.Parallel() + ctrl := gomock.NewController(t) namespaceName := "test-namespace" - mockRegistry := namespace.NewMockRegistry(gomock.NewController(t)) + mockRegistry := namespace.NewMockRegistry(ctrl) mockRegistry.EXPECT().GetNamespace(namespace.Name(namespaceName)).Return(&namespace.Namespace{}, nil).AnyTimes() - serviceResolver := membership.NewMockServiceResolver(gomock.NewController(t)) + serviceResolver := membership.NewMockServiceResolver(ctrl) serviceResolver.EXPECT().AvailableMemberCount().Return(tc.frontendServiceCount).AnyTimes() config := getTestConfig(tc) serviceErrorInterceptor := interceptor.NewServiceErrorInterceptor( dynamicconfig.GetIntPropertyFn(4000), + metrics.NewMockHandler(ctrl), + log.NewTestLogger(), ) // Create a rate limit interceptor. @@ -752,11 +759,12 @@ func TestNamespaceRateLimitMetrics(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { t.Parallel() + ctrl := gomock.NewController(t) testNS := "test_namespace" - mockRegistry := namespace.NewMockRegistry(gomock.NewController(t)) + mockRegistry := namespace.NewMockRegistry(ctrl) mockRegistry.EXPECT().GetNamespace(namespace.Name(testNS)).Return(&namespace.Namespace{}, nil).AnyTimes() - serviceResolver := membership.NewMockServiceResolver(gomock.NewController(t)) + serviceResolver := membership.NewMockServiceResolver(ctrl) serviceResolver.EXPECT().AvailableMemberCount().Return(tc.frontendServiceCount).AnyTimes() metricsHandler := metricstest.NewCaptureHandler() capture := metricsHandler.StartCapture() @@ -779,6 +787,8 @@ func TestNamespaceRateLimitMetrics(t *testing.T) { serviceErrorInterceptor := interceptor.NewServiceErrorInterceptor( dynamicconfig.GetIntPropertyFn(4000), + metrics.NewMockHandler(ctrl), + log.NewTestLogger(), ) // Create a rate limit interceptor which uses the per-instance and global RPS limits from the test case. diff --git a/service/history/fx.go b/service/history/fx.go index 54d0432c51..dcf233185f 100644 --- a/service/history/fx.go +++ b/service/history/fx.go @@ -249,9 +249,13 @@ func ConfigProvider( func ServiceErrorInterceptorProvider( dc *dynamicconfig.Collection, + metricsHandler metrics.Handler, + logger log.Logger, ) *interceptor.ServiceErrorInterceptor { return interceptor.NewServiceErrorInterceptor( dynamicconfig.MaxServiceErrorMessageLength.Get(dc), + metricsHandler, + logger, ) } diff --git a/service/matching/fx.go b/service/matching/fx.go index 62efeebfd6..c3d1a791cf 100644 --- a/service/matching/fx.go +++ b/service/matching/fx.go @@ -71,9 +71,13 @@ func ConfigProvider( func ServiceErrorInterceptorProvider( dc *dynamicconfig.Collection, + metricsHandler metrics.Handler, + logger log.Logger, ) *interceptor.ServiceErrorInterceptor { return interceptor.NewServiceErrorInterceptor( dynamicconfig.MaxServiceErrorMessageLength.Get(dc), + metricsHandler, + logger, ) } diff --git a/tests/schedule_test.go b/tests/schedule_test.go index 2e4db2068d..06f68e20b2 100644 --- a/tests/schedule_test.go +++ b/tests/schedule_test.go @@ -417,7 +417,6 @@ func TestScheduleCHASM(t *testing.T) { t.Run("TestSkipsWorkflowSentinelWhenDisabled", func(t *testing.T) { t.Parallel(); testSkipsWorkflowSentinelWhenDisabled(t, newContext) }) t.Run("TestLargeScheduleID", func(t *testing.T) { t.Parallel(); testLargeScheduleID(t, newContext) }) t.Run("TestUpdateScheduleMemo", func(t *testing.T) { t.Parallel(); testUpdateScheduleMemo(t, newContext) }) - t.Run("TestUpdateScheduleMemoOnly", func(t *testing.T) { t.Parallel(); testUpdateScheduleMemoOnly(t, newContext) }) t.Run("TestStateSizeBytesReported", func(t *testing.T) { t.Parallel(); testStateSizeBytesReported(t, newContext) }) t.Run("TestBufferOverrunDropsActions", func(t *testing.T) { t.Parallel(); testBufferOverrunDropsActions(t, newContext) }) t.Run("TestDescribeCatchupWindowAfterCreateAndUpdate", func(t *testing.T) { @@ -4131,80 +4130,6 @@ func testUpdateScheduleMemoRejected(t *testing.T, newContext contextFactory) { require.Contains(t, err.Error(), "memo updates are not supported on workflow-backed schedules") } -func testUpdateScheduleMemoOnly(t *testing.T, newContext contextFactory) { - // UpdateScheduleRequest uses replace semantics for the schedule field, so omitting it - // causes the schedule to be unset. Memo-only updates require the server to skip replacing - // the schedule when the field is nil, similar to how memo and search_attributes are handled. - t.Skip("memo-only updates not yet supported: omitting the schedule field unsets the schedule") - - s := newScheduleEnv(t, scheduleCommonOpts(t)...) - - sid := "sched-test-update-memo-only" - wid := "sched-test-update-memo-only-wf" - wt := "sched-test-update-memo-only-wt" - - s.SdkWorker().RegisterWorkflowWithOptions( - func(ctx workflow.Context) error { return nil }, - workflow.RegisterOptions{Name: wt}, - ) - - schedule := &schedulepb.Schedule{ - Spec: &schedulepb.ScheduleSpec{ - Interval: []*schedulepb.IntervalSpec{ - {Interval: durationpb.New(1 * time.Hour)}, - }, - }, - Action: &schedulepb.ScheduleAction{ - Action: &schedulepb.ScheduleAction_StartWorkflow{ - StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{ - WorkflowId: wid, - WorkflowType: &commonpb.WorkflowType{Name: wt}, - TaskQueue: &taskqueuepb.TaskQueue{Name: s.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, - }, - }, - }, - } - - // Create schedule with initial memo. - memo1 := payload.EncodeString("val1") - ctx := newContext(s.Context()) - _, err := s.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{ - Namespace: s.Namespace().String(), - ScheduleId: sid, - Schedule: schedule, - Identity: "test", - RequestId: uuid.NewString(), - Memo: &commonpb.Memo{ - Fields: map[string]*commonpb.Payload{"key1": memo1}, - }, - }) - require.NoError(t, err) - - // Update only memo, without setting the schedule field. - memo2 := payload.EncodeString("val2") - _, err = s.FrontendClient().UpdateSchedule(newContext(s.Context()), &workflowservice.UpdateScheduleRequest{ - Namespace: s.Namespace().String(), - ScheduleId: sid, - Identity: "test", - RequestId: uuid.NewString(), - Memo: &commonpb.Memo{ - Fields: map[string]*commonpb.Payload{"key1": memo2}, - }, - }) - require.NoError(t, err) - - // Verify memo was updated and schedule is still intact. - describeResp, err := s.FrontendClient().DescribeSchedule(newContext(s.Context()), &workflowservice.DescribeScheduleRequest{ - Namespace: s.Namespace().String(), - ScheduleId: sid, - }) - require.NoError(t, err) - require.Equal(t, memo2.Data, describeResp.Memo.Fields["key1"].Data, "memo should be updated") - require.NotNil(t, describeResp.Schedule.Spec, "schedule spec should not be nil") - require.NotEmpty(t, describeResp.Schedule.Spec.Interval, "schedule spec intervals should be preserved") - require.NotNil(t, describeResp.Schedule.Action, "schedule action should be preserved") -} - func testCHASMUnpauseResumesProcessing(t *testing.T, newContext contextFactory) { s := newScheduleEnv(t, scheduleCommonOpts(t)...) @@ -4637,18 +4562,8 @@ func testBackfillReprocessesCompletedAction( // testBackfillWithBufferOneOverlap pins the expected behavior of BUFFER_ONE // over a multi-tick backfill: the first start runs immediately, exactly one // follow-up is buffered (Attempt=-1 deferred), the rest are dropped, and the -// deferred one runs once the first completes. Currently SKIPPED: fails on -// both V1 and CHASM because the deferred start never gets re-enabled after -// the running workflow completes. The first start fires, the rest never run. -// Likely a real bug in the BUFFER_ONE + backfill (Manual=true) interaction - -// recordCompletedAction's re-enable loop on Attempt==-1 may not be running -// against backfill-buffered starts. Worth a separate investigation. +// deferred one runs once the first completes. func testBackfillWithBufferOneOverlap(t *testing.T, newContext contextFactory) { - // TODO(temporalio/temporal): track removing this skip once the BUFFER_ONE - // backfill deferred re-enable path is fixed. Verify by running: - // go test ./tests/ -run 'TestScheduleCHASM/Backfill/BufferOneOverlap' -v - t.Skip("BUFFER_ONE backfill deferred re-enable is broken on both V1 and CHASM; see test doc") - s := newScheduleEnv(t, scheduleCommonOpts(t)...) sid := testcore.RandomizeStr("sched-backfill-buffer-one")