hanzoai/tasks
2
0
Fork 0
mirror of https://github.com/hanzoai/tasks.git synced 2026-10-04 02:04:03 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
release/v1.30.x
tasks/tests/child_workflow_test.go
T
Roey BermanandGitHub 243effc6e1 Forbid the use of pborman/uuid and remove uses (#8541) 
## What

Used the `depguard` linter to prevent importing the package and modified
all uses of it.

Also fixed a bug in the membership code where we were not updating the
next page token:

https://github.com/temporalio/temporal/pull/8541/files#diff-985dd64ffa4c6bfe0248b982dc47439b725862baa3e1f2c559b7ff522bf21969R350


## Why?

From the package README in GitHub:

> This package now leverages the github.com/google/uuid package (which
is based off an earlier version of this package).

Also, there's no point in having two uuid implementations in the
codebase.

## Test

Added the invalid import and confirmed it works:

```
../service/frontend/admin_handler_test.go:13:2: import 'github.com/pborman/uuid' is not allowed from list 'main': Importing github.com/pborman/uuid is disallowed; use github.com/google/uuid instead (depguard)
        _ "github.com/pborman/uuid"

```
2025-11-26 01:33:02 +00:00

945 lines
38 KiB
Go
Raw Permalink Blame History

package tests
import (
"fmt"
"sort"
"testing"
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/suite"
commandpb "go.temporal.io/api/command/v1"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
failurepb "go.temporal.io/api/failure/v1"
filterpb "go.temporal.io/api/filter/v1"
historypb "go.temporal.io/api/history/v1"
taskqueuepb "go.temporal.io/api/taskqueue/v1"
workflowpb "go.temporal.io/api/workflow/v1"
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/server/common/log/tag"
"go.temporal.io/server/common/payload"
"go.temporal.io/server/common/payloads"
"go.temporal.io/server/tests/testcore"
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/timestamppb"
)
type ChildWorkflowSuite struct {
testcore.FunctionalTestBase
}
func TestChildWorkflowSuite(t *testing.T) {
t.Parallel()
suite.Run(t, new(ChildWorkflowSuite))
}
func (s *ChildWorkflowSuite) TestChildWorkflowExecution() {
parentID := "functional-child-workflow-test-parent"
childID := "functional-child-workflow-test-child"
grandchildID := "functional-child-workflow-test-grandchild"
wtParent := "functional-child-workflow-test-parent-type"
wtChild := "functional-child-workflow-test-child-type"
wtGrandchild := "functional-child-workflow-test-grandchild-type"
tlParent := "functional-child-workflow-test-parent-taskqueue"
tlChild := "functional-child-workflow-test-child-taskqueue"
tlGrandchild := "functional-child-workflow-test-grandchild-taskqueue"
identity := "worker1"
saName := "CustomKeywordField"
// Uncomment this line to test with mapper.
// saName = "AliasForCustomKeywordField"
parentWorkflowType := &commonpb.WorkflowType{Name: wtParent}
childWorkflowType := &commonpb.WorkflowType{Name: wtChild}
grandchildWorkflowType := &commonpb.WorkflowType{Name: wtGrandchild}
taskQueueParent := &taskqueuepb.TaskQueue{Name: tlParent, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
taskQueueChild := &taskqueuepb.TaskQueue{Name: tlChild, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
taskQueueGrandchild := &taskqueuepb.TaskQueue{Name: tlGrandchild, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
header := &commonpb.Header{
Fields: map[string]*commonpb.Payload{"tracing": payload.EncodeString("sample payload")},
}
request := &workflowservice.StartWorkflowExecutionRequest{
RequestId: uuid.NewString(),
Namespace: s.Namespace().String(),
WorkflowId: parentID,
WorkflowType: parentWorkflowType,
TaskQueue: taskQueueParent,
Input: nil,
Header: header,
WorkflowRunTimeout: durationpb.New(100 * time.Second),
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
Identity: identity,
}
we, err0 := s.FrontendClient().StartWorkflowExecution(testcore.NewContext(), request)
s.NoError(err0)
s.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
// workflow logic
childComplete := false
childExecutionStarted := false
grandchildComplete := false
grandchildExecutionStarted := false
var parentStartedEvent *historypb.HistoryEvent
var childStartedEventFromParent *historypb.HistoryEvent
var childCompletedEventFromParent *historypb.HistoryEvent
memo := &commonpb.Memo{
Fields: map[string]*commonpb.Payload{
"Info": payload.EncodeString("memo"),
},
}
attrValPayload := payload.EncodeString("attrVal")
searchAttr := &commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{
saName: attrValPayload,
},
}
// Parent workflow logic
wtHandlerParent := func(
task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
s.Logger.Info("Processing workflow task for Parent", tag.WorkflowID(task.WorkflowExecution.WorkflowId))
parentStartedEvent = task.History.Events[0]
if task.WorkflowExecution.WorkflowId == parentID {
if !childExecutionStarted {
s.Logger.Info("Starting child execution")
childExecutionStarted = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_StartChildWorkflowExecutionCommandAttributes{
StartChildWorkflowExecutionCommandAttributes: &commandpb.StartChildWorkflowExecutionCommandAttributes{
WorkflowId: childID,
WorkflowType: childWorkflowType,
TaskQueue: taskQueueChild,
Input: payloads.EncodeString("child-workflow-input"),
Header: header,
WorkflowRunTimeout: durationpb.New(200 * time.Second),
WorkflowTaskTimeout: durationpb.New(2 * time.Second),
Control: "",
Memo: memo,
SearchAttributes: searchAttr,
},
},
}}, nil
} else if task.PreviousStartedEventId > 0 {
for _, event := range task.History.Events[task.PreviousStartedEventId:] {
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED {
childStartedEventFromParent = event
return []*commandpb.Command{}, nil
}
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED {
childCompletedEventFromParent = event
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: payloads.EncodeString("Done"),
},
},
}}, nil
}
}
}
}
return nil, nil
}
var childStartedEvent *historypb.HistoryEvent
var childRunID string
// Child workflow logic
wtHandlerChild := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
if task.PreviousStartedEventId <= 0 {
childStartedEvent = task.History.Events[0]
childRunID = task.WorkflowExecution.GetRunId()
}
s.Logger.Info("Processing workflow task for Child", tag.WorkflowID(task.WorkflowExecution.WorkflowId))
if !grandchildExecutionStarted {
s.Logger.Info("Starting grandchild execution")
grandchildExecutionStarted = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_StartChildWorkflowExecutionCommandAttributes{
StartChildWorkflowExecutionCommandAttributes: &commandpb.StartChildWorkflowExecutionCommandAttributes{
WorkflowId: grandchildID,
WorkflowType: grandchildWorkflowType,
TaskQueue: taskQueueGrandchild,
Input: payloads.EncodeString("grandchild-workflow-input"),
Header: header,
WorkflowRunTimeout: durationpb.New(200 * time.Second),
WorkflowTaskTimeout: durationpb.New(2 * time.Second),
Control: "",
Memo: memo,
SearchAttributes: searchAttr,
},
},
}}, nil
}
if task.PreviousStartedEventId > 0 {
for _, event := range task.History.Events[task.PreviousStartedEventId:] {
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED {
return []*commandpb.Command{}, nil
}
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED {
childComplete = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: payloads.EncodeString("Child Done"),
},
},
}}, nil
}
}
}
return nil, nil
}
var grandchildStartedEvent *historypb.HistoryEvent
// Grandchild workflow logic to check root workflow execution is carried correctly
wtHandlerGrandchild := func(
task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
if task.PreviousStartedEventId <= 0 {
grandchildStartedEvent = task.History.Events[0]
}
s.Logger.Info("Processing workflow task for Grandchild", tag.WorkflowID(task.WorkflowExecution.WorkflowId))
grandchildComplete = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: payloads.EncodeString("Grandchild Done"),
},
},
}}, nil
}
pollerParent := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueParent,
Identity: identity,
WorkflowTaskHandler: wtHandlerParent,
Logger: s.Logger,
T: s.T(),
}
pollerChild := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueChild,
Identity: identity,
WorkflowTaskHandler: wtHandlerChild,
Logger: s.Logger,
T: s.T(),
}
pollerGrandchild := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueGrandchild,
Identity: identity,
WorkflowTaskHandler: wtHandlerGrandchild,
Logger: s.Logger,
T: s.T(),
}
// Make first workflow task to start child execution
_, err := pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(childExecutionStarted)
s.NotNil(parentStartedEvent)
parentStartedEventAttrs := parentStartedEvent.GetWorkflowExecutionStartedEventAttributes()
// top-level workflow doesn't have parent, and root is itself (nil in history event)
s.Nil(parentStartedEventAttrs.GetParentWorkflowExecution())
s.Nil(parentStartedEventAttrs.GetRootWorkflowExecution())
// Process ChildExecution Started event and Process Child Execution and complete it
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
// Process Child workflow to start grandchild execution
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.NotNil(childStartedEventFromParent)
s.NotNil(childStartedEvent)
childStartedEventAttrsFromParent := childStartedEventFromParent.GetChildWorkflowExecutionStartedEventAttributes()
childStartedEventAttrs := childStartedEvent.GetWorkflowExecutionStartedEventAttributes()
childStartedEventSearchAttrs := childStartedEventAttrs.GetSearchAttributes()
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, childStartedEvent.GetEventType())
// check parent of child workflow is the top-level workflow
s.Equal(s.Namespace().String(), childStartedEventAttrs.GetParentWorkflowNamespace())
s.Equal(parentID, childStartedEventAttrs.ParentWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.ParentWorkflowExecution.GetRunId())
s.Equal(
childStartedEventAttrsFromParent.GetInitiatedEventId(),
childStartedEventAttrs.GetParentInitiatedEventId(),
)
// check root of child workflow is the top-level workflow
s.NotNil(childStartedEventAttrs.GetRootWorkflowExecution())
s.Equal(parentID, childStartedEventAttrs.RootWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.RootWorkflowExecution.GetRunId())
s.ProtoEqual(header, childStartedEventAttrsFromParent.Header)
s.ProtoEqual(header, childStartedEventAttrs.Header)
s.ProtoEqual(memo, childStartedEventAttrs.GetMemo())
s.Equal(
searchAttr.GetIndexedFields()[saName].GetData(),
childStartedEventSearchAttrs.GetIndexedFields()[saName].GetData(),
)
s.Equal(
"Keyword",
string(childStartedEventSearchAttrs.GetIndexedFields()[saName].GetMetadata()["type"]),
)
s.Equal(time.Duration(0), childStartedEventAttrs.GetWorkflowExecutionTimeout().AsDuration())
s.Equal(200*time.Second, childStartedEventAttrs.GetWorkflowRunTimeout().AsDuration())
// Process GrandchildExecution Started event and Process Grandchild Execution and complete it
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
// Process Grandchild workflow
_, err = pollerGrandchild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(grandchildComplete)
s.NotNil(grandchildStartedEvent)
grandchildStartedEventAttrs := grandchildStartedEvent.GetWorkflowExecutionStartedEventAttributes()
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, grandchildStartedEvent.GetEventType())
// check parent of grandchild workflow is the child workflow
s.NotNil(grandchildStartedEventAttrs.GetParentWorkflowExecution())
s.Equal(childID, grandchildStartedEventAttrs.ParentWorkflowExecution.GetWorkflowId())
s.Equal(childRunID, grandchildStartedEventAttrs.ParentWorkflowExecution.GetRunId())
// check root of grandchild workflow is the top-level workflow
s.NotNil(grandchildStartedEventAttrs.GetRootWorkflowExecution())
s.Equal(parentID, grandchildStartedEventAttrs.RootWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), grandchildStartedEventAttrs.RootWorkflowExecution.GetRunId())
// Process GrandchildExecution completed event and complete child execution
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(childComplete)
// Process ChildExecution completed event and complete parent execution
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.NotNil(childCompletedEventFromParent)
completedAttributes := childCompletedEventFromParent.GetChildWorkflowExecutionCompletedEventAttributes()
s.Equal(s.Namespace().String(), completedAttributes.Namespace)
s.Equal(s.NamespaceID().String(), completedAttributes.NamespaceId)
s.NotEmpty(completedAttributes.NamespaceId)
s.Equal(childID, completedAttributes.WorkflowExecution.WorkflowId)
s.Equal(wtChild, completedAttributes.WorkflowType.Name)
s.Equal("Child Done", s.DecodePayloadsString(completedAttributes.GetResult()))
}
func (s *ChildWorkflowSuite) TestCronChildWorkflowExecution() {
parentID := "functional-cron-child-workflow-test-parent"
childID := "functional-cron-child-workflow-test-child"
wtParent := "functional-cron-child-workflow-test-parent-type"
wtChild := "functional-cron-child-workflow-test-child-type"
tlParent := "functional-cron-child-workflow-test-parent-taskqueue"
tlChild := "functional-cron-child-workflow-test-child-taskqueue"
identity := "worker1"
cronSchedule := "@every 3s"
targetBackoffDuration := time.Second * 3
parentWorkflowType := &commonpb.WorkflowType{Name: wtParent}
childWorkflowType := &commonpb.WorkflowType{Name: wtChild}
taskQueueParent := &taskqueuepb.TaskQueue{Name: tlParent, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
taskQueueChild := &taskqueuepb.TaskQueue{Name: tlChild, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
request := &workflowservice.StartWorkflowExecutionRequest{
RequestId: uuid.NewString(),
Namespace: s.Namespace().String(),
WorkflowId: parentID,
WorkflowType: parentWorkflowType,
TaskQueue: taskQueueParent,
Input: nil,
WorkflowRunTimeout: durationpb.New(100 * time.Second),
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
Identity: identity,
}
// Because of rounding in GetBackoffForNextSchedule, we'll tend to stay aligned to whatever
// phase we start in relative to second boundaries, but drift slightly later within the second
// over time. If we cross a second boundary, one of our intervals will end up being 2s instead
// of 3s. To avoid this, wait until we can start early in the second.
for time.Now().Nanosecond()/int(time.Millisecond) > 150 {
time.Sleep(50 * time.Millisecond) //nolint:forbidigo
}
startParentWorkflowTS := time.Now().UTC()
we, err0 := s.FrontendClient().StartWorkflowExecution(testcore.NewContext(), request)
s.NoError(err0)
s.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
// workflow logic
childExecutionStarted := false
seenChildStarted := false
var terminatedEvent *historypb.HistoryEvent
// Parent workflow logic
wtHandlerParent := func(
task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
s.Logger.Info("Processing workflow task for", tag.WorkflowID(task.WorkflowExecution.WorkflowId))
if !childExecutionStarted {
s.Logger.Info("Starting child execution")
childExecutionStarted = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_StartChildWorkflowExecutionCommandAttributes{
StartChildWorkflowExecutionCommandAttributes: &commandpb.StartChildWorkflowExecutionCommandAttributes{
WorkflowId: childID,
WorkflowType: childWorkflowType,
TaskQueue: taskQueueChild,
Input: nil,
WorkflowRunTimeout: durationpb.New(200 * time.Second),
WorkflowTaskTimeout: durationpb.New(2 * time.Second),
Control: "",
CronSchedule: cronSchedule,
},
},
}}, nil
}
for _, event := range task.History.Events[task.PreviousStartedEventId:] {
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED {
seenChildStarted = true
} else if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_TERMINATED {
terminatedEvent = event
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: payloads.EncodeString("Done"),
},
},
}}, nil
}
}
return nil, nil
}
var childStartedEvent *historypb.HistoryEvent
// Child workflow logic
wtHandlerChild := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
s.Logger.Info("Processing workflow task for Child", tag.WorkflowID(task.WorkflowExecution.WorkflowId))
childStartedEvent = task.History.Events[0]
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{},
},
}}, nil
}
pollerParent := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueParent,
Identity: identity,
WorkflowTaskHandler: wtHandlerParent,
Logger: s.Logger,
T: s.T(),
}
pollerChild := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueChild,
Identity: identity,
WorkflowTaskHandler: wtHandlerChild,
Logger: s.Logger,
T: s.T(),
}
// Make first workflow task to start child execution
_, err := pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(childExecutionStarted)
// Process ChildExecution Started event
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(seenChildStarted)
// Run through three executions of the child workflow
for i := 0; i < 3; i++ {
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err), tag.Counter(i))
s.NoError(err)
s.NotNil(childStartedEvent)
childStartedEventAttrs := childStartedEvent.GetWorkflowExecutionStartedEventAttributes()
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, childStartedEvent.GetEventType())
s.Equal(s.Namespace().String(), childStartedEventAttrs.GetParentWorkflowNamespace())
s.Equal(parentID, childStartedEventAttrs.ParentWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.ParentWorkflowExecution.GetRunId())
s.NotNil(childStartedEventAttrs.GetRootWorkflowExecution())
s.Equal(parentID, childStartedEventAttrs.RootWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.RootWorkflowExecution.GetRunId())
// clean up to make sure the next poll will update this var and assert correctly
childStartedEvent = nil
}
// terminate the child workflow
_, terminateErr := s.FrontendClient().TerminateWorkflowExecution(testcore.NewContext(), &workflowservice.TerminateWorkflowExecutionRequest{
Namespace: s.Namespace().String(),
WorkflowExecution: &commonpb.WorkflowExecution{
WorkflowId: childID,
},
})
s.Nil(terminateErr)
// Process ChildExecution terminated event and complete parent execution
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.NotNil(terminatedEvent)
terminatedAttributes := terminatedEvent.GetChildWorkflowExecutionTerminatedEventAttributes()
s.Equal(childID, terminatedAttributes.WorkflowExecution.WorkflowId)
s.Equal(wtChild, terminatedAttributes.WorkflowType.Name)
startFilter := &filterpb.StartTimeFilter{}
startFilter.EarliestTime = timestamppb.New(startParentWorkflowTS)
startFilter.LatestTime = timestamppb.New(time.Now().UTC())
var closedExecutions []*workflowpb.WorkflowExecutionInfo
for i := 0; i < 10; i++ {
resp, err := s.FrontendClient().ListClosedWorkflowExecutions(testcore.NewContext(), &workflowservice.ListClosedWorkflowExecutionsRequest{
Namespace: s.Namespace().String(),
MaximumPageSize: 100,
StartTimeFilter: startFilter,
})
s.NoError(err)
if len(resp.GetExecutions()) == 5 {
closedExecutions = resp.GetExecutions()
break
}
time.Sleep(200 * time.Millisecond) //nolint:forbidigo
}
s.NotNil(closedExecutions)
sort.Slice(closedExecutions, func(i, j int) bool {
return closedExecutions[i].GetStartTime().AsTime().Before(closedExecutions[j].GetStartTime().AsTime())
})
// Execution 0 is the parent, 1, 2, 3 are the child (cron) that completed, 4 is the child that was
// terminated. Even though it was terminated, ExecutionTime will be set correctly (in the future).
lastExecution := closedExecutions[1]
for i := 2; i < 5; i++ {
executionInfo := closedExecutions[i]
// Round up the time precision to seconds
expectedBackoff := executionInfo.GetExecutionTime().AsTime().Sub(lastExecution.GetExecutionTime().AsTime())
// The execution time calculated based on last execution close time.
// However, the current execution time is based on the current start time.
// This code is to remove the diff between current start time and last execution close time.
// TODO: Remove this line once we unify the time source.
executionTimeDiff := executionInfo.GetStartTime().AsTime().Sub(lastExecution.GetCloseTime().AsTime())
// The backoff between any two executions should be a multiplier of the target backoff duration which is 3 in this test
s.Equal(0, int(expectedBackoff.Seconds()-executionTimeDiff.Seconds())%int(targetBackoffDuration.Seconds()))
lastExecution = executionInfo
}
}
func (s *ChildWorkflowSuite) TestRetryChildWorkflowExecution() {
parentID := "functional-retry-child-workflow-test-parent"
childID := "functional-retry-child-workflow-test-child"
wtParent := "functional-retry-child-workflow-test-parent-type"
wtChild := "functional-retry-child-workflow-test-child-type"
tlParent := "functional-retry-child-workflow-test-parent-taskqueue"
tlChild := "functional-retry-child-workflow-test-child-taskqueue"
identity := "worker1"
parentWorkflowType := &commonpb.WorkflowType{Name: wtParent}
childWorkflowType := &commonpb.WorkflowType{Name: wtChild}
taskQueueParent := &taskqueuepb.TaskQueue{Name: tlParent, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
taskQueueChild := &taskqueuepb.TaskQueue{Name: tlChild, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
request := &workflowservice.StartWorkflowExecutionRequest{
RequestId: uuid.NewString(),
Namespace: s.Namespace().String(),
WorkflowId: parentID,
WorkflowType: parentWorkflowType,
TaskQueue: taskQueueParent,
Input: nil,
WorkflowRunTimeout: durationpb.New(100 * time.Second),
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
Identity: identity,
}
we, err0 := s.FrontendClient().StartWorkflowExecution(testcore.NewContext(), request)
s.NoError(err0)
s.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
// workflow logic
childComplete := false
childExecutionStarted := false
var startedEvent *historypb.HistoryEvent
var completedEvent *historypb.HistoryEvent
// Parent workflow logic
wtHandlerParent := func(
task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
s.Logger.Info("Processing workflow task for Parent", tag.WorkflowID(task.WorkflowExecution.WorkflowId))
if !childExecutionStarted {
s.Logger.Info("Starting child execution")
childExecutionStarted = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_StartChildWorkflowExecutionCommandAttributes{
StartChildWorkflowExecutionCommandAttributes: &commandpb.StartChildWorkflowExecutionCommandAttributes{
WorkflowId: childID,
WorkflowType: childWorkflowType,
TaskQueue: taskQueueChild,
Input: payloads.EncodeString("child-workflow-input"),
WorkflowRunTimeout: durationpb.New(200 * time.Second),
WorkflowTaskTimeout: durationpb.New(2 * time.Second),
Control: "",
RetryPolicy: &commonpb.RetryPolicy{
InitialInterval: durationpb.New(1 * time.Millisecond),
BackoffCoefficient: 2.0,
},
},
},
}}, nil
} else if task.PreviousStartedEventId > 0 {
for _, event := range task.History.Events[task.PreviousStartedEventId:] {
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED {
startedEvent = event
return []*commandpb.Command{}, nil
}
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_COMPLETED {
completedEvent = event
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: payloads.EncodeString("Done"),
},
},
}}, nil
}
}
}
return nil, nil
}
var childStartedEvent *historypb.HistoryEvent
// Child workflow logic
wtHandlerChild := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
s.Logger.Info(
"Processing workflow task for Child",
tag.WorkflowID(task.WorkflowExecution.WorkflowId),
tag.WorkflowRunID(task.WorkflowExecution.RunId),
)
childStartedEvent = task.History.Events[0]
attempt := task.History.Events[0].GetWorkflowExecutionStartedEventAttributes().Attempt
// Fail twice, succeed on third attempt
if attempt < 3 {
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{
FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
Failure: &failurepb.Failure{
Message: fmt.Sprintf("Failed attempt %d", attempt),
},
}},
}}, nil
}
childComplete = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: payloads.EncodeString("Child Done"),
},
},
}}, nil
}
pollerParent := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueParent,
Identity: identity,
WorkflowTaskHandler: wtHandlerParent,
Logger: s.Logger,
T: s.T(),
}
pollerChild := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueChild,
Identity: identity,
WorkflowTaskHandler: wtHandlerChild,
Logger: s.Logger,
T: s.T(),
}
// Make first workflow task to start child execution
_, err := pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(childExecutionStarted)
// Process ChildExecution Started event
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.NotNil(startedEvent)
// Process Child Execution #1
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.False(childComplete)
s.NotNil(childStartedEvent)
childStartedEventAttrs := childStartedEvent.GetWorkflowExecutionStartedEventAttributes()
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, childStartedEvent.GetEventType())
s.Equal(s.Namespace().String(), childStartedEventAttrs.GetParentWorkflowNamespace())
s.Equal(parentID, childStartedEventAttrs.ParentWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.ParentWorkflowExecution.GetRunId())
s.NotNil(childStartedEventAttrs.GetRootWorkflowExecution())
s.Equal(parentID, childStartedEventAttrs.RootWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.RootWorkflowExecution.GetRunId())
// clean up to make sure the next poll will update this var and assert correctly
childStartedEvent = nil
// Process Child Execution #2
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.False(childComplete)
s.NotNil(childStartedEvent)
childStartedEventAttrs = childStartedEvent.GetWorkflowExecutionStartedEventAttributes()
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, childStartedEvent.GetEventType())
s.Equal(s.Namespace().String(), childStartedEventAttrs.GetParentWorkflowNamespace())
s.Equal(parentID, childStartedEventAttrs.ParentWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.ParentWorkflowExecution.GetRunId())
s.NotNil(childStartedEventAttrs.GetRootWorkflowExecution())
s.Equal(parentID, childStartedEventAttrs.RootWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.RootWorkflowExecution.GetRunId())
// clean up to make sure the next poll will update this var and assert correctly
childStartedEvent = nil
// Process Child Execution #3
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(childComplete)
s.NotNil(childStartedEvent)
childStartedEventAttrs = childStartedEvent.GetWorkflowExecutionStartedEventAttributes()
s.Equal(enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED, childStartedEvent.GetEventType())
s.Equal(s.Namespace().String(), childStartedEventAttrs.GetParentWorkflowNamespace())
s.Equal(parentID, childStartedEventAttrs.ParentWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.ParentWorkflowExecution.GetRunId())
s.NotNil(childStartedEventAttrs.GetRootWorkflowExecution())
s.Equal(parentID, childStartedEventAttrs.RootWorkflowExecution.GetWorkflowId())
s.Equal(we.GetRunId(), childStartedEventAttrs.RootWorkflowExecution.GetRunId())
// clean up to make sure the next poll will update this var and assert correctly
childStartedEvent = nil
// Parent should see child complete
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
// Child result should be present in completion event
s.NotNil(completedEvent)
completedAttributes := completedEvent.GetChildWorkflowExecutionCompletedEventAttributes()
s.Equal("Child Done", s.DecodePayloadsString(completedAttributes.GetResult()))
}
func (s *ChildWorkflowSuite) TestRetryFailChildWorkflowExecution() {
parentID := "functional-retry-fail-child-workflow-test-parent"
childID := "functional-retry-fail-child-workflow-test-child"
wtParent := "functional-retry-fail-child-workflow-test-parent-type"
wtChild := "functional-retry-fail-child-workflow-test-child-type"
tlParent := "functional-retry-fail-child-workflow-test-parent-taskqueue"
tlChild := "functional-retry-fail-child-workflow-test-child-taskqueue"
identity := "worker1"
parentWorkflowType := &commonpb.WorkflowType{Name: wtParent}
childWorkflowType := &commonpb.WorkflowType{Name: wtChild}
taskQueueParent := &taskqueuepb.TaskQueue{Name: tlParent, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
taskQueueChild := &taskqueuepb.TaskQueue{Name: tlChild, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}
request := &workflowservice.StartWorkflowExecutionRequest{
RequestId: uuid.NewString(),
Namespace: s.Namespace().String(),
WorkflowId: parentID,
WorkflowType: parentWorkflowType,
TaskQueue: taskQueueParent,
Input: nil,
WorkflowRunTimeout: durationpb.New(100 * time.Second),
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
Identity: identity,
}
we, err0 := s.FrontendClient().StartWorkflowExecution(testcore.NewContext(), request)
s.NoError(err0)
s.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
// workflow logic
childExecutionStarted := false
var startedEvent *historypb.HistoryEvent
var completedEvent *historypb.HistoryEvent
// Parent workflow logic
wtHandlerParent := func(
task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
s.Logger.Info("Processing workflow task for Parent", tag.WorkflowID(task.WorkflowExecution.WorkflowId))
if !childExecutionStarted {
s.Logger.Info("Starting child execution")
childExecutionStarted = true
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_START_CHILD_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_StartChildWorkflowExecutionCommandAttributes{
StartChildWorkflowExecutionCommandAttributes: &commandpb.StartChildWorkflowExecutionCommandAttributes{
WorkflowId: childID,
WorkflowType: childWorkflowType,
TaskQueue: taskQueueChild,
Input: payloads.EncodeString("child-workflow-input"),
WorkflowRunTimeout: durationpb.New(200 * time.Second),
WorkflowTaskTimeout: durationpb.New(2 * time.Second),
Control: "",
RetryPolicy: &commonpb.RetryPolicy{
InitialInterval: durationpb.New(1 * time.Millisecond),
BackoffCoefficient: 2.0,
MaximumAttempts: 3,
},
},
},
}}, nil
} else if task.PreviousStartedEventId > 0 {
for _, event := range task.History.Events[task.PreviousStartedEventId:] {
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_STARTED {
startedEvent = event
return []*commandpb.Command{}, nil
}
if event.GetEventType() == enumspb.EVENT_TYPE_CHILD_WORKFLOW_EXECUTION_FAILED {
completedEvent = event
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
Result: payloads.EncodeString("Done"),
},
},
}}, nil
}
}
}
return nil, nil
}
// Child workflow logic
wtHandlerChild := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
s.Logger.Info(
"Processing workflow task for Child",
tag.WorkflowID(task.WorkflowExecution.WorkflowId),
tag.WorkflowRunID(task.WorkflowExecution.RunId),
)
attempt := task.History.Events[0].GetWorkflowExecutionStartedEventAttributes().Attempt
// We shouldn't see more than 3 attempts
s.LessOrEqual(int(attempt), 3)
// Always fail
return []*commandpb.Command{{
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{
FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
Failure: &failurepb.Failure{
Message: fmt.Sprintf("Failed attempt %d", attempt),
},
}},
}}, nil
}
pollerParent := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueParent,
Identity: identity,
WorkflowTaskHandler: wtHandlerParent,
Logger: s.Logger,
T: s.T(),
}
pollerChild := &testcore.TaskPoller{
Client: s.FrontendClient(),
Namespace: s.Namespace().String(),
TaskQueue: taskQueueChild,
Identity: identity,
WorkflowTaskHandler: wtHandlerChild,
Logger: s.Logger,
T: s.T(),
}
// Make first workflow task to start child execution
_, err := pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.True(childExecutionStarted)
// Process ChildExecution Started event
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
s.NotNil(startedEvent)
// Process Child Execution #1
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
// Process Child Execution #2
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
// Process Child Execution #3
_, err = pollerChild.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
// Parent should see child complete
_, err = pollerParent.PollAndProcessWorkflowTask()
s.Logger.Info("PollAndProcessWorkflowTask", tag.Error(err))
s.NoError(err)
// Child failure should be present in completion event
s.NotNil(completedEvent)
attrs := completedEvent.GetChildWorkflowExecutionFailedEventAttributes()
s.Equal(attrs.Failure.Message, "Failed attempt 3")
}
Reference in New Issue View Git Blame Copy Permalink