Files
temporal/tests/workflow_reset_with_child_test.go
Stephan Behnke 15ac51c3e1 Migrate workflow_reset_with_child_test.go (#10465)
WISOTT

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-11 12:45:53 -07:00

895 lines
41 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package tests
import (
"context"
"testing"
"time"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/api/workflowservice/v1"
sdkclient "go.temporal.io/sdk/client"
"go.temporal.io/sdk/workflow"
"go.temporal.io/server/common/testing/parallelsuite"
"go.temporal.io/server/tests/testcore"
)
// Tests workflow reset feature. This suite executes the following scenarios:
// 1. Reset point is before the child init and child is not running (i.e. child already completed)
// a: If the parent uses random ChildIDs (i.e. SDK generated): Expect it to make progress by starting a completely new
// child.
// b: If the parent used static ChildIDs (keeping default Child WorkflowIDReusePolicy): Expect it to make progress by
// starting another instance of the child.
// c: If the parent used static ChildIDs and set WorkflowIDReusePolicy=WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE :
// Expect the parent to make progress by overriding the WorkflowIDReusePolicy to allow duplicate.
// 2. Reset point is before the child init and child is running at the time of reset
// a: If the parent uses random ChildIDs: Expect the parent to make progress by starting a completely new child.
// Additionally assert that the ParentClosePolicy is applied to the current running child (Note: Currently this
// doesnt work because of the bug mentioned above)
// b: If the parent uses static ChildID (keeping default Child WorkflowIDReusePolicy): Expect the parent to make
// progress by terminating the old child and starting another instance of the child.
// c: If the parent used static ChildIDs and set WorkflowIDReusePolicy=WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE :
// Expect the parent to make progress by overriding the WorkflowIDReusePolicy to allow duplicate and
// WorkflowIdConflictPolicy to terminate the old child and start another instance of the child.
// 3. Reset point is after the child init and the child is running at the time of reset.
// a: If the child was already started but not yet completed: Expect the parent to make progress by receiving the
// completion event from the running child.
// b: If the child was already completed: Expect the parent to make progress by retaining the results from the
// completed child.
// c: If the child was terminated/failed at the time of reset: Expect the parent to make progress by retaining the
// termination/failure from the child.
type WorkflowResetWithChildSuite struct {
parallelsuite.Suite[*WorkflowResetWithChildSuite]
}
func TestWorkflowResetWithChildTestSuite(t *testing.T) {
parallelsuite.Run(t, &WorkflowResetWithChildSuite{})
}
func (s *WorkflowResetWithChildSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv {
env := testcore.NewEnv(s.T(), opts...)
env.SdkWorker().RegisterWorkflow(s.workflowWithChildren)
env.SdkWorker().RegisterWorkflow(s.workflowWithWaitingChild)
env.SdkWorker().RegisterWorkflow(s.child)
env.SdkWorker().RegisterWorkflow(s.waitingChild)
env.SdkWorker().RegisterActivity(s.simpleActivity)
return env
}
// Case 1.a Reset point is before the child init and child is not running (i.e. child already completed), with random ChildIDs
// This test will create 3 child workflows and resets between each child.
func (s *WorkflowResetWithChildSuite) TestResetWithChild() {
// TODO: Enable this test when reset phase 2 is enabled.
s.T().Skip("Skipping until reset phase 2 is enabled")
env := s.newTestEnv()
// Start a workflow with 3 children.
options := s.startWorkflowOptions(env)
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithChildren, false, false)
s.NoError(err)
var originalResult string
err = run.Get(s.Context(), &originalResult)
s.NoError(err)
// save child init childIDs for later comparison.
childIDs := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), run.GetRunID())
s.Len(childIDs, 3)
child1IDBeforeReset := childIDs[0]
child2IDBeforeReset := childIDs[1]
child3IDBeforeReset := childIDs[2]
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: run.GetRunID(),
},
Reason: "integration test",
}
// (reset #1) - resetting the workflow execution before both child workflows are started.
resetRequest.RequestId = "reset-request-1"
resetRequest.WorkflowTaskFinishEventId = 4
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
s.Equal(originalResult, resultAfterReset1)
childIDsAfterReset1 := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childIDsAfterReset1, 3)
// All 3 child workflow IDs should be different after reset.
s.NotEqual(child1IDBeforeReset, childIDsAfterReset1[0])
s.NotEqual(child2IDBeforeReset, childIDsAfterReset1[1])
s.NotEqual(child3IDBeforeReset, childIDsAfterReset1[2])
// (reset #2) - resetting the new workflow execution after child-1 but before child-2
resetRequest.RequestId = "reset-request-2"
resetRequest.WorkflowExecution.RunId = resp.GetRunId()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), resp.GetRunId(), childIDsAfterReset1[0].WorkflowId)
resp, err = env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset2 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset2)
s.NoError(err)
s.Equal(originalResult, resultAfterReset2)
childIDsAfterReset2 := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childIDsAfterReset2, 3)
s.Equal(childIDsAfterReset1[0], childIDsAfterReset2[0]) // child-1 should be the same as before reset.
s.NotEqual(childIDsAfterReset1[1], childIDsAfterReset2[1]) // child-2 should be different after reset.
s.NotEqual(childIDsAfterReset1[2], childIDsAfterReset2[2]) // Child-3 should be different after reset.
// (reset #3) - resetting the new workflow execution after child-2 but before child-3
resetRequest.RequestId = "reset-request-3"
resetRequest.WorkflowExecution.RunId = resp.GetRunId()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), resp.GetRunId(), childIDsAfterReset2[1].WorkflowId)
resp, err = env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset3 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset3)
s.NoError(err)
s.Equal(originalResult, resultAfterReset3)
childIDsAfterReset3 := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childIDsAfterReset3, 3)
// child-1 & child-2 workflow IDs should be the same as before reset. Child-3 should be different.
s.Equal(childIDsAfterReset2[0], childIDsAfterReset3[0])
s.Equal(childIDsAfterReset2[1], childIDsAfterReset3[1])
s.NotEqual(childIDsAfterReset2[2], childIDsAfterReset3[2])
// (reset #3) - resetting the new workflow execution one last time after child-3
// This should successfully replay all child events and not change the child workflow IDs from previous run.
resetRequest.RequestId = "reset-request-4"
resetRequest.WorkflowExecution.RunId = resp.GetRunId()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), resp.GetRunId(), childIDsAfterReset3[2].WorkflowId)
resp, err = env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
childIDsFinal := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childIDsFinal, 3)
s.Equal(childIDsAfterReset3[0], childIDsFinal[0])
s.Equal(childIDsAfterReset3[1], childIDsFinal[1])
s.Equal(childIDsAfterReset3[2], childIDsFinal[2])
}
// Case 1.b Reset point is before the child init and child is not running (i.e. child already completed), with specified ChildIDs
// This test will create 3 child workflows and resets between each child.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_WithChildID() {
// TODO: Enable this test when reset phase 2 is enabled.
s.T().Skip("Skipping until reset phase 2 is enabled")
env := s.newTestEnv()
// Start a workflow with 3 children.
options := s.startWorkflowOptions(env)
// Executing WorkflowWithChildren with true for input fixedWID, and false for input rejectDuplicate.
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithChildren, true, false)
s.NoError(err)
var originalResult string
err = run.Get(s.Context(), &originalResult)
s.NoError(err)
// save child init initialChildExecutions for later comparison.
initialChildExecutions := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), run.GetRunID())
s.Len(initialChildExecutions, 3)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: run.GetRunID(),
},
Reason: "integration test",
}
// (reset #1) - resetting the workflow execution before all 3 child workflows are started.
resetRequest.RequestId = "reset-request-1"
resetRequest.WorkflowTaskFinishEventId = 4
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
s.Equal(originalResult, resultAfterReset1)
childExecutionsAfterReset1 := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childExecutionsAfterReset1, 3)
// All 3 child workflow IDs should be same after reset.
s.Equal(initialChildExecutions[0].WorkflowId, childExecutionsAfterReset1[0].WorkflowId)
s.Equal(initialChildExecutions[1].WorkflowId, childExecutionsAfterReset1[1].WorkflowId)
s.Equal(initialChildExecutions[2].WorkflowId, childExecutionsAfterReset1[2].WorkflowId)
// All 3 child rus IDs should be different after reset.
s.NotEqual(initialChildExecutions[0].RunId, childExecutionsAfterReset1[0].RunId)
s.NotEqual(initialChildExecutions[1].RunId, childExecutionsAfterReset1[1].RunId)
s.NotEqual(initialChildExecutions[2].RunId, childExecutionsAfterReset1[2].RunId)
// (reset #2) - resetting the new workflow execution after child-1 but before child-2
resetRequest.RequestId = "reset-request-2"
resetRequest.WorkflowExecution.RunId = resp.GetRunId()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), resp.GetRunId(), childExecutionsAfterReset1[0].WorkflowId)
resp, err = env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset2 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset2)
s.NoError(err)
s.Equal(originalResult, resultAfterReset2)
childExecutionsAfterReset2 := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childExecutionsAfterReset2, 3)
// All 3 child workflow IDs should be same after reset.
s.Equal(childExecutionsAfterReset1[0].WorkflowId, childExecutionsAfterReset2[0].WorkflowId)
s.Equal(childExecutionsAfterReset1[1].WorkflowId, childExecutionsAfterReset2[1].WorkflowId)
s.Equal(childExecutionsAfterReset1[2].WorkflowId, childExecutionsAfterReset2[2].WorkflowId)
// First child run ID should be same. But the other 2 should be different.
s.Equal(childExecutionsAfterReset1[0].RunId, childExecutionsAfterReset2[0].RunId)
s.NotEqual(childExecutionsAfterReset1[1].RunId, childExecutionsAfterReset2[1].RunId)
s.NotEqual(childExecutionsAfterReset1[2].RunId, childExecutionsAfterReset2[2].RunId)
// (reset #3) - resetting the new workflow execution after child-2 but before child-3
resetRequest.RequestId = "reset-request-3"
resetRequest.WorkflowExecution.RunId = resp.GetRunId()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), resp.GetRunId(), childExecutionsAfterReset2[1].WorkflowId)
resp, err = env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset3 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset3)
s.NoError(err)
s.Equal(originalResult, resultAfterReset3)
childExecutionsAfterReset3 := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childExecutionsAfterReset3, 3)
// All 3 child workflow IDs should be same after reset.
s.Equal(childExecutionsAfterReset2[0].WorkflowId, childExecutionsAfterReset3[0].WorkflowId)
s.Equal(childExecutionsAfterReset2[1].WorkflowId, childExecutionsAfterReset3[1].WorkflowId)
s.Equal(childExecutionsAfterReset2[2].WorkflowId, childExecutionsAfterReset3[2].WorkflowId)
// First two child run ID should be same. Third one should be different.
s.Equal(childExecutionsAfterReset2[0].RunId, childExecutionsAfterReset3[0].RunId)
s.Equal(childExecutionsAfterReset2[1].RunId, childExecutionsAfterReset3[1].RunId)
s.NotEqual(childExecutionsAfterReset2[2].RunId, childExecutionsAfterReset3[2].RunId)
// (reset #3) - resetting the new workflow execution one last time after child-3
// This should successfully replay all child events and not change the child workflow IDs from previous run.
resetRequest.RequestId = "reset-request-4"
resetRequest.WorkflowExecution.RunId = resp.GetRunId()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), resp.GetRunId(), childExecutionsAfterReset3[2].WorkflowId)
resp, err = env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset4 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset4)
s.NoError(err)
s.Equal(originalResult, resultAfterReset4)
childIDsFinal := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childIDsFinal, 3)
s.Equal(childExecutionsAfterReset3[0].WorkflowId, childIDsFinal[0].WorkflowId)
s.Equal(childExecutionsAfterReset3[1].WorkflowId, childIDsFinal[1].WorkflowId)
s.Equal(childExecutionsAfterReset3[2].WorkflowId, childIDsFinal[2].WorkflowId)
// All 3 child run IDs should be same after reset.
s.Equal(childExecutionsAfterReset3[0].RunId, childIDsFinal[0].RunId)
s.Equal(childExecutionsAfterReset3[1].RunId, childIDsFinal[1].RunId)
s.Equal(childExecutionsAfterReset3[2].RunId, childIDsFinal[2].RunId)
}
// Case 1.c Reset point is before the child init and child is not running (i.e. child already completed), with specified
// ChildIDs and WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE. This test will create 3 child workflows and resets between each child.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_WithChildID_WithRejectDuplicate() {
// TODO: Enable this test when reset phase 2 is enabled.
s.T().Skip("Skipping until reset phase 2 is enabled")
env := s.newTestEnv()
// Start a workflow with 3 children.
options := s.startWorkflowOptions(env)
// Executing WorkflowWithChildren with true for input fixedWID, and rejectDuplicate.
run, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithChildren, true, true)
s.NoError(err)
var originalResult string
err = run.Get(s.Context(), &originalResult)
s.NoError(err)
s.verifyReusePolicyIsSetForAllChild(env, env.Tv().WorkflowID(), run.GetRunID(), enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE)
// save child init initialChildExecutions for later comparison.
initialChildExecutions := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), run.GetRunID())
s.Len(initialChildExecutions, 3)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: run.GetRunID(),
},
Reason: "integration test",
}
// (reset #1) - resetting the workflow execution before all 3 child workflows are started.
resetRequest.RequestId = "reset-request-1"
resetRequest.WorkflowTaskFinishEventId = 4
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
s.Equal(originalResult, resultAfterReset1)
childExecutionsAfterReset1 := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childExecutionsAfterReset1, 3)
// All 3 child workflow IDs should be same after reset.
s.Equal(initialChildExecutions[0].WorkflowId, childExecutionsAfterReset1[0].WorkflowId)
s.Equal(initialChildExecutions[1].WorkflowId, childExecutionsAfterReset1[1].WorkflowId)
s.Equal(initialChildExecutions[2].WorkflowId, childExecutionsAfterReset1[2].WorkflowId)
// All 3 child rus IDs should be different after reset.
s.NotEqual(initialChildExecutions[0].RunId, childExecutionsAfterReset1[0].RunId)
s.NotEqual(initialChildExecutions[1].RunId, childExecutionsAfterReset1[1].RunId)
s.NotEqual(initialChildExecutions[2].RunId, childExecutionsAfterReset1[2].RunId)
// resetting the new workflow execution after child-3
// This should successfully replay all child events and not change the child workflow IDs from previous run.
resetRequest.RequestId = "reset-request-4"
resetRequest.WorkflowExecution.RunId = resp.GetRunId()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), resp.GetRunId(), initialChildExecutions[2].WorkflowId)
resp, err = env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset4 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset4)
s.NoError(err)
s.Equal(originalResult, resultAfterReset4)
childIDsFinal := s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
s.Len(childIDsFinal, 3)
s.Equal(childExecutionsAfterReset1[0].WorkflowId, childIDsFinal[0].WorkflowId)
s.Equal(childExecutionsAfterReset1[1].WorkflowId, childIDsFinal[1].WorkflowId)
s.Equal(childExecutionsAfterReset1[2].WorkflowId, childIDsFinal[2].WorkflowId)
// All 3 child run IDs should be same after reset.
s.Equal(childExecutionsAfterReset1[0].RunId, childIDsFinal[0].RunId)
s.Equal(childExecutionsAfterReset1[1].RunId, childIDsFinal[1].RunId)
s.Equal(childExecutionsAfterReset1[2].RunId, childIDsFinal[2].RunId)
}
// 2.a Reset point is before the child init and child is running at the time of reset. Child uses random WorkflowID.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_RunningChild_RandomWID() {
// TODO: Enable this test when reset phase 2 is enabled.
s.T().Skip("Skipping until reset phase 2 is enabled")
env := s.newTestEnv()
options := s.startWorkflowOptions(env)
firstRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithWaitingChild, false, false)
s.NoError(err)
// save child init initialChildExecutions for later comparison.
var initialChildExecutions []*commonpb.WorkflowExecution
s.Eventually(func() bool {
initialChildExecutions = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), firstRun.GetRunID())
return len(initialChildExecutions) == 1
}, 5*time.Second, 100*time.Millisecond)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: firstRun.GetRunID(),
},
Reason: "integration test",
}
// resetting the workflow execution before child workflow starts.
resetRequest.RequestId = "reset-request-1"
resetRequest.WorkflowTaskFinishEventId = 4
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
var childExecutionsAfterReset1 []*commonpb.WorkflowExecution
s.Eventually(func() bool {
childExecutionsAfterReset1 = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
return len(childExecutionsAfterReset1) == 1
}, 5*time.Second, 100*time.Millisecond)
// Let the second child finish by sending a signal.
err = env.SdkClient().SignalWorkflow(s.Context(), childExecutionsAfterReset1[0].WorkflowId, childExecutionsAfterReset1[0].RunId, "continue", "")
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
s.NotEqual(initialChildExecutions[0].WorkflowId, childExecutionsAfterReset1[0].WorkflowId)
s.NotEqual(initialChildExecutions[0].RunId, childExecutionsAfterReset1[0].RunId)
// Verify that the first child is still running.
descResp, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId)
s.NoError(err)
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, descResp.GetWorkflowExecutionInfo().GetStatus(),
"Child workflow should be running")
// Let the first child finish execution by sending a signal.
err = env.SdkClient().SignalWorkflow(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId, "continue", "")
s.NoError(err)
err = env.SdkClient().GetWorkflow(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
}
// 2.b Reset point is before the child init and child is running at the time of reset. Child uses fixed WorkflowID.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_RunningChild_SetWID() {
// TODO: Enable this test when reset phase 2 is enabled.
s.T().Skip("Skipping until reset phase 2 is enabled")
env := s.newTestEnv()
options := s.startWorkflowOptions(env)
firstRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithWaitingChild, true, false)
s.NoError(err)
// save child init initialChildExecutions for later comparison.
var initialChildExecutions []*commonpb.WorkflowExecution
s.Eventually(func() bool {
initialChildExecutions = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), firstRun.GetRunID())
return len(initialChildExecutions) == 1
}, 5*time.Second, 100*time.Millisecond)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: firstRun.GetRunID(),
},
Reason: "integration test",
}
// resetting the workflow execution before all 3 child workflows are started.
resetRequest.RequestId = "reset-request-1"
resetRequest.WorkflowTaskFinishEventId = 4
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Let the second child finish
var childExecutionsAfterReset1 []*commonpb.WorkflowExecution
s.Eventually(func() bool {
childExecutionsAfterReset1 = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
return len(childExecutionsAfterReset1) == 1
}, 5*time.Second, 100*time.Millisecond)
err = env.SdkClient().SignalWorkflow(s.Context(), childExecutionsAfterReset1[0].WorkflowId, childExecutionsAfterReset1[0].RunId, "continue", "")
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
s.Equal(initialChildExecutions[0].WorkflowId, childExecutionsAfterReset1[0].WorkflowId)
s.NotEqual(initialChildExecutions[0].RunId, childExecutionsAfterReset1[0].RunId)
// Verify that the first child was terminated.
descResp, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId)
s.NoError(err)
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, descResp.GetWorkflowExecutionInfo().GetStatus(),
"Child workflow should have status TERMINATED")
}
// 2.c Reset point is before the child init and child is running at the time of reset. Child uses fixed WorkflowID and POLICY_REJECT_DUPLICATE.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_RunningChild_SetWID_WithRejectDuplicate() {
// TODO: Enable this test when reset phase 2 is enabled.
s.T().Skip("Skipping until reset phase 2 is enabled")
env := s.newTestEnv()
options := s.startWorkflowOptions(env)
firstRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithWaitingChild, true, true)
s.NoError(err)
// save child init initialChildExecutions for later comparison.
var initialChildExecutions []*commonpb.WorkflowExecution
s.Eventually(func() bool {
initialChildExecutions = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), firstRun.GetRunID())
return len(initialChildExecutions) == 1
}, 5*time.Second, 100*time.Millisecond)
s.verifyReusePolicyIsSetForAllChild(env, env.Tv().WorkflowID(), firstRun.GetRunID(), enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: firstRun.GetRunID(),
},
Reason: "integration test",
}
// resetting the workflow execution before all 3 child workflows are started.
resetRequest.RequestId = "reset-request-1"
resetRequest.WorkflowTaskFinishEventId = 4
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Let the second child finish
var childExecutionsAfterReset1 []*commonpb.WorkflowExecution
s.Eventually(func() bool {
childExecutionsAfterReset1 = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), resp.GetRunId())
return len(childExecutionsAfterReset1) == 1
}, 5*time.Second, 100*time.Millisecond)
err = env.SdkClient().SignalWorkflow(s.Context(), childExecutionsAfterReset1[0].WorkflowId, childExecutionsAfterReset1[0].RunId, "continue", "")
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
s.Equal(initialChildExecutions[0].WorkflowId, childExecutionsAfterReset1[0].WorkflowId)
s.NotEqual(initialChildExecutions[0].RunId, childExecutionsAfterReset1[0].RunId)
// Verify that the first child was terminated.
descResp, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId)
s.NoError(err)
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, descResp.GetWorkflowExecutionInfo().GetStatus(),
"Child workflow should have status TERMINATED")
}
// 3.a Reset point is after the child init and the child is running at the time of reset. Child started but not completed.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_AfterStartingChild() {
env := s.newTestEnv()
options := s.startWorkflowOptions(env)
firstRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithWaitingChild, false, false)
s.NoError(err)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: firstRun.GetRunID(),
},
Reason: "integration test",
}
// save child init initialChildExecutions for later comparison.
var initialChildExecutions []*commonpb.WorkflowExecution
s.Eventually(func() bool {
initialChildExecutions = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), firstRun.GetRunID())
if len(initialChildExecutions) == 0 {
return false
}
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChildInit(env, env.Tv().WorkflowID(), firstRun.GetRunID(), initialChildExecutions[0].WorkflowId)
return resetRequest.WorkflowTaskFinishEventId != 0
}, 5*time.Second, 100*time.Millisecond)
// resetting the new workflow execution after child-1 while child-1 is still running
resetRequest.RequestId = "reset-request-2"
resetRequest.WorkflowExecution.RunId = firstRun.GetRunID()
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Let the second child finish now
err = env.SdkClient().SignalWorkflow(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId, "continue", "")
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
// verify that the child completed.
descResp, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId)
s.NoError(err)
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, descResp.GetWorkflowExecutionInfo().GetStatus(),
"Child workflow should have status COMPLETED")
}
// 3.b Reset point is after the child init and the child finished at the time of reset.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_AfterChildCompletes() {
env := s.newTestEnv()
options := s.startWorkflowOptions(env)
firstRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithWaitingChild, false, false)
s.NoError(err)
// save child init initialChildExecutions for later comparison.
var initialChildExecutions []*commonpb.WorkflowExecution
s.Eventually(func() bool {
initialChildExecutions = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), firstRun.GetRunID())
return len(initialChildExecutions) == 1
}, 5*time.Second, 100*time.Millisecond)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: firstRun.GetRunID(),
},
Reason: "integration test",
}
// Let the child finish now
err = env.SdkClient().SignalWorkflow(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId, "continue", "")
s.NoError(err)
// Wait for the child to complete.
s.Eventually(func() bool {
return s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), firstRun.GetRunID(), initialChildExecutions[0].WorkflowId) != 0
}, 5*time.Second, 100*time.Millisecond)
// resetting the new workflow execution after child initiation.
resetRequest.RequestId = "reset-request-2"
resetRequest.WorkflowExecution.RunId = firstRun.GetRunID()
resetRequest.WorkflowTaskFinishEventId = s.getWorkflowTaskFinishEventIDAfterChildInit(env, env.Tv().WorkflowID(), firstRun.GetRunID(), initialChildExecutions[0].WorkflowId)
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.NoError(err)
}
// 3.c Reset point is after the child init and the child terminated at the time of reset.
func (s *WorkflowResetWithChildSuite) TestResetWithChild_AfterChildTerminated() {
env := s.newTestEnv()
options := s.startWorkflowOptions(env)
firstRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), options, s.workflowWithWaitingChild, false, false)
s.NoError(err)
// save child init initialChildExecutions for later comparison.
var initialChildExecutions []*commonpb.WorkflowExecution
s.Eventually(func() bool {
initialChildExecutions = s.getChildWFIDsFromHistory(env, env.Tv().WorkflowID(), firstRun.GetRunID())
return len(initialChildExecutions) == 1
}, 5*time.Second, 100*time.Millisecond)
resetRequest := &workflowservice.ResetWorkflowExecutionRequest{
Namespace: env.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: env.Tv().WorkflowID(),
RunId: firstRun.GetRunID(),
},
Reason: "integration test",
}
// Terminate child
err = env.SdkClient().TerminateWorkflow(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId, "test")
s.NoError(err)
// Wait until the parent has recorded the child's terminated event and a subsequent WFT completed.
var wftAfterChildTerminated int64
s.Eventually(func() bool {
wftAfterChildTerminated = s.getWorkflowTaskFinishEventIDAfterChild(env, env.Tv().WorkflowID(), firstRun.GetRunID(), initialChildExecutions[0].WorkflowId)
return wftAfterChildTerminated != 0
}, 5*time.Second, 200*time.Millisecond)
// resetting the new workflow execution after child initiation.
resetRequest.RequestId = "reset-request-2"
resetRequest.WorkflowExecution.RunId = firstRun.GetRunID()
resetRequest.WorkflowTaskFinishEventId = wftAfterChildTerminated
resp, err := env.SdkClient().ResetWorkflowExecution(s.Context(), resetRequest)
s.NoError(err)
// Wait for the new run to complete.
var resultAfterReset1 string
err = env.SdkClient().GetWorkflow(s.Context(), env.Tv().WorkflowID(), resp.GetRunId()).Get(s.Context(), &resultAfterReset1)
s.ErrorContains(err, "child workflow execution error")
s.ErrorContains(err, "terminated")
// verify that the child is terminated.
descResp, err := env.SdkClient().DescribeWorkflowExecution(s.Context(), initialChildExecutions[0].WorkflowId, initialChildExecutions[0].RunId)
s.NoError(err)
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, descResp.GetWorkflowExecutionInfo().GetStatus(),
"Child workflow should have status TERMINATED")
}
func (s *WorkflowResetWithChildSuite) startWorkflowOptions(env *testcore.TestEnv) sdkclient.StartWorkflowOptions {
var wfOptions = sdkclient.StartWorkflowOptions{
ID: env.Tv().WorkflowID(),
TaskQueue: env.WorkerTaskQueue(),
WorkflowExecutionTimeout: 15 * time.Second,
WorkflowTaskTimeout: time.Second,
WorkflowIDReusePolicy: enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE,
}
return wfOptions
}
// workflowWithChildren starts two child workflows and waits for them to complete in sequence.
func (s *WorkflowResetWithChildSuite) workflowWithChildren(ctx workflow.Context, fixedWID bool, rejectDuplicatePolicy bool) (string, error) {
wfID := workflow.GetInfo(ctx).WorkflowExecution.ID
opt := workflow.ChildWorkflowOptions{}
if rejectDuplicatePolicy {
opt.WorkflowIDReusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE
}
if fixedWID {
opt.WorkflowID = wfID + "/test-child-workflow-1"
}
childCtx := workflow.WithChildOptions(ctx, opt)
var result string
err := workflow.ExecuteChildWorkflow(childCtx, s.child, "hello child-1").Get(ctx, &result)
if err != nil {
return "", err
}
if fixedWID {
opt.WorkflowID = wfID + "/test-child-workflow-2"
}
childCtx = workflow.WithChildOptions(ctx, opt)
var result2 string
err = workflow.ExecuteChildWorkflow(childCtx, s.child, "hello child-2").Get(ctx, &result2)
if err != nil {
return "", err
}
if fixedWID {
opt.WorkflowID = wfID + "/test-child-workflow-3"
}
childCtx = workflow.WithChildOptions(ctx, opt)
var result3 string
err = workflow.ExecuteChildWorkflow(childCtx, s.child, "hello child-2").Get(ctx, &result3)
if err != nil {
return "", err
}
return "Parent Workflow Complete", nil
}
// workflowWithWaitingChild starts three child workflows and waits for them to complete in sequence.
func (s *WorkflowResetWithChildSuite) workflowWithWaitingChild(ctx workflow.Context, fixedWID bool, rejectDuplicatePolicy bool) (string, error) {
wfID := workflow.GetInfo(ctx).WorkflowExecution.ID
var result string
opt := workflow.ChildWorkflowOptions{}
if rejectDuplicatePolicy {
opt.WorkflowIDReusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE
}
if fixedWID {
opt.WorkflowID = wfID + "/test-child-workflow-1"
}
childCtx := workflow.WithChildOptions(ctx, opt)
err := workflow.ExecuteChildWorkflow(childCtx, s.waitingChild, "hello child").Get(ctx, &result)
if err != nil {
return "", err
}
return result, err
}
func (s *WorkflowResetWithChildSuite) child(ctx workflow.Context, arg string, mustFail bool) (string, error) {
var result string
ctx = workflow.WithActivityOptions(ctx, defaultActivityOptions())
err := workflow.ExecuteActivity(ctx, s.simpleActivity, arg).Get(ctx, &result)
return result, err
}
func (s *WorkflowResetWithChildSuite) waitingChild(ctx workflow.Context, arg string) (string, error) {
ctx = workflow.WithActivityOptions(ctx, defaultActivityOptions())
workflow.GetSignalChannel(ctx, "continue").Receive(ctx, nil)
return arg, nil
}
// getWorkflowTaskFinishEventIDAfterChild gets the event ID of the first WFT completed after the child completed event.
// It does so by scanning the history of runID for any child completed events (completed, failed, canceled, timed out, terminated) and then the first WFT completed after that.
func (s *WorkflowResetWithChildSuite) getWorkflowTaskFinishEventIDAfterChild(env *testcore.TestEnv, wfID string, runID string, childID string) int64 {
iter := env.SdkClient().GetWorkflowHistory(s.Context(), wfID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
childClosedSeen := false
for iter.HasNext() {
event, err := iter.Next()
if err != nil {
break
}
switch event.GetEventType() {
case enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED:
if event.GetChildWorkflowExecutionCompletedEventAttributes().GetWorkflowExecution().GetWorkflowId() == childID {
childClosedSeen = true
}
case enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_FAILED:
if event.GetChildWorkflowExecutionFailedEventAttributes().GetWorkflowExecution().GetWorkflowId() == childID {
childClosedSeen = true
}
case enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_CANCELED:
if event.GetChildWorkflowExecutionCanceledEventAttributes().GetWorkflowExecution().GetWorkflowId() == childID {
childClosedSeen = true
}
case enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TIMED_OUT:
if event.GetChildWorkflowExecutionTimedOutEventAttributes().GetWorkflowExecution().GetWorkflowId() == childID {
childClosedSeen = true
}
case enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TERMINATED:
if event.GetChildWorkflowExecutionTerminatedEventAttributes().GetWorkflowExecution().GetWorkflowId() == childID {
childClosedSeen = true
}
default:
// Do nothing and fall through.
}
if !childClosedSeen {
continue
}
if event.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED {
return event.GetEventId()
}
}
return 0
}
func (s *WorkflowResetWithChildSuite) getWorkflowTaskFinishEventIDAfterChildInit(env *testcore.TestEnv, wfID string, runID string, childID string) int64 {
iter := env.SdkClient().GetWorkflowHistory(s.Context(), wfID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
childFound := false
for iter.HasNext() {
event, err := iter.Next()
if err != nil {
break
}
if event.GetEventType() == enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED {
if event.GetStartChildWorkflowExecutionInitiatedEventAttributes().GetWorkflowId() == childID {
childFound = true
}
}
if !childFound {
continue
}
if event.GetEventType() == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED {
return event.GetEventId()
}
}
return 0
}
func (s *WorkflowResetWithChildSuite) getChildWFIDsFromHistory(env *testcore.TestEnv, wfID string, runID string) []*commonpb.WorkflowExecution {
iter := env.SdkClient().GetWorkflowHistory(s.Context(), wfID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
var childExecutions []*commonpb.WorkflowExecution
for iter.HasNext() {
event, err1 := iter.Next()
if err1 != nil {
break
}
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED {
childExecutions = append(childExecutions, event.GetChildWorkflowExecutionStartedEventAttributes().GetWorkflowExecution())
}
}
return childExecutions
}
// verifyReusePolicy checks if a given workflow was started with REJECT_DUPLICATE
func (s *WorkflowResetWithChildSuite) verifyReusePolicyIsSetForAllChild(env *testcore.TestEnv, workflowID, runID string, expected enumspb.WorkflowIdReusePolicy) {
iter := env.SdkClient().GetWorkflowHistory(s.Context(), workflowID, runID, false, enumspb.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
if !iter.HasNext() {
s.Fail("No events found")
}
for iter.HasNext() {
event, err := iter.Next()
if err != nil {
s.Fail("Failed to get event")
return
}
if event.GetEventType() == enumspb.EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED {
policy := event.GetStartChildWorkflowExecutionInitiatedEventAttributes().GetWorkflowIdReusePolicy()
s.Equal(expected, policy)
}
}
}
func (s *WorkflowResetWithChildSuite) simpleActivity(ctx context.Context) error {
return nil
}
func defaultActivityOptions() workflow.ActivityOptions {
return workflow.ActivityOptions{
ScheduleToStartTimeout: 5 * time.Second,
ScheduleToCloseTimeout: 5 * time.Second,
StartToCloseTimeout: 9 * time.Second,
}
}