From 4ae488f23e8a435ef3645a7500a751bdd36e5346 Mon Sep 17 00:00:00 2001 From: Yichao Yang Date: Tue, 3 Sep 2024 15:37:33 -0700 Subject: [PATCH] Track sub state machine tombstones (#6422) ## What changed? - Track sub state machine tombstones in mutable state ## Why? - For new state-based replication stack ## How did you test it? - Unit test ## Potential risks - N/A, feature flag not enabled. ## Documentation ## Is hotfix candidate? --- common/dynamicconfig/constants.go | 5 + service/history/configs/config.go | 2 + .../history/workflow/mutable_state_impl.go | 99 ++++++++- .../workflow/mutable_state_impl_test.go | 195 +++++++++++++++++- 4 files changed, 295 insertions(+), 6 deletions(-) diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 5d6611b728..5e242c7797 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -409,6 +409,11 @@ If exceeded, failure will be truncated before being stored in mutable state.`, 1*1024*1024, `MutableStateSizeLimitWarn is the per workflow execution mutable state size limit in bytes for warning`, ) + MutableStateTombstoneCountLimit = NewGlobalIntSetting( + "limit.mutableStateTombstoneCountLimit", + 16, + `MutableStateTombstoneCountLimit is the maximum number of deleted sub state machines tracked in mutable state.`, + ) HistoryCountSuggestContinueAsNew = NewNamespaceIntSetting( "limit.historyCount.suggestContinueAsNew", 4*1024, diff --git a/service/history/configs/config.go b/service/history/configs/config.go index 03d4e80e00..e42fcc0e6d 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -235,6 +235,7 @@ type Config struct { MutableStateActivityFailureSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter MutableStateSizeLimitError dynamicconfig.IntPropertyFn MutableStateSizeLimitWarn dynamicconfig.IntPropertyFn + MutableStateTombstoneCountLimit dynamicconfig.IntPropertyFn NumPendingChildExecutionsLimit dynamicconfig.IntPropertyFnWithNamespaceFilter NumPendingActivitiesLimit dynamicconfig.IntPropertyFnWithNamespaceFilter NumPendingSignalsLimit dynamicconfig.IntPropertyFnWithNamespaceFilter @@ -573,6 +574,7 @@ func NewConfig( MutableStateActivityFailureSizeLimitWarn: dynamicconfig.MutableStateActivityFailureSizeLimitWarn.Get(dc), MutableStateSizeLimitError: dynamicconfig.MutableStateSizeLimitError.Get(dc), MutableStateSizeLimitWarn: dynamicconfig.MutableStateSizeLimitWarn.Get(dc), + MutableStateTombstoneCountLimit: dynamicconfig.MutableStateTombstoneCountLimit.Get(dc), ThrottledLogRPS: dynamicconfig.HistoryThrottledLogRPS.Get(dc), EnableStickyQuery: dynamicconfig.EnableStickyQuery.Get(dc), diff --git a/service/history/workflow/mutable_state_impl.go b/service/history/workflow/mutable_state_impl.go index 2ea83a5eb2..ad853debe9 100644 --- a/service/history/workflow/mutable_state_impl.go +++ b/service/history/workflow/mutable_state_impl.go @@ -155,7 +155,7 @@ type ( pendingSignalRequestedIDs map[string]struct{} // Set of signaled requestIds updateSignalRequestedIDs map[string]struct{} // Set of signaled requestIds since last update - deleteSignalRequestedIDs map[string]struct{} // Deleted signaled requestId + deleteSignalRequestedIDs map[string]struct{} // Deleted signaled requestId since last update executionInfo *persistencespb.WorkflowExecutionInfo // Workflow mutable state info. executionState *persistencespb.WorkflowExecutionState @@ -167,6 +167,8 @@ type ( // Running approximate total size of mutable state fields (except buffered events) when written to DB in bytes. // Buffered events are added to this value when calling GetApproximatePersistedSize. approximateSize int + // Total number of tomestones tracked in mutable state + totalTombstones int // Buffer events from DB bufferEventsInDB []*historypb.HistoryEvent // Indicates the workflow state in DB, can be used to calculate @@ -269,6 +271,7 @@ func NewMutableState( deleteSignalRequestedIDs: make(map[string]struct{}), approximateSize: 0, + totalTombstones: 0, currentVersion: namespaceEntry.FailoverVersion(), bufferEventsInDB: nil, stateInDB: enumsspb.WORKFLOW_EXECUTION_STATE_VOID, @@ -413,6 +416,10 @@ func NewMutableStateFromDB( mutableState.approximateSize += len(requestID) } + for _, tombstoneBatch := range dbRecord.ExecutionInfo.SubStateMachineTombstoneBatches { + mutableState.totalTombstones += len(tombstoneBatch.StateMachineTombstones) + } + mutableState.approximateSize += dbRecord.ExecutionState.Size() - mutableState.executionState.Size() mutableState.executionState = dbRecord.ExecutionState mutableState.approximateSize += dbRecord.ExecutionInfo.Size() - mutableState.executionInfo.Size() @@ -5214,6 +5221,7 @@ func (ms *MutableStateImpl) closeTransaction( if err := ms.closeTransactionUpdateTransitionHistory( transactionPolicy, workflowEventsSeq, + bufferEvents, ); err != nil { return closeTransactionResult{}, err } @@ -5222,6 +5230,8 @@ func (ms *MutableStateImpl) closeTransaction( transactionPolicy, ) + ms.closeTransactionTrackTombstones(transactionPolicy) + if err := ms.closeTransactionPrepareTasks( transactionPolicy, eventBatches, @@ -5318,6 +5328,7 @@ func (ms *MutableStateImpl) closeTransactionHandleSpeculativeWorkflowTask( func (ms *MutableStateImpl) closeTransactionUpdateTransitionHistory( transactionPolicy TransactionPolicy, workflowEventsSeq []*persistence.WorkflowEvents, + newBufferEvents []*historypb.HistoryEvent, ) error { if len(workflowEventsSeq) > 0 { lastEvents := workflowEventsSeq[len(workflowEventsSeq)-1].Events @@ -5340,7 +5351,12 @@ func (ms *MutableStateImpl) closeTransactionUpdateTransitionHistory( return nil } - if !ms.HSM().Dirty() && len(workflowEventsSeq) == 0 && len(ms.syncActivityTasks) == 0 { + // TODO: treat changes for transient workflow task or signalRequestID removal as state transition as well. + // Those changes are not replicated today. + if !ms.HSM().Dirty() && + len(workflowEventsSeq) == 0 && + len(newBufferEvents) == 0 && + len(ms.syncActivityTasks) == 0 { return nil } @@ -5361,7 +5377,7 @@ func (ms *MutableStateImpl) closeTransactionTrackLastUpdateVersionedTransition( return } - if len(ms.executionInfo.TransitionHistory) == 0 { + if !ms.transitionHistoryEnabled { // transition history is not enabled return } @@ -5410,6 +5426,83 @@ func (ms *MutableStateImpl) closeTransactionTrackLastUpdateVersionedTransition( // LastUpdateVersionTransition for HSM nodes already updated when transitioning the nodes. } +func (ms *MutableStateImpl) closeTransactionTrackTombstones( + transactionPolicy TransactionPolicy, +) { + if transactionPolicy != TransactionPolicyActive { + // Passive/Replication logic will update tombstone list when applying mutable state + // snapshot or mutation. + return + } + + if !ms.transitionHistoryEnabled { + // transition history is not enabled + return + } + + var tombstones []*persistencespb.StateMachineTombstone + for scheduledEventID := range ms.deleteActivityInfos { + tombstones = append(tombstones, &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_ActivityScheduledEventId{ + ActivityScheduledEventId: scheduledEventID, + }, + }) + } + for timerID := range ms.deleteTimerInfos { + tombstones = append(tombstones, &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_TimerId{ + TimerId: timerID, + }, + }) + } + for initiatedEventId := range ms.deleteChildExecutionInfos { + tombstones = append(tombstones, &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_ChildExecutionInitiatedEventId{ + ChildExecutionInitiatedEventId: initiatedEventId, + }, + }) + } + for initiatedEventId := range ms.deleteRequestCancelInfos { + tombstones = append(tombstones, &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_RequestCancelInitiatedEventId{ + RequestCancelInitiatedEventId: initiatedEventId, + }, + }) + } + for initiatedEventId := range ms.deleteSignalInfos { + tombstones = append(tombstones, &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_SignalExternalInitiatedEventId{ + SignalExternalInitiatedEventId: initiatedEventId, + }, + }) + } + // Entire signalRequestedIDs will be synced if updated, so we don't track individual signalRequestedID tombstone. + // TODO: Track signalRequestedID tombstone when we support syncing partial signalRequestedIDs. + // This requires tracking the lastUpdateVersionedTransition for each signalRequestedID, + // which is not supported by today's DB schema. + // TODO: we don't delete updateInfo and StateMachine today. Track them here when we do. + + tombstoneBatch := &persistencespb.StateMachineTombstoneBatch{ + VersionedTransition: ms.executionInfo.TransitionHistory[len(ms.executionInfo.TransitionHistory)-1], + StateMachineTombstones: tombstones, + } + ms.executionInfo.SubStateMachineTombstoneBatches = append(ms.executionInfo.SubStateMachineTombstoneBatches, tombstoneBatch) + + ms.totalTombstones += len(tombstones) + ms.capTombstoneCount() +} + +// capTombstoneCount limits the total number of tombstones stored in the mutable state. +// This method should be called whenever tombstone batch list is updated or synced. +func (ms *MutableStateImpl) capTombstoneCount() { + tombstoneCountLimit := ms.config.MutableStateTombstoneCountLimit() + for ms.totalTombstones > tombstoneCountLimit && + len(ms.executionInfo.SubStateMachineTombstoneBatches) > 0 { + ms.totalTombstones -= len(ms.executionInfo.SubStateMachineTombstoneBatches[0].StateMachineTombstones) + ms.executionInfo.SubStateMachineTombstoneBatches = ms.executionInfo.SubStateMachineTombstoneBatches[1:] + } +} + func (ms *MutableStateImpl) closeTransactionPrepareTasks( transactionPolicy TransactionPolicy, eventBatches [][]*historypb.HistoryEvent, diff --git a/service/history/workflow/mutable_state_impl_test.go b/service/history/workflow/mutable_state_impl_test.go index 6f696a794a..a7913ecad8 100644 --- a/service/history/workflow/mutable_state_impl_test.go +++ b/service/history/workflow/mutable_state_impl_test.go @@ -1037,13 +1037,14 @@ func (s *mutableStateSuite) newNamespaceCacheEntry() *namespace.Namespace { } func (s *mutableStateSuite) buildWorkflowMutableState() *persistencespb.WorkflowMutableState { - namespaceID := tests.NamespaceID + + namespaceID := s.namespaceEntry.ID() we := &commonpb.WorkflowExecution{ WorkflowId: "wId", RunId: tests.RunID, } tl := "testTaskQueue" - failoverVersion := int64(300) + failoverVersion := s.namespaceEntry.FailoverVersion() startTime := timestamppb.New(time.Date(2020, 8, 22, 1, 2, 3, 4, time.UTC)) info := &persistencespb.WorkflowExecutionInfo{ @@ -1833,6 +1834,32 @@ func (s *mutableStateSuite) TestCloseTransactionUpdateTransition() { }, versionedTransitionUpdated: true, }, + { + name: "CloseTransactionAsMutation_BufferedEvents", + dbStateMutationFn: func(dbState *persistencespb.WorkflowMutableState) { + dbState.BufferedEvents = nil + }, + txFunc: func(ms MutableState) (*persistencespb.WorkflowExecutionInfo, error) { + var activityScheduleEventID int64 + for activityScheduleEventID = range s.mutableState.GetPendingActivityInfos() { + break + } + _, err := s.mutableState.AddActivityTaskTimedOutEvent( + activityScheduleEventID, + common.EmptyEventID, + failure.NewTimeoutFailure("test-timeout", enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START), + enumspb.RETRY_STATE_TIMEOUT, + ) + s.NoError(err) + + mutation, _, err := ms.CloseTransactionAsMutation(TransactionPolicyActive) + if err != nil { + return nil, err + } + return mutation.ExecutionInfo, err + }, + versionedTransitionUpdated: true, + }, { name: "CloseTransactionAsMutation_SyncActivity", dbStateMutationFn: func(dbState *persistencespb.WorkflowMutableState) { @@ -1922,7 +1949,6 @@ func (s *mutableStateSuite) TestCloseTransactionUpdateTransition() { protorequire.ProtoSliceEqual(t, expectedTransitionHistory, execInfo.TransitionHistory) }) } - } func (s *mutableStateSuite) TestCloseTransactionTrackLastUpdateVersionedTransition() { @@ -2783,3 +2809,166 @@ func (s *mutableStateSuite) TestCloseTransactionPrepareReplicationTasks_SyncHSMT ) } } + +func (s *mutableStateSuite) TestCloseTransactionTrackTombstones() { + testCases := []struct { + name string + tombstoneFn func(ms MutableState) (*persistencespb.StateMachineTombstone, error) + }{ + { + name: "Activity", + tombstoneFn: func(mutableState MutableState) (*persistencespb.StateMachineTombstone, error) { + var activityScheduleEventID int64 + for activityScheduleEventID = range mutableState.GetPendingActivityInfos() { + break + } + _, err := mutableState.AddActivityTaskTimedOutEvent( + activityScheduleEventID, + common.EmptyEventID, + failure.NewTimeoutFailure("test-timeout", enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START), + enumspb.RETRY_STATE_TIMEOUT, + ) + return &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_ActivityScheduledEventId{ + ActivityScheduledEventId: activityScheduleEventID, + }, + }, err + }, + }, + { + name: "UserTimer", + tombstoneFn: func(mutableState MutableState) (*persistencespb.StateMachineTombstone, error) { + var timerID string + for timerID = range mutableState.GetPendingTimerInfos() { + break + } + _, err := mutableState.AddTimerFiredEvent(timerID) + return &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_TimerId{ + TimerId: timerID, + }, + }, err + }, + }, + { + name: "ChildWorkflow", + tombstoneFn: func(mutableState MutableState) (*persistencespb.StateMachineTombstone, error) { + var initiatedEventId int64 + var ci *persistencespb.ChildExecutionInfo + for initiatedEventId, ci = range mutableState.GetPendingChildExecutionInfos() { + break + } + childExecution := &commonpb.WorkflowExecution{ + WorkflowId: uuid.New(), + RunId: uuid.New(), + } + _, err := mutableState.AddChildWorkflowExecutionStartedEvent( + childExecution, + &commonpb.WorkflowType{Name: ci.WorkflowTypeName}, + initiatedEventId, + nil, + nil, + ) + if err != nil { + return nil, err + } + _, err = mutableState.AddChildWorkflowExecutionTerminatedEvent( + initiatedEventId, + childExecution, + nil, + ) + return &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_ChildExecutionInitiatedEventId{ + ChildExecutionInitiatedEventId: initiatedEventId, + }, + }, err + }, + }, + { + name: "RequestCancelExternal", + tombstoneFn: func(mutableState MutableState) (*persistencespb.StateMachineTombstone, error) { + var initiatedEventId int64 + for initiatedEventId = range mutableState.GetPendingRequestCancelExternalInfos() { + break + } + _, err := mutableState.AddRequestCancelExternalWorkflowExecutionFailedEvent( + initiatedEventId, + s.namespaceEntry.Name(), + s.namespaceEntry.ID(), + uuid.New(), + uuid.New(), + enumspb.CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_EXTERNAL_WORKFLOW_EXECUTION_NOT_FOUND, + ) + return &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_RequestCancelInitiatedEventId{ + RequestCancelInitiatedEventId: initiatedEventId, + }, + }, err + }, + }, + { + name: "SignalExternal", + tombstoneFn: func(mutableState MutableState) (*persistencespb.StateMachineTombstone, error) { + var initiatedEventId int64 + for initiatedEventId = range mutableState.GetPendingSignalExternalInfos() { + break + } + _, err := mutableState.AddSignalExternalWorkflowExecutionFailedEvent( + initiatedEventId, + s.namespaceEntry.Name(), + s.namespaceEntry.ID(), + uuid.New(), + uuid.New(), + "", + enumspb.SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_EXTERNAL_WORKFLOW_EXECUTION_NOT_FOUND, + ) + return &persistencespb.StateMachineTombstone{ + StateMachineKey: &persistencespb.StateMachineTombstone_SignalExternalInitiatedEventId{ + SignalExternalInitiatedEventId: initiatedEventId, + }, + }, err + }, + }, + } + + for _, tc := range testCases { + s.T().Run(tc.name, func(t *testing.T) { + dbState := s.buildWorkflowMutableState() + + mutableState, err := NewMutableStateFromDB(s.mockShard, s.mockEventsCache, s.logger, s.namespaceEntry, dbState, 123) + s.NoError(err) + + transitionHistory := mutableState.GetExecutionInfo().TransitionHistory + currentVersionedTransition := transitionHistory[len(transitionHistory)-1] + newVersionedTranstion := common.CloneProto(currentVersionedTransition) + newVersionedTranstion.TransitionCount += 1 + + _, err = mutableState.StartTransaction(s.namespaceEntry) + s.NoError(err) + + expectedTombstone, err := tc.tombstoneFn(mutableState) + s.NoError(err) + + _, _, err = mutableState.CloseTransactionAsMutation(TransactionPolicyActive) + s.NoError(err) + + tombstoneBatches := mutableState.GetExecutionInfo().SubStateMachineTombstoneBatches + s.Len(tombstoneBatches, 1) + tombstoneBatch := tombstoneBatches[0] + protorequire.ProtoEqual(s.T(), newVersionedTranstion, tombstoneBatch.VersionedTransition) + s.True(tombstoneExists(tombstoneBatch.StateMachineTombstones, expectedTombstone)) + }) + } +} + +func tombstoneExists( + tombstones []*persistencespb.StateMachineTombstone, + expectedTombstone *persistencespb.StateMachineTombstone, +) bool { + for _, tombstone := range tombstones { + if tombstone.Equal(expectedTombstone) { + return true + } + } + return false +}