Fix execution last running clock on start (#11578)

## What changed? - Update execution's LastRunningClock when in ID reuse case and create new execution as current. - Two CHASM specific changes: 1. CHASM executions now consume a number from shard clock when updating LastRunningClock to guarantee uniqueness. 2. CHASM engine methods now block until shard is acquired and ready to serve traffic. ## Why? - The start execution flow attempts two creation, first as branchNew and a second one as current. However the mutable state snapshot is prepared at the first creation time and reused for the second attempt. If the previous run is closed after the snapshot is prepared, then the previous run's lastRunningClock will be larger than the new run's lastRunningClock. This will cause standby cluster to treat the new run as the older one and put it in zombie state. ## 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)
2026-08-30 18:41:49 -07:00 · 2026-08-17 17:20:58 -07:00
parent f3ebe4f99f
commit 2c358772d2
7 changed files with 377 additions and 71 deletions
--- a/service/history/api/startworkflow/api.go
+++ b/service/history/api/startworkflow/api.go
@@ -392,13 +392,30 @@ func (s *Starter) createAsCurrent(
 	if _, err := s.createOrUpdateLeaseFn(creationParams.workflowLease, s.shardContext, nil); err != nil {
 		return err
 	}
+
+	// If current workflow is closed after the original creationParams were prepared,
+	// the LastRunningClock in the prepared workflow snapshot will be smaller than
+	// the current workflow's LastRunningClock, causing the new workflow to be marked
+	// as zombie in standby cluster.
+	//
+	// Here we basically refresh the LastRunningClock in the prepared workflow snapshot to avoid this issue.
+	mutableState := creationParams.workflowLease.GetMutableState()
+	updateExecutionInfo, updatedWorkflowEventBatches, err := mutableState.UpdateLastRunningClock(creationParams.workflowEventBatches)
+	if err != nil {
+		return err
+	}
+	// Following assignments are technically not necessary since those pointers point to the same underlying fields that are updated,
+	// but it makes the code more explicit and easier to read.
+	creationParams.workflowSnapshot.ExecutionInfo = updateExecutionInfo
+	creationParams.workflowEventBatches = updatedWorkflowEventBatches
+
 	return creationParams.workflowLease.GetContext().CreateWorkflowExecution(
 		ctx,
 		s.shardContext,
 		persistence.CreateWorkflowModeUpdateCurrent,
 		currentWorkflowConditionFailed.RunID,
 		currentWorkflowConditionFailed.LastWriteVersion,
-		creationParams.workflowLease.GetMutableState(),
+		mutableState,
 		creationParams.workflowSnapshot,
 		creationParams.workflowEventBatches,
 		historyi.TransactionPolicyActive,
--- a/service/history/chasm_engine.go
+++ b/service/history/chasm_engine.go
@@ -189,7 +189,7 @@ func (e *ChasmEngine) startExecution(
 	startFn func(chasm.MutableContext) (chasm.RootComponent, error),
 	options chasm.TransitionOptions,
 ) (result chasm.StartExecutionResult, retErr error) {
-	shardContext, err := e.getShardContext(executionRef)
+	shardContext, err := e.getShardContext(ctx, executionRef)
 	if err != nil {
 		return chasm.StartExecutionResult{}, err
 	}
@@ -285,7 +285,7 @@ func (e *ChasmEngine) updateWithStartExecution(
 	updateFn func(chasm.MutableContext, chasm.Component) error,
 	options chasm.TransitionOptions,
 ) (result chasm.EngineUpdateWithStartExecutionResult, retError error) {
-	shardContext, err := e.getShardContext(executionRef)
+	shardContext, err := e.getShardContext(ctx, executionRef)
 	if err != nil {
 		return chasm.EngineUpdateWithStartExecutionResult{}, err
 	}
@@ -1137,7 +1137,22 @@ func (e *ChasmEngine) handleReusePolicy(
 		)
 	}

-	err := newExecutionParams.executionContext.CreateWorkflowExecution(
+	// If current execution is closed after the newExecutionParams are prepared,
+	// the LastRunningClock in the prepared execution snapshot will be smaller than
+	// the current execution's LastRunningClock, causing the new execution to be marked
+	// as zombie in standby cluster.
+	//
+	// Here we basically refresh the LastRunningClock in the prepared execution snapshot to avoid this issue.
+	updatedExecutionInfo, updatedEvents, err := newExecutionParams.mutableState.UpdateLastRunningClock(newExecutionParams.events)
+	if err != nil {
+		return chasm.StartExecutionResult{}, err
+	}
+	// Following assignments are technically not necessary since those pointers point to the same underlying fields that are updated,
+	// but it makes the code more explicit and easier to read.
+	newExecutionParams.snapshot.ExecutionInfo = updatedExecutionInfo
+	newExecutionParams.events = updatedEvents
+
+	if err := newExecutionParams.executionContext.CreateWorkflowExecution(
 		ctx,
 		shardContext,
 		persistence.CreateWorkflowModeUpdateCurrent,
@@ -1147,8 +1162,7 @@ func (e *ChasmEngine) handleReusePolicy(
 		newExecutionParams.snapshot,
 		newExecutionParams.events,
 		historyi.TransactionPolicyActive,
-	)
-	if err != nil {
+	); err != nil {
 		return chasm.StartExecutionResult{}, err
 	}

@@ -1166,15 +1180,26 @@ func (e *ChasmEngine) handleReusePolicy(
 }

 func (e *ChasmEngine) getShardContext(
+	ctx context.Context,
 	ref chasm.ComponentRef,
 ) (historyi.ShardContext, error) {
-	return e.shardController.GetShardByID(
+	shardContext, err := e.shardController.GetShardByID(
 		common.WorkflowIDToHistoryShard(
 			ref.NamespaceID,
 			ref.BusinessID,
 			e.config.NumberOfShards,
 		),
 	)
+	if err != nil {
+		return nil, err
+	}
+
+	// Block until shard is acquired and ready to serve traffic.
+	_, err = shardContext.GetEngine(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return shardContext, nil
 }

 // getExecutionLease returns shard context and mutable state for the chasm execution, with the lock
@@ -1187,7 +1212,7 @@ func (e *ChasmEngine) getExecutionLease(
 	ctx context.Context,
 	ref chasm.ComponentRef,
 ) (historyi.ShardContext, api.WorkflowLease, error) {
-	shardContext, err := e.getShardContext(ref)
+	shardContext, err := e.getShardContext(ctx, ref)
 	if err != nil {
 		return nil, nil, err
 	}
--- a/service/history/chasm_engine_test.go
+++ b/service/history/chasm_engine_test.go
@@ -143,7 +143,7 @@ func (s *chasmEngineSuite) initAssertions() {
 	s.ProtoAssertions = protorequire.New(s.T())
 }

-func (s *chasmEngineSuite) TestNewExecution_BrandNew() {
+func (s *chasmEngineSuite) TestStartExecution_BrandNew() {
 	tv := testvars.New(s.T())

 	ref := chasm.NewComponentRef[*testComponent](
@@ -161,7 +161,7 @@ func (s *chasmEngineSuite) TestNewExecution_BrandNew() {
 			_ context.Context,
 			request *persistence.CreateWorkflowExecutionRequest,
 		) (*persistence.CreateWorkflowExecutionResponse, error) {
-			s.validateCreateRequest(request, s.archetypeID, newActivityID, "", 0)
+			s.validateCreateRequest(request, s.archetypeID, newActivityID, "", 0, 0)
 			runID = request.NewWorkflowSnapshot.ExecutionState.RunId
 			return tests.CreateWorkflowExecutionResponse, nil
 		},
@@ -188,6 +188,41 @@ func (s *chasmEngineSuite) TestNewExecution_BrandNew() {
 	s.True(result.Created)
 }

+func (s *chasmEngineSuite) TestStartExecution_WaitsForShardEngine() {
+	tv := testvars.New(s.T())
+
+	ref := chasm.NewComponentRef[*testComponent](
+		chasm.ExecutionKey{
+			NamespaceID: string(tests.NamespaceID),
+			BusinessID:  tv.WorkflowID(),
+			RunID:       "",
+		},
+	)
+
+	// Use a mock shard context here which is easier to assert that GetEngine method is called.
+	mockShardContext := historyi.NewMockShardContext(s.controller)
+	mockShardController := shard.NewMockController(s.controller)
+	s.engine.SetShardController(mockShardController)
+
+	expectedErr := serviceerror.NewUnavailable("shard not ready")
+	mockShardController.EXPECT().GetShardByID(gomock.Any()).Return(mockShardContext, nil).Times(1)
+	mockShardContext.EXPECT().GetEngine(gomock.Any()).Return(nil, expectedErr).Times(1)
+
+	startFnCalled := false
+	result, err := s.engine.StartExecution(
+		context.Background(),
+		ref,
+		func(chasm.MutableContext) (chasm.RootComponent, error) {
+			startFnCalled = true
+			return &testComponent{}, nil
+		},
+	)
+
+	s.ErrorIs(err, expectedErr)
+	s.False(startFnCalled)
+	s.False(result.Created)
+}
+
 func (s *chasmEngineSuite) TestStartExecution_SetsContextMetadata() {
 	tv := testvars.New(s.T())

@@ -217,7 +252,7 @@ func (s *chasmEngineSuite) TestStartExecution_SetsContextMetadata() {
 	s.assertTestContextMetadata(requestCtx, newActivityID, "start-request")
 }

-func (s *chasmEngineSuite) TestNewExecution_RequestIDDedup() {
+func (s *chasmEngineSuite) TestStartExecution_RequestIDDedup() {
 	tv := testvars.New(s.T())
 	tv = tv.WithRunID(tv.Any().RunID())

@@ -257,7 +292,7 @@ func (s *chasmEngineSuite) TestNewExecution_RequestIDDedup() {
 	s.False(result.Created)
 }

-func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_AllowDuplicate() {
+func (s *chasmEngineSuite) TestStartExecution_ReusePolicy_AllowDuplicate() {
 	tv := testvars.New(s.T())
 	tv = tv.WithRunID(tv.Any().RunID())

@@ -275,17 +310,37 @@ func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_AllowDuplicate() {
 		enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
 	)

-	var runID string
-	s.mockExecutionManager.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).Return(
-		nil,
-		currentRunConditionFailedErr,
-	).Times(1)
+	var currentExecutionLastRunningClock int64
 	s.mockExecutionManager.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
 		func(
 			_ context.Context,
 			request *persistence.CreateWorkflowExecutionRequest,
 		) (*persistence.CreateWorkflowExecutionResponse, error) {
-			s.validateCreateRequest(request, s.archetypeID, newActivityID, tv.RunID(), currentRunConditionFailedErr.LastWriteVersion)
+			// Test the case where current execution is closed after new execution's mutable state
+			// snapshot is prepared in memory.
+			var err error
+			currentExecutionLastRunningClock, err = s.mockShard.GenerateTaskID()
+			if err != nil {
+				return nil, err
+			}
+			return nil, currentRunConditionFailedErr
+		},
+	).Times(1)
+
+	var runID string
+	s.mockExecutionManager.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(
+			_ context.Context,
+			request *persistence.CreateWorkflowExecutionRequest,
+		) (*persistence.CreateWorkflowExecutionResponse, error) {
+			s.validateCreateRequest(
+				request,
+				s.archetypeID,
+				newActivityID,
+				tv.RunID(),
+				currentRunConditionFailedErr.LastWriteVersion,
+				currentExecutionLastRunningClock,
+			)
 			runID = request.NewWorkflowSnapshot.ExecutionState.RunId
 			return tests.CreateWorkflowExecutionResponse, nil
 		},
@@ -313,7 +368,7 @@ func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_AllowDuplicate() {
 	s.True(result.Created)
 }

-func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_FailedOnly_Success() {
+func (s *chasmEngineSuite) TestStartExecution_ReusePolicy_FailedOnly_Success() {
 	tv := testvars.New(s.T())
 	tv = tv.WithRunID(tv.Any().RunID())

@@ -331,17 +386,37 @@ func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_FailedOnly_Success() {
 		enumspb.WORKFLOW_EXECUTION_STATUS_FAILED,
 	)

-	var runID string
-	s.mockExecutionManager.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).Return(
-		nil,
-		currentRunConditionFailedErr,
-	).Times(1)
+	var currentExecutionLastRunningClock int64
 	s.mockExecutionManager.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
 		func(
 			_ context.Context,
 			request *persistence.CreateWorkflowExecutionRequest,
 		) (*persistence.CreateWorkflowExecutionResponse, error) {
-			s.validateCreateRequest(request, s.archetypeID, newActivityID, tv.RunID(), currentRunConditionFailedErr.LastWriteVersion)
+			// Test the case where current execution is closed after new execution's mutable state
+			// snapshot is prepared in memory.
+			var err error
+			currentExecutionLastRunningClock, err = s.mockShard.GenerateTaskID()
+			if err != nil {
+				return nil, err
+			}
+			return nil, currentRunConditionFailedErr
+		},
+	).Times(1)
+
+	var runID string
+	s.mockExecutionManager.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(
+			_ context.Context,
+			request *persistence.CreateWorkflowExecutionRequest,
+		) (*persistence.CreateWorkflowExecutionResponse, error) {
+			s.validateCreateRequest(
+				request,
+				s.archetypeID,
+				newActivityID,
+				tv.RunID(),
+				currentRunConditionFailedErr.LastWriteVersion,
+				currentExecutionLastRunningClock,
+			)
 			runID = request.NewWorkflowSnapshot.ExecutionState.RunId
 			return tests.CreateWorkflowExecutionResponse, nil
 		},
@@ -369,7 +444,7 @@ func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_FailedOnly_Success() {
 	s.True(result.Created)
 }

-func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_FailedOnly_Fail() {
+func (s *chasmEngineSuite) TestStartExecution_ReusePolicy_FailedOnly_Fail() {
 	tv := testvars.New(s.T())
 	tv = tv.WithRunID(tv.Any().RunID())

@@ -404,7 +479,7 @@ func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_FailedOnly_Fail() {
 	s.False(result.Created)
 }

-func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_RejectDuplicate() {
+func (s *chasmEngineSuite) TestStartExecution_ReusePolicy_RejectDuplicate() {
 	tv := testvars.New(s.T())
 	tv = tv.WithRunID(tv.Any().RunID())

@@ -439,7 +514,7 @@ func (s *chasmEngineSuite) TestNewExecution_ReusePolicy_RejectDuplicate() {
 	s.False(result.Created)
 }

-func (s *chasmEngineSuite) TestNewExecution_ConflictPolicy_UseExisting() {
+func (s *chasmEngineSuite) TestStartExecution_ConflictPolicy_UseExisting() {
 	tv := testvars.New(s.T())
 	tv = tv.WithRunID(tv.Any().RunID())

@@ -484,7 +559,7 @@ func (s *chasmEngineSuite) TestNewExecution_ConflictPolicy_UseExisting() {
 	s.False(result.Created)
 }

-func (s *chasmEngineSuite) TestNewExecution_ConflictPolicy_TerminateExisting() {
+func (s *chasmEngineSuite) TestStartExecution_ConflictPolicy_TerminateExisting() {
 	tv := testvars.New(s.T())
 	tv = tv.WithRunID(tv.Any().RunID())

@@ -591,15 +666,17 @@ func (s *chasmEngineSuite) validateCreateRequest(
 	expectedActivityID string,
 	expectedPreviousRunID string,
 	expectedPreviousLastWriteVersion int64,
+	previousLastRunningClock int64,
 ) {
 	s.Equal(expectedArchetypeID, request.ArchetypeID)

-	if expectedPreviousRunID == "" && expectedPreviousLastWriteVersion == 0 {
+	if expectedPreviousRunID == "" && expectedPreviousLastWriteVersion == 0 && previousLastRunningClock == 0 {
 		s.Equal(persistence.CreateWorkflowModeBrandNew, request.Mode)
 	} else {
 		s.Equal(persistence.CreateWorkflowModeUpdateCurrent, request.Mode)
 		s.Equal(expectedPreviousRunID, request.PreviousRunID)
 		s.Equal(expectedPreviousLastWriteVersion, request.PreviousLastWriteVersion)
+		s.Less(previousLastRunningClock, request.NewWorkflowSnapshot.ExecutionInfo.LastRunningClock)
 	}

 	s.Len(request.NewWorkflowSnapshot.ChasmNodes, 1)
@@ -1769,6 +1846,8 @@ func (s *chasmEngineSuite) TestUpdateWithStartExecution_NotFound() {
 			_ context.Context,
 			request *persistence.CreateWorkflowExecutionRequest,
 		) (*persistence.CreateWorkflowExecutionResponse, error) {
+			s.Equal(persistence.CreateWorkflowModeBrandNew, request.Mode)
+
 			s.NotNil(request.NewWorkflowSnapshot)
 			createdRunID = request.NewWorkflowSnapshot.ExecutionState.RunId
 			s.NotEmpty(createdRunID)
@@ -1879,22 +1958,36 @@ func (s *chasmEngineSuite) TestUpdateWithStartExecution_ExistingClosed() {
 		}, nil).Times(2)

 	// Mock GetWorkflowExecution for the closed execution.
-	s.mockExecutionManager.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).
-		Return(&persistence.GetWorkflowExecutionResponse{
-			State: s.buildPersistenceMutableState(
-				chasm.ExecutionKey{
-					NamespaceID: executionKey.NamespaceID,
-					BusinessID:  executionKey.BusinessID,
-					RunID:       tv.RunID(),
-				},
-				&persistencespb.ActivityInfo{
-					ActivityId: tv.ActivityID(),
-				},
-				enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
-				enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
-				nil,
-			),
-		}, nil).Times(1)
+	var currentExecutionLastRunningClock int64
+	s.mockExecutionManager.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(
+			_ context.Context,
+			request *persistence.GetWorkflowExecutionRequest,
+		) (*persistence.GetWorkflowExecutionResponse, error) {
+			// Test the case where current execution is closed after new execution's mutable state
+			// snapshot is prepared in memory.
+			var err error
+			currentExecutionLastRunningClock, err = s.mockShard.GenerateTaskID()
+			if err != nil {
+				return nil, err
+			}
+			return &persistence.GetWorkflowExecutionResponse{
+				State: s.buildPersistenceMutableState(
+					chasm.ExecutionKey{
+						NamespaceID: executionKey.NamespaceID,
+						BusinessID:  executionKey.BusinessID,
+						RunID:       tv.RunID(),
+					},
+					&persistencespb.ActivityInfo{
+						ActivityId: tv.ActivityID(),
+					},
+					enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED,
+					enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED,
+					nil,
+				),
+			}, nil
+		},
+	).Times(1)

 	// Mock CreateWorkflowExecution for new execution with UpdateCurrent mode
 	// (since we already have a lease on the closed execution).
@@ -1910,6 +2003,7 @@ func (s *chasmEngineSuite) TestUpdateWithStartExecution_ExistingClosed() {
 			createdRunID = request.NewWorkflowSnapshot.ExecutionState.RunId
 			s.NotEmpty(createdRunID)
 			s.NotEqual(tv.RunID(), createdRunID) // New run should have different RunID.
+			s.Less(currentExecutionLastRunningClock, request.NewWorkflowSnapshot.ExecutionInfo.LastRunningClock)

 			return tests.CreateWorkflowExecutionResponse, nil
 		},
--- a/service/history/history_engine2_test.go
+++ b/service/history/history_engine2_test.go
@@ -1819,10 +1819,34 @@ func (s *engine2Suite) TestStartWorkflowExecution_Dedup() {
 		s.Run("with success", func() {

 			s.Run("and id reuse policy is ALLOW_DUPLICATE", func() {
+				var currentExecutionLastRunningClock int64
 				s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), brandNewExecutionRequest).
-					Return(nil, makeCurrentWorkflowConditionFailedError(prevRequestID))
+					DoAndReturn(
+						func(
+							_ context.Context,
+							request *persistence.CreateWorkflowExecutionRequest,
+						) (*persistence.CreateWorkflowExecutionResponse, error) {
+							// Test the case where current execution is closed after new execution's mutable state
+							// snapshot is prepared in memory.
+							var err error
+							currentExecutionLastRunningClock, err = s.mockShard.GenerateTaskID()
+							if err != nil {
+								return nil, err
+							}
+							return nil, makeCurrentWorkflowConditionFailedError(prevRequestID)
+						},
+					)
+
 				s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), updateExecutionRequest).
-					Return(tests.CreateWorkflowExecutionResponse, nil)
+					DoAndReturn(
+						func(
+							_ context.Context,
+							request *persistence.CreateWorkflowExecutionRequest,
+						) (*persistence.CreateWorkflowExecutionResponse, error) {
+							s.assertWorkflowLastRunningClockUpdated(currentExecutionLastRunningClock, request)
+							return tests.CreateWorkflowExecutionResponse, nil
+						},
+					)

 				resp, err := s.historyEngine.StartWorkflowExecution(
 					metrics.AddMetricsContext(context.Background()),
@@ -1834,10 +1858,34 @@ func (s *engine2Suite) TestStartWorkflowExecution_Dedup() {
 			})

 			s.Run("and id reuse policy is TERMINATE_IF_RUNNING", func() {
+				var currentExecutionLastRunningClock int64
 				s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), brandNewExecutionRequest).
-					Return(nil, makeCurrentWorkflowConditionFailedError(prevRequestID))
+					DoAndReturn(
+						func(
+							_ context.Context,
+							request *persistence.CreateWorkflowExecutionRequest,
+						) (*persistence.CreateWorkflowExecutionResponse, error) {
+							// Test the case where current execution is closed after new execution's mutable state
+							// snapshot is prepared in memory.
+							var err error
+							currentExecutionLastRunningClock, err = s.mockShard.GenerateTaskID()
+							if err != nil {
+								return nil, err
+							}
+							return nil, makeCurrentWorkflowConditionFailedError(prevRequestID)
+						},
+					)
+
 				s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), updateExecutionRequest).
-					Return(tests.CreateWorkflowExecutionResponse, nil)
+					DoAndReturn(
+						func(
+							_ context.Context,
+							request *persistence.CreateWorkflowExecutionRequest,
+						) (*persistence.CreateWorkflowExecutionResponse, error) {
+							s.assertWorkflowLastRunningClockUpdated(currentExecutionLastRunningClock, request)
+							return tests.CreateWorkflowExecutionResponse, nil
+						},
+					)

 				resp, err := s.historyEngine.StartWorkflowExecution(
 					metrics.AddMetricsContext(context.Background()),
@@ -1920,10 +1968,34 @@ func (s *engine2Suite) TestStartWorkflowExecution_Dedup() {
 					})

 					s.Run("and id reuse policy ALLOW_DUPLICATE_FAILED_ONLY", func() {
+						var currentExecutionLastRunningClock int64
 						s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), brandNewExecutionRequest).
-							Return(nil, makeCurrentWorkflowConditionFailedError(prevRequestID))
+							DoAndReturn(
+								func(
+									_ context.Context,
+									request *persistence.CreateWorkflowExecutionRequest,
+								) (*persistence.CreateWorkflowExecutionResponse, error) {
+									// Test the case where current execution is closed after new execution's mutable state
+									// snapshot is prepared in memory.
+									var err error
+									currentExecutionLastRunningClock, err = s.mockShard.GenerateTaskID()
+									if err != nil {
+										return nil, err
+									}
+									return nil, makeCurrentWorkflowConditionFailedError(prevRequestID)
+								},
+							)
+
 						s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), updateExecutionRequest).
-							Return(tests.CreateWorkflowExecutionResponse, nil)
+							DoAndReturn(
+								func(
+									_ context.Context,
+									request *persistence.CreateWorkflowExecutionRequest,
+								) (*persistence.CreateWorkflowExecutionResponse, error) {
+									s.assertWorkflowLastRunningClockUpdated(currentExecutionLastRunningClock, request)
+									return tests.CreateWorkflowExecutionResponse, nil
+								},
+							)

 						resp, err := s.historyEngine.StartWorkflowExecution(
 							metrics.AddMetricsContext(context.Background()),
@@ -2042,7 +2114,15 @@ func (s *engine2Suite) TestSignalWithStartWorkflowExecution_WorkflowNotExist() {
 	notExistErr := serviceerror.NewNotFound("Workflow not exist")

 	s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), gomock.Any()).Return(nil, notExistErr)
-	s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).Return(tests.CreateWorkflowExecutionResponse, nil)
+	s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(
+			_ context.Context,
+			request *persistence.CreateWorkflowExecutionRequest,
+		) (*persistence.CreateWorkflowExecutionResponse, error) {
+			s.Equal(persistence.CreateWorkflowModeBrandNew, request.Mode)
+			return tests.CreateWorkflowExecutionResponse, nil
+		},
+	)

 	resp, err := s.historyEngine.SignalWithStartWorkflowExecution(metrics.AddMetricsContext(context.Background()), sRequest)
 	s.Nil(err)
@@ -2107,9 +2187,32 @@ func (s *engine2Suite) TestSignalWithStartWorkflowExecution_WorkflowNotRunning()
 	gwmsResponse := &persistence.GetWorkflowExecutionResponse{State: wfMs}
 	gceResponse := &persistence.GetCurrentExecutionResponse{RunID: runID}

+	var currentExecutionLastRunningClock int64
 	s.mockExecutionMgr.EXPECT().GetCurrentExecution(gomock.Any(), gomock.Any()).Return(gceResponse, nil).AnyTimes()
-	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(gwmsResponse, nil)
-	s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).Return(tests.CreateWorkflowExecutionResponse, nil)
+	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(
+			_ context.Context,
+			request *persistence.GetWorkflowExecutionRequest,
+		) (*persistence.GetWorkflowExecutionResponse, error) {
+			// Test the case where current execution is closed after new execution's mutable state is created in memory.
+			var err error
+			currentExecutionLastRunningClock, err = s.mockShard.GenerateTaskID()
+			if err != nil {
+				return nil, err
+			}
+			return gwmsResponse, nil
+		},
+	)
+	s.mockExecutionMgr.EXPECT().CreateWorkflowExecution(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(
+			_ context.Context,
+			request *persistence.CreateWorkflowExecutionRequest,
+		) (*persistence.CreateWorkflowExecutionResponse, error) {
+			s.Equal(persistence.CreateWorkflowModeUpdateCurrent, request.Mode)
+			s.assertWorkflowLastRunningClockUpdated(currentExecutionLastRunningClock, request)
+			return tests.CreateWorkflowExecutionResponse, nil
+		},
+	)

 	resp, err := s.historyEngine.SignalWithStartWorkflowExecution(metrics.AddMetricsContext(context.Background()), sRequest)
 	s.Nil(err)
@@ -3030,14 +3133,22 @@ func (s *engine2Suite) getMutableState(namespaceID namespace.ID, we *commonpb.Wo
 	return weContext.(*workflow.ContextImpl).MutableState
 }

-type createWorkflowExecutionRequestMatcher struct {
-	f func(request *persistence.CreateWorkflowExecutionRequest) bool
+func (s *engine2Suite) assertWorkflowLastRunningClockUpdated(
+	currentExecutionLastRunningClock int64,
+	request *persistence.CreateWorkflowExecutionRequest,
+) {
+	updatedLastRunningClock := request.NewWorkflowSnapshot.ExecutionInfo.LastRunningClock
+	s.Less(currentExecutionLastRunningClock, updatedLastRunningClock)
+	s.NotEmpty(request.NewWorkflowEvents)
+
+	lastBatch := request.NewWorkflowEvents[len(request.NewWorkflowEvents)-1]
+	s.NotEmpty(lastBatch.Events)
+	lastEvent := lastBatch.Events[len(lastBatch.Events)-1]
+	s.Equal(updatedLastRunningClock, lastEvent.GetTaskId())
 }

-func newCreateWorkflowExecutionRequestMatcher(f func(request *persistence.CreateWorkflowExecutionRequest) bool) gomock.Matcher {
-	return &createWorkflowExecutionRequestMatcher{
-		f: f,
-	}
+type createWorkflowExecutionRequestMatcher struct {
+	f func(request *persistence.CreateWorkflowExecutionRequest) bool
 }

 func (m *createWorkflowExecutionRequestMatcher) Matches(x any) bool {
--- a/service/history/interfaces/mutable_state.go
+++ b/service/history/interfaces/mutable_state.go
@@ -350,6 +350,7 @@ type (
 		// CloseTransactionAsSnapshot closes the mutable state transaction (different from DB transaction) and prepares the current snapshot of the state to be persisted and bumps the DBRecordVersion.
 		// You should ideally not make any changes to the mutable state after this call.
 		CloseTransactionAsSnapshot(ctx context.Context, transactionPolicy TransactionPolicy) (*persistence.WorkflowSnapshot, []*persistence.WorkflowEvents, error)
+		UpdateLastRunningClock([]*persistence.WorkflowEvents) (*persistencespb.WorkflowExecutionInfo, []*persistence.WorkflowEvents, error)
 		GenerateMigrationTasks(targetClusters []string) ([]tasks.Task, int64, error)

 		// ContinueAsNewMinBackoff calculate minimal backoff for next ContinueAsNew run.
--- a/service/history/interfaces/mutable_state_mock.go
+++ b/service/history/interfaces/mutable_state_mock.go
@@ -3837,6 +3837,22 @@ func (mr *MockMutableStateMockRecorder) UpdateDuplicatedResource(resourceDedupKe
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateDuplicatedResource", reflect.TypeOf((*MockMutableState)(nil).UpdateDuplicatedResource), resourceDedupKey)
 }

+// UpdateLastRunningClock mocks base method.
+func (m *MockMutableState) UpdateLastRunningClock(arg0 []*persistence0.WorkflowEvents) (*persistence.WorkflowExecutionInfo, []*persistence0.WorkflowEvents, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "UpdateLastRunningClock", arg0)
+	ret0, _ := ret[0].(*persistence.WorkflowExecutionInfo)
+	ret1, _ := ret[1].([]*persistence0.WorkflowEvents)
+	ret2, _ := ret[2].(error)
+	return ret0, ret1, ret2
+}
+
+// UpdateLastRunningClock indicates an expected call of UpdateLastRunningClock.
+func (mr *MockMutableStateMockRecorder) UpdateLastRunningClock(arg0 any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UpdateLastRunningClock", reflect.TypeOf((*MockMutableState)(nil).UpdateLastRunningClock), arg0)
+}
+
 // UpdateResetRunID mocks base method.
 func (m *MockMutableState) UpdateResetRunID(runID string) {
 	m.ctrl.T.Helper()
--- a/service/history/workflow/mutable_state_impl.go
+++ b/service/history/workflow/mutable_state_impl.go
@@ -7853,7 +7853,9 @@ func (ms *MutableStateImpl) closeTransaction(
 			return closeTransactionResult{}, err
 		}
 		ms.closeTransactionHandleUnknownVersionedTransition()
-		ms.closeTransactionUpdateLastRunningClock(transactionPolicy, workflowEventsSeq)
+		if err := ms.closeTransactionUpdateLastRunningClock(transactionPolicy, workflowEventsSeq); err != nil {
+			return closeTransactionResult{}, err
+		}
 	}

 	// todo@TimeSkipping, we can move update versioned transition to inside closeTransactionHandleWorkflowTimeSkipping
@@ -8123,28 +8125,68 @@ func (ms *MutableStateImpl) closeTransactionHandleUnknownVersionedTransition() {
 func (ms *MutableStateImpl) closeTransactionUpdateLastRunningClock(
 	transactionPolicy historyi.TransactionPolicy,
 	workflowEventsSeq []*persistence.WorkflowEvents,
-) {
+) error {
 	if transactionPolicy != historyi.TransactionPolicyActive {
-		return
+		return nil
 	}

 	// Events can only be generated while mutable state is running,
 	// so we can update LastRunningClock blindly.
+	//
+	// NOT reusing the UpdateLastRunningClock() logic here since event taskIDs are already assigned in EventStore.
+	// TODO: Move assignTaskIDs logic in EventStore to here.
 	if len(workflowEventsSeq) > 0 {
 		lastEvents := workflowEventsSeq[len(workflowEventsSeq)-1].Events
 		lastEvent := lastEvents[len(lastEvents)-1]
 		ms.executionInfo.LastRunningClock = lastEvent.GetTaskId()
-		return
+		return nil
 	}

-	if !ms.IsWorkflowExecutionRunning() && !ms.IsCurrentWorkflowGuaranteed() {
-		// If workflow currently is not running and also not running at the beginning of the transaction,
-		// then don't update the lastRunningClock
-		// NOTE: running at the beginning of the transaction == it's a current workflow in DB.
-		return
+	_, _, err := ms.UpdateLastRunningClock(nil)
+	return err
+}
+
+func (ms *MutableStateImpl) UpdateLastRunningClock(
+	eventsSeq []*persistence.WorkflowEvents,
+) (*persistencespb.WorkflowExecutionInfo, []*persistence.WorkflowEvents, error) {
+	if len(eventsSeq) > 0 || ms.IsWorkflowExecutionRunning() || ms.IsCurrentWorkflowGuaranteed() {
+		// Only update the lastRunningClock when the workflow is
+		// 1. Execution generated events in this transaction, which means it must be running at the beginning of the transaction
+		// 2. Running at the end of the transaction or
+		// 3. Running at the beginning of the transaction
+		//
+		// Condition 1 is good enough for workflow executions, but we need 2 and 3 for chasm executions.
+		//
+		// A running execution (before the transaction) is guaranteed to be the current execution.
+		// so we check 3 by calling IsCurrentWorkflowGuaranteed().
+
+		if len(eventsSeq) > 0 {
+			eventCount := 0
+			for _, batch := range eventsSeq {
+				eventCount += len(batch.Events)
+			}
+			taskIDs, err := ms.shard.GenerateTaskIDs(eventCount)
+			if err != nil {
+				return nil, nil, err
+			}
+			taskIDIndex := 0
+			for _, batch := range eventsSeq {
+				for _, event := range batch.Events {
+					event.TaskId = taskIDs[taskIDIndex]
+					taskIDIndex++
+				}
+			}
+			ms.executionInfo.LastRunningClock = taskIDs[len(taskIDs)-1]
+		} else {
+			lastRunningClock, err := ms.shard.GenerateTaskID()
+			if err != nil {
+				return nil, nil, err
+			}
+			ms.executionInfo.LastRunningClock = lastRunningClock
+		}
 	}

-	ms.executionInfo.LastRunningClock = ms.shard.CurrentVectorClock().GetClock()
+	return ms.executionInfo, eventsSeq, nil
 }

 func (ms *MutableStateImpl) closeTransactionTrackTombstones(