Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-31 02:51:51 -07:00
Code Issues Packages Projects Releases Wiki Activity

Track sub state machine tombstones (#6422)

Browse Source 
## What changed?
<!-- Describe what has changed in this PR -->
- Track sub state machine tombstones in mutable state

## Why?
<!-- Tell your future self why have you made these changes -->
- For new state-based replication stack

## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
- Unit test

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
- N/A, feature flag not enabled.

## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->

## Is hotfix candidate?
<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->
This commit is contained in:
Yichao Yang
2024-09-03 15:37:33 -07:00
committed by GitHub
parent 22700ccf80
commit 4ae488f23e
4 changed files with 295 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
common/dynamicconfig/constants.go
View File

@@ -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,

2
service/history/configs/config.go
View File

@@ -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),

99
service/history/workflow/mutable_state_impl.go
View File

@@ -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,

195
service/history/workflow/mutable_state_impl_test.go
View File

@@ -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
}