Prevent workflow task and activity task generation when workflow is paused (temporalio#8687)

## What changed?
- Prevents workflow task generation when paused (ex: signals, activity
completions, child workflow completions etc)
- Prevents activity task generation when a workflow is paused and an
inflight workflow task comes back with start activity commands.
- Prevents eager activity starts when the workflow is paused.

## Why?
When a workflow is paused, we should not generate any more new activity
and workflow tasks.

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


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Prevents workflow/activity task creation and certain API operations
when a workflow is paused, adds safeguards and tests across history,
NDC, and task refresh paths.
> 
> - **Behavior/State**:
> - Add `MutableState.IsWorkflowExecutionStatusPaused()` and honor
paused status across scheduling paths.
>   - `workflow/util.ScheduleWorkflowTask` returns early if paused.
> - **API Handlers**:
> - `RespondWorkflowTaskCompleted`: reject `ForceCreateNewWorkflowTask`
when paused; block eager activity and bypass activity task gen when
paused.
> - `SignalWithStartWorkflow`: do not schedule a new workflow task if
paused.
>   - `UpdateWorkflow`: reject updates while paused.
>   - Activity failure/respond paths: avoid new WFT when paused.
> - **NDC/Replication**:
> - Reapply events and buffered-event flusher: skip scheduling WFT if
paused.
> - `ndc.Workflow.FlushBufferedEvents`: don't schedule WFT when paused.
> - **Task Refresher**:
> - Skip refreshing WFT/activity tasks when paused; allow
timers/search-attr updates; skip close tasks for running/paused.
> - **Interfaces/Mocks**:
> - Add `IsWorkflowExecutionStatusPaused` to `MutableState` interface
and update mocks.
> - **Tests**:
> - New unit tests for paused behavior in handlers, NDC, task refresher,
and transfer queue; functional tests for pause/unpause workflow
behavior.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
17c11c6. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: gow

service/history/api/recordchildworkflowcompleted/api_test.go

@@ -72,6 +72,7 @@ func Test_Recordchildworkflowcompleted_WithForwards(t *testing.T) {
 
 	newParentMutableState := historyi.NewMockMutableState(ctrl)
 	newParentMutableState.EXPECT().IsWorkflowExecutionRunning().Return(true)
+	newParentMutableState.EXPECT().IsWorkflowExecutionStatusPaused().Return(false)
 	newParentMutableState.EXPECT().GetNextEventID().Return(int64(10))
 	newParentMutableState.EXPECT().AddChildWorkflowExecutionCompletedEvent(anyArg, anyArg, anyArg).Return(nil, nil)
 	childExecutionInfo := &persistencespb.ChildExecutionInfo{

service/history/api/respondactivitytaskfailed/api_test.go

@@ -305,6 +305,7 @@ func (s *workflowSuite) Test_NoMoreRetriesAndMutableStateHasNoPendingTasks_WillR
 		request.FailedRequest.GetIdentity(),
 		request.FailedRequest.WorkerVersion,
 	).Return(nil, nil)
+	s.currentMutableState.EXPECT().IsWorkflowExecutionStatusPaused().Return(false)
 	s.currentMutableState.EXPECT().AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_NORMAL)
 	s.currentMutableState.EXPECT().GetEffectiveVersioningBehavior().Return(enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED)
 	s.workflowContext.EXPECT().UpdateWorkflowExecutionAsActive(ctx, s.shardContext).Return(nil)

service/history/api/respondworkflowtaskcompleted/api.go

@@ -195,6 +195,13 @@ func (handler *WorkflowTaskCompletedHandler) Invoke(
 		return nil, serviceerror.NewNotFound("Workflow task not found.")
 	}
 
+	// We don't accept the request to create a new workflow task if the workflow is paused.
+	if ms.IsWorkflowExecutionStatusPaused() && request.GetForceCreateNewWorkflowTask() {
+		// Mutable state wasn't changed yet and doesn't have to be cleared.
+		releaseLeaseWithError = false
+		return nil, serviceerror.NewFailedPrecondition("Workflow is paused and force create new workflow task is not allowed.")
+	}
+
 	behavior := request.GetVersioningBehavior()
 	deployment := worker_versioning.DeploymentFromDeploymentVersion(worker_versioning.DeploymentVersionFromOptions(request.GetDeploymentOptions()))
 	//nolint:staticcheck // SA1019 deprecated Deployment will clean up later

service/history/api/respondworkflowtaskcompleted/api_test.go

@@ -435,13 +435,39 @@ func (s *WorkflowTaskCompletedHandlerSuite) TestUpdateWorkflow() {
 		s.Equal("rejection-of-"+tv.UpdateID(), updStatus.Outcome.GetFailure().GetMessage())
 
 		s.EqualHistoryEvents(`
- 13 WorkflowTaskScheduled // WFT events were created even if it was a rejection (because number of events > 10). 
+ 13 WorkflowTaskScheduled // WFT events were created even if it was a rejection (because number of events > 10).
  14 WorkflowTaskStarted
  15 WorkflowTaskCompleted
 `, <-writtenHistoryCh)
 	})
 }
 
+func (s *WorkflowTaskCompletedHandlerSuite) TestForceCreateNewWorkflowTaskOnPausedWorkflow() {
+	s.Run("Returns error when workflow is paused and ForceCreateNewWorkflowTask is true", func() {
+		tv := testvars.New(s.T())
+		tv = tv.WithRunID(tv.Any().RunID())
+		s.mockNamespaceCache.EXPECT().GetNamespaceByID(tv.NamespaceID()).Return(tv.Namespace(), nil).AnyTimes()
+		wfContext, serializedTaskToken := s.createPausedWorkflowWithWFT(tv)
+
+		ms, err := wfContext.LoadMutableState(context.Background(), s.workflowTaskCompletedHandler.shardContext)
+		s.NoError(err)
+		s.True(ms.IsWorkflowExecutionStatusPaused())
+
+		_, err = s.workflowTaskCompletedHandler.Invoke(context.Background(), &historyservice.RespondWorkflowTaskCompletedRequest{
+			NamespaceId: tv.NamespaceID().String(),
+			CompleteRequest: &workflowservice.RespondWorkflowTaskCompletedRequest{
+				TaskToken:                  serializedTaskToken,
+				Identity:                   tv.Any().String(),
+				ForceCreateNewWorkflowTask: true,
+			},
+		})
+		s.Error(err)
+		var failedPrecondition *serviceerror.FailedPrecondition
+		s.ErrorAs(err, &failedPrecondition)
+		s.Contains(err.Error(), "Workflow is paused and force create new workflow task is not allowed")
+	})
+}
+
 func (s *WorkflowTaskCompletedHandlerSuite) TestHandleBufferedQueries() {
 	constructQueryResults := func(ids []string, resultSize int) map[string]*querypb.WorkflowQueryResult {
 		results := make(map[string]*querypb.WorkflowQueryResult)
@@ -617,3 +643,95 @@ func (s *WorkflowTaskCompletedHandlerSuite) createSentUpdate(tv *testvars.TestVa
 
 	return updRequestMsg, upd, serializedTaskToken
 }
+
+func (s *WorkflowTaskCompletedHandlerSuite) createPausedWorkflowWithWFT(tv *testvars.TestVars) (historyi.WorkflowContext, []byte) {
+	ms := workflow.TestLocalMutableState(s.workflowTaskCompletedHandler.shardContext, s.mockEventsCache, tv.Namespace(),
+		tv.WorkflowID(), tv.RunID(), log.NewTestLogger())
+
+	startRequest := &workflowservice.StartWorkflowExecutionRequest{
+		WorkflowId:               tv.WorkflowID(),
+		WorkflowType:             tv.WorkflowType(),
+		TaskQueue:                tv.TaskQueue(),
+		Input:                    tv.Any().Payloads(),
+		WorkflowExecutionTimeout: tv.Any().InfiniteTimeout(),
+		WorkflowRunTimeout:       tv.Any().InfiniteTimeout(),
+		WorkflowTaskTimeout:      tv.Any().InfiniteTimeout(),
+		Identity:                 tv.ClientIdentity(),
+	}
+
+	_, _ = ms.AddWorkflowExecutionStartedEvent(
+		tv.WorkflowExecution(),
+		&historyservice.StartWorkflowExecutionRequest{
+			Attempt:             1,
+			NamespaceId:         tv.NamespaceID().String(),
+			StartRequest:        startRequest,
+			ParentExecutionInfo: nil,
+		},
+	)
+
+	// Complete the first workflow task to transition state to Running.
+	wt, _ := ms.AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_NORMAL)
+	_, _, _ = ms.AddWorkflowTaskStartedEvent(
+		wt.ScheduledEventID,
+		tv.RunID(),
+		tv.TaskQueue(),
+		tv.Any().String(),
+		nil,
+		nil,
+		nil,
+		false,
+	)
+	_, _ = ms.AddWorkflowTaskCompletedEvent(wt, &workflowservice.RespondWorkflowTaskCompletedRequest{
+		Identity: tv.Any().String(),
+	}, historyi.WorkflowTaskCompletionLimits{MaxResetPoints: 10, MaxSearchAttributeValueSize: 2048})
+
+	// Add a speculative WFT before pausing.
+	wt2, _ := ms.AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE)
+	_, _, _ = ms.AddWorkflowTaskStartedEvent(
+		wt2.ScheduledEventID,
+		tv.RunID(),
+		tv.StickyTaskQueue(),
+		tv.Any().String(),
+		nil,
+		nil,
+		nil,
+		false,
+	)
+	taskToken := &tokenspb.Task{
+		Attempt:          1,
+		NamespaceId:      tv.NamespaceID().String(),
+		WorkflowId:       tv.WorkflowID(),
+		RunId:            tv.RunID(),
+		ScheduledEventId: wt2.ScheduledEventID,
+	}
+	serializedTaskToken, err := taskToken.Marshal()
+	s.NoError(err)
+
+	// Pause the workflow
+	_, err = ms.AddWorkflowExecutionPausedEvent("test-identity", "test-reason", tv.Any().String())
+	s.NoError(err)
+	s.True(ms.IsWorkflowExecutionStatusPaused())
+	ms.FlushBufferedEvents()
+
+	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, request *persistence.GetWorkflowExecutionRequest) (*persistence.GetWorkflowExecutionResponse, error) {
+			return &persistence.GetWorkflowExecutionResponse{State: workflow.TestCloneToProto(ms)}, nil
+		}).AnyTimes()
+
+	wfContext, release, err := s.workflowCache.GetOrCreateWorkflowExecution(
+		metrics.AddMetricsContext(context.Background()),
+		s.mockShard,
+		tv.NamespaceID(),
+		tv.WorkflowExecution(),
+		locks.PriorityHigh,
+	)
+	s.NoError(err)
+	s.NotNil(wfContext)
+
+	loadedMS, err := wfContext.LoadMutableState(context.Background(), s.mockShard)
+	s.NoError(err)
+	s.NotNil(loadedMS)
+	release(nil)
+
+	return wfContext, serializedTaskToken
+}

service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go

@@ -497,10 +497,17 @@ func (handler *workflowTaskCompletedHandler) handleCommandScheduleActivity(
 	eagerStartActivity := attr.RequestEagerExecution && handler.config.EnableActivityEagerExecution(namespace) &&
 		(!versioningUsed || attr.UseWorkflowBuildId)
 
+	// if the workflow is paused, we bypass activity task generation and also prevent eager activity execution.
+	bypassActivityTaskGeneration := eagerStartActivity
+	if handler.mutableState.GetExecutionState().Status == enumspb.WORKFLOW_EXECUTION_STATUS_PAUSED {
+		bypassActivityTaskGeneration = true
+		eagerStartActivity = false
+	}
+
 	event, _, err := handler.mutableState.AddActivityTaskScheduledEvent(
 		handler.workflowTaskCompletedID,
 		attr,
-		eagerStartActivity,
+		bypassActivityTaskGeneration,
 	)
 	if err != nil {
 		return nil, nil, handler.failWorkflowTaskOnInvalidArgument(enumspb.WORKFLOW_TASK_FAILED_CAUSE_SCHEDULE_ACTIVITY_DUPLICATE_ID, err)

service/history/api/signalwithstartworkflow/signal_with_start_workflow.go

@@ -311,7 +311,7 @@ func signalWorkflow(
 	}
 
 	// Create a transfer task to schedule a workflow task
-	if !mutableState.HasPendingWorkflowTask() {
+	if !mutableState.HasPendingWorkflowTask() && !mutableState.IsWorkflowExecutionStatusPaused() {
 
 		executionInfo := mutableState.GetExecutionInfo()
 		executionState := mutableState.GetExecutionState()

service/history/api/signalwithstartworkflow/signal_with_start_workflow_test.go

@@ -143,6 +143,7 @@ func (s *signalWithStartWorkflowSuite) TestSignalWorkflow_NewWorkflowTask() {
 		request.GetLinks(),
 	).Return(&historypb.HistoryEvent{}, nil)
 	s.currentMutableState.EXPECT().HasPendingWorkflowTask().Return(false)
+	s.currentMutableState.EXPECT().IsWorkflowExecutionStatusPaused().Return(false)
 	s.currentMutableState.EXPECT().HadOrHasWorkflowTask().Return(true)
 	s.currentMutableState.EXPECT().AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_NORMAL).Return(&historyi.WorkflowTaskInfo{}, nil)
 	s.currentContext.EXPECT().UpdateWorkflowExecutionAsActive(ctx, s.shardContext).Return(nil)
@@ -187,6 +188,40 @@ func (s *signalWithStartWorkflowSuite) TestSignalWorkflow_NoNewWorkflowTask() {
 	s.NoError(err)
 }
 
+// Tests SignalWithStart when the workflow is paused.
+// Asserts that no new workflow task is scheduled.
+func (s *signalWithStartWorkflowSuite) TestSignalWorkflow_WhenPaused() {
+	ctx := context.Background()
+	currentWorkflowLease := api.NewWorkflowLease(
+		s.currentContext,
+		wcache.NoopReleaseFn,
+		s.currentMutableState,
+	)
+	request := s.randomRequest()
+
+	s.currentMutableState.EXPECT().IsWorkflowCloseAttempted().Return(false)
+	s.currentMutableState.EXPECT().IsSignalRequested(request.GetRequestId()).Return(false)
+	s.currentMutableState.EXPECT().AddSignalRequested(request.GetRequestId())
+	s.currentMutableState.EXPECT().AddWorkflowExecutionSignaled(
+		request.GetSignalName(),
+		request.GetSignalInput(),
+		request.GetIdentity(),
+		request.GetHeader(),
+		request.GetLinks(),
+	).Return(&historypb.HistoryEvent{}, nil)
+	s.currentMutableState.EXPECT().HasPendingWorkflowTask().Return(false)
+	s.currentMutableState.EXPECT().IsWorkflowExecutionStatusPaused().Return(true)
+	s.currentContext.EXPECT().UpdateWorkflowExecutionAsActive(ctx, s.shardContext).Return(nil)
+
+	err := signalWorkflow(
+		ctx,
+		s.shardContext,
+		currentWorkflowLease,
+		request,
+	)
+	s.NoError(err)
+}
+
 func (s *signalWithStartWorkflowSuite) randomRequest() *workflowservice.SignalWithStartWorkflowExecutionRequest {
 	var request workflowservice.SignalWithStartWorkflowExecutionRequest
 	_ = fakedata.FakeStruct(&request)

service/history/api/update_workflow_util.go

@@ -78,8 +78,8 @@ func UpdateWorkflowWithNew(
 
 	mutableState := workflowLease.GetMutableState()
 	if postActions.CreateWorkflowTask {
-		// Create a transfer task to schedule a workflow task
-		if !mutableState.HasPendingWorkflowTask() {
+		// Create a transfer task to schedule a workflow task only if the workflow is not paused and there is no pending workflow task.
+		if !mutableState.HasPendingWorkflowTask() && !mutableState.IsWorkflowExecutionStatusPaused() {
 			if _, err := mutableState.AddWorkflowTaskScheduledEvent(
 				false,
 				enumsspb.WORKFLOW_TASK_TYPE_NORMAL,

service/history/api/updateworkflow/api.go

@@ -124,6 +124,11 @@ func (u *Updater) ApplyRequest(
 		return nil, consts.ErrWorkflowCompleted
 	}
 
+	// We don't accept the request to update the workflow if the workflow is paused.
+	if ms.IsWorkflowExecutionStatusPaused() {
+		return nil, serviceerror.NewFailedPrecondition("Workflow is paused. Cannot update the workflow.")
+	}
+
 	if ms.GetExecutionInfo().WorkflowTaskAttempt >= failUpdateWorkflowTaskAttemptCount {
 		// If workflow task is constantly failing, the update to that workflow will also fail.
 		// Additionally, workflow update can't "fix" workflow state because updates (delivered with messages)

service/history/api/updateworkflow/api_test.go

@@ -0,0 +1,56 @@
+package updateworkflow
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/api/serviceerror"
+	updatepb "go.temporal.io/api/update/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/common/definition"
+	historyi "go.temporal.io/server/service/history/interfaces"
+	"go.temporal.io/server/service/history/workflow/update"
+	"go.uber.org/mock/gomock"
+)
+
+func TestApplyRequest_RejectsUpdateOnPausedWorkflow(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+
+	ms := historyi.NewMockMutableState(ctrl)
+	ms.EXPECT().GetWorkflowKey().Return(definition.NewWorkflowKey("ns-id", "wf-id", "run-id"))
+	ms.EXPECT().IsWorkflowExecutionRunning().Return(true)
+	ms.EXPECT().IsWorkflowExecutionStatusPaused().Return(true)
+	// Required by update.NewRegistry
+	ms.EXPECT().GetCurrentVersion().Return(int64(1))
+	ms.EXPECT().VisitUpdates(gomock.Any())
+
+	updateReg := update.NewRegistry(ms)
+
+	updater := &Updater{
+		req: createUpdateRequest("test-update-id"),
+	}
+
+	action, err := updater.ApplyRequest(context.Background(), updateReg, ms)
+
+	require.Nil(t, action)
+	require.Error(t, err)
+	var failedPrecondition *serviceerror.FailedPrecondition
+	require.ErrorAs(t, err, &failedPrecondition)
+	require.Contains(t, err.Error(), "Workflow is paused")
+}
+
+func createUpdateRequest(updateID string) *historyservice.UpdateWorkflowExecutionRequest {
+	return &historyservice.UpdateWorkflowExecutionRequest{
+		Request: &workflowservice.UpdateWorkflowExecutionRequest{
+			Request: &updatepb.Request{
+				Meta: &updatepb.Meta{
+					UpdateId: updateID,
+				},
+			},
+		},
+	}
+}