[Reset] Track child workflows initiated after reset (temporalio#7210)

## What changed?
In this PR I'm recording a set of children that were initialized after
the reset point. The children are identified by
"workflow_type:workflow_id" key.

## Why?
When the parent starts to make progress after reset, it needs to know
all the children that were already started after the reset point. In the
current iteration of the feature we will use this information to
terminate any running child before starting another instance.
We need this information mainly for 2 reasons.
1. If we don't attempt to terminate the parent may not make progress due
to ID conflict policy
2. On the other hand, always terminating all children before starting is
also wrong since we may,
- Terminate any children that may be started in the future that were not
present at the time of reset.
    - Terminate children belonging to a different parent.
So, we need to record an exact set of children that were observed (at
the time of reset) to have been initialized after the reset point.

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

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->

## 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) -->

Author: gow

service/history/ndc/workflow_resetter.go

@@ -59,6 +59,14 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
+const (
+	maxChildrenInResetMutableState = 1000 // max number of children tracked in reset mutable state
+)
+
+var (
+	errWorkflowResetterMaxChildren = serviceerror.NewInvalidArgument(fmt.Sprintf("WorkflowResetter encountered max allowed children [%d] while resetting.", maxChildrenInResetMutableState))
+)
+
 type (
 	workflowResetReapplyEventsFn func(ctx context.Context, resetMutableState workflow.MutableState) error
 
@@ -648,6 +656,11 @@ func (r *workflowResetterImpl) reapplyContinueAsNewWorkflowEvents(
 		// All subsequent events should be excluded from being re-applied. So, do nothing and return.
 		return lastVisitedRunID, nil
 	}
+
+	// track the child workflows initiated after reset.
+	// This will be saved in the parent workflow (in execution info) and used by the parent later to determine how to start these child workflows again.
+	childrenInitializedAfterReset := make(map[string]*persistencespb.ResetChildInfo)
+
 	// First, special handling of remaining events for base workflow
 	nextRunID, err := r.reapplyEventsFromBranch(
 		ctx,
@@ -656,6 +669,7 @@ func (r *workflowResetterImpl) reapplyContinueAsNewWorkflowEvents(
 		baseNextEventID,
 		baseBranchToken,
 		resetReapplyExcludeTypes,
+		childrenInitializedAfterReset,
 	)
 	switch err.(type) {
 	case nil:
@@ -721,6 +735,7 @@ func (r *workflowResetterImpl) reapplyContinueAsNewWorkflowEvents(
 			nextWorkflowNextEventID,
 			nextWorkflowBranchToken,
 			resetReapplyExcludeTypes,
+			childrenInitializedAfterReset,
 		)
 		switch err.(type) {
 		case nil:
@@ -733,6 +748,9 @@ func (r *workflowResetterImpl) reapplyContinueAsNewWorkflowEvents(
 			return "", err
 		}
 	}
+	if len(childrenInitializedAfterReset) > 0 {
+		resetMutableState.SetChildrenInitializedPostResetPoint(childrenInitializedAfterReset)
+	}
 	return lastVisitedRunID, nil
 }
 
@@ -743,6 +761,7 @@ func (r *workflowResetterImpl) reapplyEventsFromBranch(
 	nextEventID int64,
 	branchToken []byte,
 	resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{},
+	childrenInitializedAfterReset map[string]*persistencespb.ResetChildInfo,
 ) (string, error) {
 
 	// TODO change this logic to fetching all workflow [baseWorkflow, currentWorkflow]
@@ -768,6 +787,21 @@ func (r *workflowResetterImpl) reapplyEventsFromBranch(
 		if _, err := r.reapplyEvents(ctx, mutableState, lastEvents, resetReapplyExcludeTypes); err != nil {
 			return "", err
 		}
+		// track the child workflows initiated after reset-point
+		for _, event := range lastEvents {
+			if event.GetEventType() == enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED {
+				attr := event.GetStartChildWorkflowExecutionInitiatedEventAttributes()
+				// TODO: there is a possibility the childIDs constructed this way may not be unique. But the probability of that is very low.
+				// Need to figure out a better way to track these child workflows.
+				childID := fmt.Sprintf("%s:%s", attr.GetWorkflowType().Name, attr.GetWorkflowId())
+				childrenInitializedAfterReset[childID] = &persistencespb.ResetChildInfo{
+					ShouldTerminateAndStart: true,
+				}
+				if len(childrenInitializedAfterReset) > maxChildrenInResetMutableState {
+					return "", errWorkflowResetterMaxChildren
+				}
+			}
+		}
 	}
 
 	if len(lastEvents) > 0 {

service/history/ndc/workflow_resetter_test.go

@@ -824,6 +824,7 @@ func (s *workflowResetterSuite) TestReapplyWorkflowEvents() {
 		nextEventID,
 		branchToken,
 		nil,
+		map[string]*persistencespb.ResetChildInfo{},
 	)
 	s.NoError(err)
 	s.Equal(newRunID, nextRunID)