handle state machine deletion for state-based replication #7177

hai719 · 2025-01-29T18:21:03Z

What changed?

Handle state machine deletion for state-based replication.

Why?

Deletion is needed as we want to provide support for more nexus operations.

How did you test it?

unit test.

Potential risks

Documentation

Is hotfix candidate?

bergundy · 2025-01-30T18:19:28Z

service/history/workflow/mutable_state_impl.go

@@ -4264,6 +4264,29 @@ func (ms *MutableStateImpl) AddWorkflowExecutionUpdateAdmittedEvent(request *upd
 	return event, nil
 }

+func (ms *MutableStateImpl) DeleteSubStateMachine(path *persistencespb.StateMachinePath) error {
+	incomingPath := []hsm.Key{}


Suggested change

incomingPath := []hsm.Key{}

incomingPath := make([]hsm.Key, len(path))

And then you can just assign with the index.

bergundy · 2025-01-30T18:20:39Z

service/history/workflow/mutable_state_impl.go

+		ms.logDataInconsistency()
+		return nil
+	}
+	return root.DeleteChild(node.Key)


You'd need to find the parent of the node to delete the immediate child.

bergundy · 2025-01-30T18:21:36Z

service/history/workflow/mutable_state_impl.go

@@ -7265,41 +7288,8 @@ func (ms *MutableStateImpl) syncExecutionInfo(current *persistencespb.WorkflowEx
 }

 func (ms *MutableStateImpl) syncSubStateMachinesByType(incoming map[string]*persistencespb.StateMachineMap) error {


@yycptt do you see an issue with this? I assume that we always just want to do a full sync here.

hmm are we still using the TransitionCount field in StateMachineNode? If no, then I think this should work.

We can also remove the PopTask() logic in workflowStateReplicator.applySnapshot I think as we are no longer calling node.Sync which regenerates the tasks.

when source cluster is generating snapshot, the TransitionCount is reset to 1.

From the code, looks like TransitionCount is checked in ValidateNotTransitioned, and ValidateNotTransitioned is used only in components/dummy and hsmtest.

@bergundy can you confirm?

We don't use TransitionCount anymore. It's only used in unit tests.

PopTask() removed.

yycptt · 2025-01-31T01:09:52Z

service/history/workflow/mutable_state_impl.go

@@ -7265,41 +7288,8 @@ func (ms *MutableStateImpl) syncExecutionInfo(current *persistencespb.WorkflowEx
 }

 func (ms *MutableStateImpl) syncSubStateMachinesByType(incoming map[string]*persistencespb.StateMachineMap) error {


hmm are we still using the TransitionCount field in StateMachineNode? If no, then I think this should work.

We can also remove the PopTask() logic in workflowStateReplicator.applySnapshot I think as we are no longer calling node.Sync which regenerates the tasks.

yycptt · 2025-01-31T01:11:46Z

service/history/workflow/mutable_state_impl.go

@@ -7331,6 +7321,8 @@ func (ms *MutableStateImpl) applyTombstones(tombstoneBatches []*persistencespb.S
 				if _, ok := ms.pendingSignalInfoIDs[tombstone.GetSignalExternalInitiatedEventId()]; ok {
 					err = ms.DeletePendingSignal(tombstone.GetSignalExternalInitiatedEventId())
 				}
+			case *persistencespb.StateMachineTombstone_StateMachinePath:
+				err = ms.DeleteSubStateMachine(tombstone.GetStateMachinePath())


also need to TrimStateMachineTimers when nodes are being deleted.

I have a TODO in task_refresher.go about this (~L660). Probably need to find a way to trigger the trim only once.

We need to trim every time when we close a transaction, @justinp-tt is working on it.

add a flag to track if subStateMachine is deleted.

yycptt · 2025-01-31T01:33:36Z

service/history/workflow/mutable_state_impl.go

+			return err
+		}
+		ms.logError(
+			fmt.Sprintf("unable to find path: %v in subStateMachine", incomingPath),


hmm I think this could happen? If the mutation being synced may have a start point older than the workflow's current versioned transition. So some tombstones have already been synced before.

well I guess I just can't prove that situation won't happen... 🤷‍♂️

It's just printing logs so no strong opinion.

bergundy · 2025-02-04T21:37:15Z

@hai719 note that you'll need to revert #7163 and uncomment this TODO:

temporal/tests/xdc/nexus_state_replication_test.go

Line 75 in e4cfc5a

    
           // TODO(hai719): Enable this test once state based replication works with HSM node deletion.

for state based replication and deletion to work.

hai719 · 2025-02-06T22:02:11Z

@hai719 note that you'll need to revert #7163 and uncomment this TODO:

temporal/tests/xdc/nexus_state_replication_test.go

Line 75 in e4cfc5a

// TODO(hai719): Enable this test once state based replication works with HSM node deletion.

for state based replication and deletion to work.

enabled node deletion for state based replication and enabled tests.

bergundy · 2025-02-06T22:11:06Z

components/nexusoperations/events.go

@@ -306,12 +306,5 @@ func findOperationNode(root *hsm.Node, event *historypb.HistoryEvent) (*hsm.Node
 }

 func maybeDeleteNode(node *hsm.Node) error {


You can get rid of this function entirely, it's just confusing now.

bergundy · 2025-02-06T22:15:37Z

service/history/workflow/mutable_state_impl.go

+	ms.executionInfo.SubStateMachinesByType = incoming
+	ms.mustInitHSM()


Just please make sure that we regenerate the relevant timers from mutable state and delete any timers for state machines that were deleted during the full sync.
The important thing to note here is that the state machine timer group has a "scheduled" flag that indicates that there's a corresponding task in the queue, we want to make sure we don't regenerate the already scheduled tasks.

updated as discussed offline.

bergundy · 2025-02-06T22:16:53Z

service/history/workflow/mutable_state_impl.go

+	if err != nil {
+		return err
+	}
+	ms.subStateMachineDeleted = true


I don't see this being reset anywhere.

reset in cleanupTransaction()

bergundy · 2025-02-07T00:55:06Z

service/history/workflow/state_machine_timers.go

@@ -36,10 +37,16 @@ import (
 // AddNextStateMachineTimerTask generates a state machine timer task if the first deadline doesn't have a task scheduled
 // yet.
 func AddNextStateMachineTimerTask(ms MutableState) {
-	timers := ms.GetExecutionInfo().StateMachineTimers
+	// filter out empty timer groups
+	timers := util.FilterSlice(ms.GetExecutionInfo().StateMachineTimers, func(timerGroup *persistencespb.StateMachineTimerGroup) bool {


nit: I think you can use slices.DeleteFunc here but not critical.

hai719 added 2 commits January 29, 2025 10:20

handle state machine deletion for state-based replication

fb4f321

fix

d65ca80

bergundy reviewed Jan 30, 2025

View reviewed changes

yycptt reviewed Jan 31, 2025

View reviewed changes

hai719 added 2 commits February 6, 2025 10:36

Merge branch 'main' into hai719/node_deletion_for_state

430fbe0

address comments

d20d766

hai719 marked this pull request as ready for review February 6, 2025 20:06

hai719 requested a review from a team as a code owner February 6, 2025 20:06

enable node deletion for state based replication and enable tests.

e095370

bergundy reviewed Feb 6, 2025

View reviewed changes

address comments

276c3b5

hai719 requested a review from bergundy February 7, 2025 00:16

bergundy approved these changes Feb 7, 2025

View reviewed changes

address comment

d877cdb

hai719 enabled auto-merge (squash) February 7, 2025 01:29

hai719 merged commit 9a0114b into main Feb 7, 2025
50 checks passed

hai719 deleted the hai719/node_deletion_for_state branch February 7, 2025 01:31

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

handle state machine deletion for state-based replication #7177

handle state machine deletion for state-based replication #7177

hai719 commented Jan 29, 2025 •

edited

Loading

bergundy Jan 30, 2025

hai719 Feb 6, 2025

bergundy Jan 30, 2025

hai719 Feb 6, 2025

bergundy Jan 30, 2025

yycptt Jan 31, 2025

hai719 Jan 31, 2025

hai719 Feb 3, 2025

bergundy Feb 4, 2025

hai719 Feb 6, 2025

yycptt Jan 31, 2025

yycptt Jan 31, 2025

bergundy Feb 4, 2025

hai719 Feb 6, 2025

yycptt Jan 31, 2025

hai719 Feb 6, 2025

bergundy commented Feb 4, 2025

hai719 commented Feb 6, 2025

bergundy Feb 6, 2025

hai719 Feb 7, 2025

bergundy Feb 6, 2025

hai719 Feb 7, 2025

bergundy Feb 6, 2025

hai719 Feb 7, 2025

bergundy Feb 7, 2025

hai719 Feb 7, 2025

	incomingPath := []hsm.Key{}
	incomingPath := make([]hsm.Key, len(path))

		@@ -7265,41 +7288,8 @@ func (ms MutableStateImpl) syncExecutionInfo(current persistencespb.WorkflowEx
		}

		func (ms MutableStateImpl) syncSubStateMachinesByType(incoming map[string]persistencespb.StateMachineMap) error {

		@@ -306,12 +306,5 @@ func findOperationNode(root hsm.Node, event historypb.HistoryEvent) (*hsm.Node
		}

		func maybeDeleteNode(node *hsm.Node) error {

		ms.executionInfo.SubStateMachinesByType = incoming
		ms.mustInitHSM()

handle state machine deletion for state-based replication #7177

handle state machine deletion for state-based replication #7177

Conversation

hai719 commented Jan 29, 2025 • edited Loading

What changed?

Why?

How did you test it?

Potential risks

Documentation

Is hotfix candidate?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bergundy commented Feb 4, 2025

hai719 commented Feb 6, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hai719 commented Jan 29, 2025 •

edited

Loading