DeleteExecutions workflow: pass nextPageToken with ContinueAsNewError (#3966)

Author: alexshtin

service/worker/deletenamespace/deleteexecutions/workflow.go

@@ -50,6 +50,7 @@ type (
 		PreviousSuccessCount int
 		PreviousErrorCount   int
 		ContinueAsNewCount   int
+		NextPageToken        []byte
 	}
 
 	DeleteExecutionsResult struct {
@@ -105,7 +106,7 @@ func DeleteExecutionsWorkflow(ctx workflow.Context, params DeleteExecutionsParam
 	logger.Info("Effective config.", tag.Value(params.Config.String()))
 
 	var a *Activities
-	var nextPageToken []byte
+	nextPageToken := params.NextPageToken
 	runningDeleteExecutionsActivityCount := 0
 	runningDeleteExecutionsSelector := workflow.NewSelector(ctx)
 	var lastDeleteExecutionsActivityErr error
@@ -186,6 +187,7 @@ func DeleteExecutionsWorkflow(ctx workflow.Context, params DeleteExecutionsParam
 	params.PreviousSuccessCount = result.SuccessCount
 	params.PreviousErrorCount = result.ErrorCount
 	params.ContinueAsNewCount++
+	params.NextPageToken = nextPageToken
 
 	logger.Info("There are more workflows to delete. Continuing workflow as new.", tag.WorkflowType(WorkflowName), tag.WorkflowNamespace(params.Namespace.String()), tag.DeletedExecutionsCount(result.SuccessCount), tag.DeletedExecutionsErrorCount(result.ErrorCount), tag.Counter(params.ContinueAsNewCount))
 	return result, workflow.NewContinueAsNewError(ctx, DeleteExecutionsWorkflow, params)

service/worker/deletenamespace/deleteexecutions/workflow_test.go

@@ -44,6 +44,7 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/payloads"
 	"go.temporal.io/server/common/persistence/visibility/manager"
 )
 
@@ -189,7 +190,7 @@ func Test_DeleteExecutionsWorkflow_ManyExecutions_ContinueAsNew(t *testing.T) {
 	var a *Activities
 
 	env.OnActivity(a.GetNextPageTokenActivity, mock.Anything, mock.Anything).Return([]byte{3, 22, 83}, nil).Times(78)
-	env.OnActivity(a.DeleteExecutionsActivity, mock.Anything, mock.Anything).Return(DeleteExecutionsActivityResult{}, nil).Times(78)
+	env.OnActivity(a.DeleteExecutionsActivity, mock.Anything, mock.Anything).Return(DeleteExecutionsActivityResult{SuccessCount: 1}, nil).Times(78)
 
 	env.ExecuteWorkflow(DeleteExecutionsWorkflow, DeleteExecutionsParams{
 		NamespaceID: "namespace-id",
@@ -205,6 +206,14 @@ func Test_DeleteExecutionsWorkflow_ManyExecutions_ContinueAsNew(t *testing.T) {
 	require.Error(t, wfErr)
 	var errContinueAsNew *workflow.ContinueAsNewError
 	require.ErrorAs(t, wfErr, &errContinueAsNew)
+
+	require.NotNil(t, errContinueAsNew.Input)
+	var newWfParams DeleteExecutionsParams
+	err := payloads.Decode(errContinueAsNew.Input, &newWfParams)
+	require.NoError(t, err)
+	require.Equal(t, 78, newWfParams.PreviousSuccessCount)
+	require.Equal(t, 0, newWfParams.PreviousErrorCount)
+	require.Equal(t, []byte{3, 22, 83}, newWfParams.NextPageToken)
 }
 
 func Test_DeleteExecutionsWorkflow_ManyExecutions_ActivityError(t *testing.T) {