Merge branch 'main' into kannan/fix-pns-worker-data-race

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{

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")
+}

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))
 }

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 = "... <truncated>"
 
 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)
+}

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)
+			}
+		})
+	}
+}

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 })

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

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)
 				},
 			}))
 

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,
 	)
 }
 

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.

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,
 	)
 }
 

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,
 	)
 }
 

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")