Prevent malformed retry delays from poisoning Nexus completions (#11617)

## What changed?
Validate the retry delay is a valid proto duration.

## Why?
Prevent malformed retry delays from poisoning Nexus completions.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [x] added new functional test(s)

## Potential risks
Potentially users could have been sending an invalid proto duration,
unclear how exactly, and now we would fail their request.

service/frontend/workflow_handler.go

@@ -19,6 +19,7 @@ import (
 	commonpb "go.temporal.io/api/common/v1"
 	deploymentpb "go.temporal.io/api/deployment/v1"
 	enumspb "go.temporal.io/api/enums/v1"
+	failurepb "go.temporal.io/api/failure/v1"
 	filterpb "go.temporal.io/api/filter/v1"
 	historypb "go.temporal.io/api/history/v1"
 	querypb "go.temporal.io/api/query/v1"
@@ -1818,6 +1819,17 @@ func (wh *WorkflowHandler) RespondActivityTaskCompletedById(ctx context.Context,
 	return &workflowservice.RespondActivityTaskCompletedByIdResponse{}, nil
 }
 
+func validateActivityFailureNextRetryDelays(activityFailure *failurepb.Failure) error {
+	for current := activityFailure; current != nil; current = current.GetCause() {
+		if delay := current.GetApplicationFailureInfo().GetNextRetryDelay(); delay != nil {
+			if err := delay.CheckValid(); err != nil {
+				return serviceerror.NewInvalidArgumentf("NextRetryDelay is not a valid duration: %v", err)
+			}
+		}
+	}
+	return nil
+}
+
 // RespondActivityTaskFailed is called by application worker when it is done processing an ActivityTask.  It will
 // result in a new 'ActivityTaskFailed' event being written to the workflow history and a new WorkflowTask
 // created for the workflow instance so new commands could be made.  Use the 'taskToken' provided as response of
@@ -1852,6 +1864,9 @@ func (wh *WorkflowHandler) RespondActivityTaskFailed(
 	if request.GetFailure() != nil && request.GetFailure().GetApplicationFailureInfo() == nil {
 		return nil, errFailureMustHaveApplicationFailureInfo
 	}
+	if err := validateActivityFailureNextRetryDelays(request.GetFailure()); err != nil {
+		return nil, err
+	}
 
 	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
 		return nil, errIdentityTooLong
@@ -1987,6 +2002,9 @@ func (wh *WorkflowHandler) RespondActivityTaskFailedById(ctx context.Context, re
 	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())
 
 	response := workflowservice.RespondActivityTaskFailedByIdResponse{}
+	if err := validateActivityFailureNextRetryDelays(request.GetFailure()); err != nil {
+		return nil, err
+	}
 
 	if request.GetLastHeartbeatDetails() != nil {
 		if err := common.CheckEventBlobSizeLimit(

service/frontend/workflow_handler_test.go

@@ -5331,6 +5331,90 @@ func (s *WorkflowHandlerSuite) TestPatchSchedule_ValidationAndErrors() {
 // carry (10000 years). Anything beyond it fails durationpb's CheckValid.
 const maxProtoDurationSeconds = int64(315576000000)
 
+func TestValidateActivityFailureNextRetryDelays(t *testing.T) {
+	applicationFailure := func(delay *durationpb.Duration, cause *failurepb.Failure) *failurepb.Failure {
+		return &failurepb.Failure{
+			Cause: cause,
+			FailureInfo: &failurepb.Failure_ApplicationFailureInfo{
+				ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{NextRetryDelay: delay},
+			},
+		}
+	}
+
+	testCases := []struct {
+		name    string
+		failure *failurepb.Failure
+		wantErr bool
+	}{
+		{name: "nil failure"},
+		{name: "nil retry delay", failure: applicationFailure(nil, nil)},
+		{name: "zero retry delay", failure: applicationFailure(durationpb.New(0), nil)},
+		{name: "negative retry delay", failure: applicationFailure(durationpb.New(-time.Second), nil)},
+		{
+			name:    "maximum duration",
+			failure: applicationFailure(&durationpb.Duration{Seconds: maxProtoDurationSeconds}, nil),
+		},
+		{
+			name:    "minimum duration",
+			failure: applicationFailure(&durationpb.Duration{Seconds: -maxProtoDurationSeconds}, nil),
+		},
+		{
+			name:    "mismatched positive seconds",
+			failure: applicationFailure(&durationpb.Duration{Seconds: 1, Nanos: -1}, nil),
+			wantErr: true,
+		},
+		{
+			name:    "mismatched negative seconds",
+			failure: applicationFailure(&durationpb.Duration{Seconds: -1, Nanos: 1}, nil),
+			wantErr: true,
+		},
+		{
+			name:    "nanos above range",
+			failure: applicationFailure(&durationpb.Duration{Nanos: 1000000000}, nil),
+			wantErr: true,
+		},
+		{
+			name:    "nanos below range",
+			failure: applicationFailure(&durationpb.Duration{Nanos: -1000000000}, nil),
+			wantErr: true,
+		},
+		{
+			name:    "seconds above range",
+			failure: applicationFailure(&durationpb.Duration{Seconds: maxProtoDurationSeconds + 1}, nil),
+			wantErr: true,
+		},
+		{
+			name:    "seconds below range",
+			failure: applicationFailure(&durationpb.Duration{Seconds: -maxProtoDurationSeconds - 1}, nil),
+			wantErr: true,
+		},
+		{
+			name: "invalid nested retry delay",
+			failure: applicationFailure(nil, &failurepb.Failure{
+				FailureInfo: &failurepb.Failure_TimeoutFailureInfo{
+					TimeoutFailureInfo: &failurepb.TimeoutFailureInfo{},
+				},
+				Cause: applicationFailure(&durationpb.Duration{Seconds: 1, Nanos: -1}, nil),
+			}),
+			wantErr: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := validateActivityFailureNextRetryDelays(tc.failure)
+			if !tc.wantErr {
+				require.NoError(t, err)
+				return
+			}
+
+			var invalidArgument *serviceerror.InvalidArgument
+			require.ErrorAs(t, err, &invalidArgument)
+			require.ErrorContains(t, err, "NextRetryDelay is not a valid duration")
+		})
+	}
+}
+
 // newScheduleSpecHandler builds the minimum WorkflowHandler that canonicalizeScheduleSpec
 // needs: a config, a spec builder, and a logger for the non-enforcing path. configure is
 // nil to exercise the shipped dynamic config defaults.

tests/activity_standalone_test.go

@@ -1492,6 +1492,69 @@ func (s *standaloneActivityTestSuite) TestFail() {
 		env.validateFailure(s.Context(), t, activityID, runID, nil, env.Tv().WorkerIdentity())
 	})
 
+	t.Run("InvalidNextRetryDelay", func(t *testing.T) {
+		for _, tc := range []struct {
+			name string
+			byID bool
+		}{
+			{name: "ByToken"},
+			{name: "ByID", byID: true},
+		} {
+			t.Run(tc.name, func(t *testing.T) {
+				activityID := testcore.RandomizeStr(t.Name())
+				taskQueue := testcore.RandomizeStr(t.Name())
+
+				startResp := env.startAndValidateActivity(s.Context(), t, activityID, taskQueue)
+				runID := startResp.RunId
+				pollResp := env.pollActivityTaskAndValidate(s.Context(), t, activityID, taskQueue, runID)
+
+				invalidFailure := &failurepb.Failure{
+					Message: "invalid retry delay",
+					FailureInfo: &failurepb.Failure_ApplicationFailureInfo{
+						ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
+							NonRetryable:   true,
+							NextRetryDelay: &durationpb.Duration{Seconds: 1, Nanos: -1},
+						},
+					},
+				}
+
+				respondFailed := func(failure *failurepb.Failure) error {
+					if tc.byID {
+						_, err := env.FrontendClient().RespondActivityTaskFailedById(s.Context(), &workflowservice.RespondActivityTaskFailedByIdRequest{
+							Namespace:  env.Namespace().String(),
+							RunId:      runID,
+							ActivityId: activityID,
+							Failure:    failure,
+						})
+						return err
+					}
+					_, err := env.FrontendClient().RespondActivityTaskFailed(s.Context(), &workflowservice.RespondActivityTaskFailedRequest{
+						Namespace: env.Namespace().String(),
+						TaskToken: pollResp.TaskToken,
+						Failure:   failure,
+					})
+					return err
+				}
+
+				err := respondFailed(invalidFailure)
+				var invalidArgument *serviceerror.InvalidArgument
+				require.ErrorAs(t, err, &invalidArgument)
+				require.ErrorContains(t, err, "NextRetryDelay is not a valid duration")
+
+				describeResp, err := env.FrontendClient().DescribeActivityExecution(s.Context(), &workflowservice.DescribeActivityExecutionRequest{
+					Namespace:  env.Namespace().String(),
+					ActivityId: activityID,
+					RunId:      runID,
+				})
+				require.NoError(t, err)
+				require.Equal(t, enumspb.ACTIVITY_EXECUTION_STATUS_RUNNING, describeResp.GetInfo().GetStatus())
+
+				require.NoError(t, respondFailed(defaultFailure))
+				env.validateFailure(s.Context(), t, activityID, runID, nil, "")
+			})
+		}
+	})
+
 	t.Run("StaleToken", func(t *testing.T) {
 		activityID := testcore.RandomizeStr(t.Name())
 		taskQueue := testcore.RandomizeStr(t.Name())