Mutable state changes for unpausing a single workflow. (#8674)

## What changed?
- Recording unpause event
- Regenerating workflow and activity tasks
- Made EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED a buffered event (since
pause event is buffered)
- Added unit tests (Functional tests are added along with the API
changes)

## Why?
This is needed to implement pause/unpause operation APIs.

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



<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Introduce workflow "unpaused" history event end-to-end, buffer
pause/unpause consistently, and reschedule activities on unpause with
tests.
> 
> - **History Events**:
> - Add `CreateWorkflowExecutionUnpausedEvent` and
`HistoryBuilder.AddWorkflowExecutionUnpausedEvent`.
> - Update event buffering: buffer both `WORKFLOW_EXECUTION_PAUSED` and
`WORKFLOW_EXECUTION_UNPAUSED` in `historybuilder/event_store.go`.
> - **Mutable State**:
> - Add `AddWorkflowExecutionUnpausedEvent` and
`ApplyWorkflowExecutionUnpausedEvent` to transition to `RUNNING`, clear
`PauseInfo`, bump activity stamps, and (re)generate activity tasks as
needed.
>   - Extend `interfaces.MutableState` + mocks with unpause methods.
>   - Rebuilder applies `WORKFLOW_EXECUTION_UNPAUSED` events.
> - **Logging/Tags**:
>   - Add `tag.WorkflowActionWorkflowUnpaused`.
> - **Tests**:
> - Add unit tests for pause/unpause flow, stamps, and buffering
behavior; adjust existing buffering test expectations.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
f139027. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

Co-authored-by: Sean Kane <[email protected]>

Author: gow

common/log/tag/values.go

@@ -11,6 +11,7 @@ var (
 	WorkflowActionWorkflowTerminated    = workflowAction("add-workflow-terminated-event")
 	WorkflowActionWorkflowContinueAsNew = workflowAction("add-workflow-continue-as-new-event")
 	WorkflowActionWorkflowPaused        = workflowAction("add-workflow-paused-event")
+	WorkflowActionWorkflowUnpaused      = workflowAction("add-workflow-unpaused-event")
 
 	// workflow cancellation / sign / update-options
 	WorkflowActionWorkflowCancelRequested        = workflowAction("add-workflow-cancel-requested-event")

service/history/historybuilder/event_factory.go

@@ -1024,6 +1024,22 @@ func (b *EventFactory) CreateWorkflowExecutionPausedEvent(
 	return event
 }
 
+func (b *EventFactory) CreateWorkflowExecutionUnpausedEvent(
+	identity string,
+	reason string,
+	requestID string,
+) *historypb.HistoryEvent {
+	event := b.createHistoryEvent(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED, b.timeSource.Now())
+	event.Attributes = &historypb.HistoryEvent_WorkflowExecutionUnpausedEventAttributes{
+		WorkflowExecutionUnpausedEventAttributes: &historypb.WorkflowExecutionUnpausedEventAttributes{
+			Identity:  identity,
+			Reason:    reason,
+			RequestId: requestID,
+		},
+	}
+	return event
+}
+
 func (b *EventFactory) createHistoryEvent(
 	eventType enumspb.EventType,
 	time time.Time,

service/history/historybuilder/event_store.go

@@ -306,11 +306,10 @@ func (b *EventStore) bufferEvent(
 		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_COMPLETED:
 		return false
 
-	// Unpaused event should not be buffered since it transitions the workflow from paused to running status.
-	case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED:
-		return false
 	// A paused workflow event *should be* allowed to be buffered since we want to accept any inflight workflow task completion.
-	case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED:
+	// Since we buffer the paused event, we need to buffer unpaused event as well so that they don't go out of order.
+	case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_PAUSED,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED:
 		return true
 
 	default:

service/history/historybuilder/history_builder.go

@@ -288,6 +288,18 @@ func (b *HistoryBuilder) AddWorkflowExecutionPausedEvent(
 	return event
 }
 
+func (b *HistoryBuilder) AddWorkflowExecutionUnpausedEvent(
+	identity string,
+	reason string,
+	requestID string,
+) *historypb.HistoryEvent {
+	event := b.CreateWorkflowExecutionUnpausedEvent(identity, reason, requestID)
+	// Mark the event as 'worker may ignore' so that older SDKs can safely ignore it.
+	event.WorkerMayIgnore = true
+	event, _ = b.add(event)
+	return event
+}
+
 func (b *HistoryBuilder) AddActivityTaskScheduledEvent(
 	workflowTaskCompletedEventID int64,
 	command *commandpb.ScheduleActivityTaskCommandAttributes,

service/history/historybuilder/history_builder_test.go

@@ -2222,7 +2222,6 @@ func (s *historyBuilderSuite) TestBufferEvent() {
 		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED:       true,
 		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW: true,
 		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED:         true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED:         true,
 	}
 
 	// workflow task events will be assign event ID immediately

service/history/interfaces/mutable_state.go

@@ -215,6 +215,8 @@ type (
 		IsWorkflowExecutionRunning() bool
 		AddWorkflowExecutionPausedEvent(identity string, reason string, requestID string) (*historypb.HistoryEvent, error)
 		ApplyWorkflowExecutionPausedEvent(event *historypb.HistoryEvent) error
+		AddWorkflowExecutionUnpausedEvent(identity string, reason string, requestID string) (*historypb.HistoryEvent, error)
+		ApplyWorkflowExecutionUnpausedEvent(event *historypb.HistoryEvent) error
 		IsResourceDuplicated(resourceDedupKey definition.DeduplicationID) bool
 		IsWorkflowPendingOnWorkflowTaskBackoff() bool
 		UpdateDuplicatedResource(resourceDedupKey definition.DeduplicationID)

service/history/interfaces/mutable_state_mock.go

@@ -2866,6 +2866,66 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionPausedEvent(event *historypb.H
 	return nil
 }
 
+func (ms *MutableStateImpl) AddWorkflowExecutionUnpausedEvent(
+	identity string,
+	reason string,
+	requestID string,
+) (*historypb.HistoryEvent, error) {
+	opTag := tag.WorkflowActionWorkflowUnpaused
+	if err := ms.checkMutability(opTag); err != nil {
+		return nil, err
+	}
+	event := ms.hBuilder.AddWorkflowExecutionUnpausedEvent(identity, reason, requestID)
+	if err := ms.ApplyWorkflowExecutionUnpausedEvent(event); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+// ApplyWorkflowExecutionUnpausedEvent applies the unpaused event to the mutable state. It updates the workflow execution status to running and clears the pause info.
+func (ms *MutableStateImpl) ApplyWorkflowExecutionUnpausedEvent(event *historypb.HistoryEvent) error {
+	// Update workflow status.
+	if _, err := ms.UpdateWorkflowStateStatus(ms.executionState.GetState(), enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING); err != nil {
+		return err
+	}
+
+	// save pauseInfoSize before clearing so that we can adjust approximate size later before returning success
+	pauseInfoSize := 0
+	if ms.executionInfo.PauseInfo != nil {
+		pauseInfoSize = ms.GetExecutionInfo().GetPauseInfo().Size()
+		// Clear pause info in mutable state.
+		ms.executionInfo.PauseInfo = nil
+	}
+
+	// Reschedule any pending activities
+	// Note: workflow task is scheduled in the unpause API. So no need to schedule it here.
+	for _, ai := range ms.GetPendingActivityInfos() {
+		// Bump activity stamp to force replication so that the passive cluster can recreate the activity task.
+		if err := ms.UpdateActivity(ai.ScheduledEventId, func(activityInfo *persistencespb.ActivityInfo, _ historyi.MutableState) error {
+			activityInfo.Stamp = activityInfo.Stamp + 1
+			return nil
+		}); err != nil {
+			return err
+		}
+
+		// Check activity scheduled time and generate activity retry task if scheduled time is in the future.
+		if ai.GetScheduledTime().AsTime().After(ms.timeSource.Now().UTC()) {
+			if err := ms.taskGenerator.GenerateActivityRetryTasks(ai); err != nil {
+				return err
+			}
+		} else {
+			// Generate activity task to resend the activity to matching immediately.
+			if err := ms.taskGenerator.GenerateActivityTasks(ai.ScheduledEventId); err != nil {
+				return err
+			}
+		}
+	}
+
+	// Update approximate size of the mutable state.
+	ms.approximateSize -= pauseInfoSize
+	return nil
+}
+
 func (ms *MutableStateImpl) addCompletionCallbacks(
 	event *historypb.HistoryEvent,
 	requestID string,

service/history/workflow/mutable_state_impl.go

@@ -1754,6 +1754,85 @@ func (s *mutableStateSuite) TestAddWorkflowExecutionPausedEvent() {
 	s.True(pausedEvent.GetWorkerMayIgnore())
 }
 
+func (s *mutableStateSuite) TestAddWorkflowExecutionUnpausedEvent() {
+	s.SetupSubTest()
+	s.mockEventsCache.EXPECT().PutEvent(gomock.Any(), gomock.Any()).AnyTimes()
+
+	tq := &taskqueuepb.TaskQueue{Name: "tq"}
+	s.createVersionedMutableStateWithCompletedWFT(tq)
+
+	// Complete another WFT to obtain a valid completed event id for scheduling an activity.
+	wft, err := s.mutableState.AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_NORMAL)
+	s.NoError(err)
+	_, wft, err = s.mutableState.AddWorkflowTaskStartedEvent(
+		wft.ScheduledEventID,
+		"",
+		tq,
+		"",
+		worker_versioning.StampFromBuildId("b1"),
+		nil,
+		nil,
+		false,
+	)
+	s.NoError(err)
+	completedEvent, err := s.mutableState.AddWorkflowTaskCompletedEvent(
+		wft,
+		&workflowservice.RespondWorkflowTaskCompletedRequest{},
+		workflowTaskCompletionLimits,
+	)
+	s.NoError(err)
+
+	// Schedule an activity (pending) using the completed WFT event id.
+	_, activityInfo, err := s.mutableState.AddActivityTaskScheduledEvent(
+		completedEvent.GetEventId(),
+		&commandpb.ScheduleActivityTaskCommandAttributes{
+			ActivityId:   "act-1",
+			ActivityType: &commonpb.ActivityType{Name: "activity-type"},
+			TaskQueue:    tq,
+		},
+		false,
+	)
+	s.NoError(err)
+	// Create a pending workflow task.
+	pendingWFT, err := s.mutableState.AddWorkflowTaskScheduledEvent(false, enumsspb.WORKFLOW_TASK_TYPE_NORMAL)
+	s.NoError(err)
+
+	// Pause first to simulate paused workflow state.
+	_, err = s.mutableState.AddWorkflowExecutionPausedEvent("tester", "reason", uuid.New())
+	s.NoError(err)
+
+	// Capture stamps after pause.
+	pausedActivityInfo, ok := s.mutableState.GetActivityInfo(activityInfo.ScheduledEventId)
+	s.True(ok)
+	pausedActivityStamp := pausedActivityInfo.Stamp
+	pausedWFT := s.mutableState.GetPendingWorkflowTask()
+	s.NotNil(pausedWFT)
+	pausedWFTStamp := pausedWFT.Stamp
+
+	// Unpause and verify.
+	unpausedEvent, err := s.mutableState.AddWorkflowExecutionUnpausedEvent("tester", "reason", uuid.New())
+	s.NoError(err)
+
+	// PauseInfo should be cleared and status should be RUNNING.
+	s.Nil(s.mutableState.executionInfo.PauseInfo)
+	s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, s.mutableState.executionState.Status)
+
+	// Stamps should be incremented again (only for activities) on unpause.
+	updatedActivityInfo, ok := s.mutableState.GetActivityInfo(activityInfo.ScheduledEventId)
+	s.True(ok)
+	s.Greater(updatedActivityInfo.Stamp, pausedActivityStamp)
+
+	currentWFT := s.mutableState.GetPendingWorkflowTask()
+	s.NotNil(currentWFT)
+	s.Equal(currentWFT.Stamp, pausedWFTStamp) // workflow task stamp should not change between pause and unpause.
+
+	// assert the event is marked as 'worker may ignore' so that older SDKs can safely ignore it.
+	s.True(unpausedEvent.GetWorkerMayIgnore())
+
+	// Ensure the pending workflow task we created earlier still exists (no unexpected removal).
+	s.Equal(pendingWFT.ScheduledEventID, currentWFT.ScheduledEventID)
+}
+
 func (s *mutableStateSuite) TestPauseWorkflowExecution_FailStateValidation() {
 	s.SetupSubTest()
 	s.mockEventsCache.EXPECT().PutEvent(gomock.Any(), gomock.Any()).AnyTimes()

service/history/workflow/mutable_state_impl_test.go

@@ -666,6 +666,10 @@ func (b *MutableStateRebuilderImpl) applyEvents(
 			if err := b.mutableState.ApplyWorkflowExecutionPausedEvent(event); err != nil {
 				return nil, err
 			}
+		case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UNPAUSED:
+			if err := b.mutableState.ApplyWorkflowExecutionUnpausedEvent(event); err != nil {
+				return nil, err
+			}
 
 		default:
 			def, ok := b.shard.StateMachineRegistry().EventDefinition(event.GetEventType())