mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
18c8ab7f77
## What changed? Replace `payload.Encode` with `sadefs.MustEncodeValue` for encoding predefined search attributes used internally. ## Why? `sadefs.MustEncodeValue` adds metadata type to the payload. ## How did you test it? - [x] built - [ ] run locally and tested manually - [x] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks
643 lines
24 KiB
Go
643 lines
24 KiB
Go
package tests
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"sort"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
commandpb "go.temporal.io/api/command/v1"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
filterpb "go.temporal.io/api/filter/v1"
|
|
taskqueuepb "go.temporal.io/api/taskqueue/v1"
|
|
workflowpb "go.temporal.io/api/workflow/v1"
|
|
"go.temporal.io/api/workflowservice/v1"
|
|
sdkclient "go.temporal.io/sdk/client"
|
|
"go.temporal.io/sdk/workflow"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/failure"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/payload"
|
|
"go.temporal.io/server/common/payloads"
|
|
"go.temporal.io/server/common/searchattribute/sadefs"
|
|
"go.temporal.io/server/common/testing/parallelsuite"
|
|
"go.temporal.io/server/tests/testcore"
|
|
"google.golang.org/protobuf/types/known/durationpb"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
type CronTestSuite struct {
|
|
parallelsuite.Suite[*CronTestSuite]
|
|
}
|
|
|
|
type CronTestClientSuite struct {
|
|
parallelsuite.Suite[*CronTestClientSuite]
|
|
}
|
|
|
|
func TestCronTestSuite(t *testing.T) {
|
|
parallelsuite.Run(t, &CronTestSuite{})
|
|
}
|
|
|
|
func TestCronTestClientSuite(t *testing.T) {
|
|
parallelsuite.Run(t, &CronTestClientSuite{})
|
|
}
|
|
|
|
func (s *CronTestSuite) TestCronWorkflow_Failed_Infinite() {
|
|
env := testcore.NewEnv(s.T())
|
|
|
|
id := "functional-wf-cron-failed-infinite-test"
|
|
wt := "functional-wf-cron-failed-infinite-type"
|
|
tl := "functional-wf-cron-failed-infinite-taskqueue"
|
|
identity := "worker1"
|
|
cronSchedule := "@every 5s"
|
|
|
|
request := &workflowservice.StartWorkflowExecutionRequest{
|
|
RequestId: uuid.NewString(),
|
|
Namespace: env.Namespace().String(),
|
|
WorkflowId: id,
|
|
WorkflowType: &commonpb.WorkflowType{Name: wt},
|
|
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
|
|
Input: nil,
|
|
WorkflowRunTimeout: durationpb.New(5 * time.Second),
|
|
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
|
|
Identity: identity,
|
|
CronSchedule: cronSchedule, // minimum interval by standard spec is 1m (* * * * *, use non-standard descriptor for short interval for test
|
|
RetryPolicy: &commonpb.RetryPolicy{
|
|
MaximumAttempts: 2,
|
|
MaximumInterval: durationpb.New(1 * time.Second),
|
|
BackoffCoefficient: 1.2,
|
|
},
|
|
}
|
|
|
|
we, err0 := env.FrontendClient().StartWorkflowExecution(s.Context(), request)
|
|
s.NoError(err0)
|
|
|
|
env.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
|
|
|
|
respondFailed := false
|
|
seeRetry := false
|
|
wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
|
|
|
|
if !respondFailed {
|
|
respondFailed = true
|
|
|
|
return []*commandpb.Command{
|
|
{
|
|
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
|
|
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{
|
|
FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
|
|
Failure: failure.NewServerFailure("cron error for retry", false),
|
|
}},
|
|
}}, nil
|
|
}
|
|
|
|
startEvent := task.History.Events[0]
|
|
seeRetry = startEvent.GetWorkflowExecutionStartedEventAttributes().Initiator == enumspb.CONTINUE_AS_NEW_INITIATOR_RETRY
|
|
return []*commandpb.Command{
|
|
{
|
|
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
|
|
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
|
|
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
|
|
Result: nil,
|
|
}},
|
|
}}, nil
|
|
}
|
|
|
|
poller := &testcore.TaskPoller{
|
|
Client: env.FrontendClient(),
|
|
Namespace: env.Namespace().String(),
|
|
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
|
|
Identity: identity,
|
|
WorkflowTaskHandler: wtHandler,
|
|
Logger: env.Logger,
|
|
T: s.T(),
|
|
}
|
|
|
|
env.Logger.Info("Process first cron run which fails")
|
|
_, err := poller.PollAndProcessWorkflowTask(testcore.WithDumpHistory)
|
|
s.NoError(err)
|
|
|
|
env.Logger.Info("Process first cron run which completes")
|
|
_, err = poller.PollAndProcessWorkflowTask(testcore.WithDumpHistory)
|
|
s.NoError(err)
|
|
|
|
s.True(seeRetry)
|
|
}
|
|
|
|
func (s *CronTestSuite) TestCronWorkflow() {
|
|
env := testcore.NewEnv(s.T())
|
|
|
|
id := "functional-wf-cron-test"
|
|
wt := "functional-wf-cron-type"
|
|
tl := "functional-wf-cron-taskqueue"
|
|
identity := "worker1"
|
|
cronSchedule := "@every 3s"
|
|
|
|
targetBackoffDuration := time.Second * 3
|
|
backoffDurationTolerance := time.Millisecond * 500
|
|
|
|
memo := &commonpb.Memo{
|
|
Fields: map[string]*commonpb.Payload{"memoKey": payload.EncodeString("memoVal")},
|
|
}
|
|
searchAttr := &commonpb.SearchAttributes{
|
|
IndexedFields: map[string]*commonpb.Payload{
|
|
"CustomKeywordField": sadefs.MustEncodeValue("keyword-value", enumspb.INDEXED_VALUE_TYPE_KEYWORD),
|
|
},
|
|
}
|
|
|
|
request := &workflowservice.StartWorkflowExecutionRequest{
|
|
RequestId: uuid.NewString(),
|
|
Namespace: env.Namespace().String(),
|
|
WorkflowId: id,
|
|
WorkflowType: &commonpb.WorkflowType{Name: wt},
|
|
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
|
|
Input: nil,
|
|
WorkflowRunTimeout: durationpb.New(100 * time.Second),
|
|
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
|
|
Identity: identity,
|
|
CronSchedule: cronSchedule, // minimum interval by standard spec is 1m (* * * * *, use non-standard descriptor for short interval for test
|
|
Memo: memo,
|
|
SearchAttributes: searchAttr,
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
startWorkflowTS := time.Now().UTC()
|
|
we, err0 := env.FrontendClient().StartWorkflowExecution(s.Context(), request)
|
|
s.NoError(err0)
|
|
|
|
env.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
|
|
|
|
var executions []*commonpb.WorkflowExecution
|
|
|
|
wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
|
|
if task.PreviousStartedEventId == common.EmptyEventID {
|
|
startedEvent := task.History.Events[0]
|
|
if startedEvent.GetEventType() != enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
|
|
return []*commandpb.Command{
|
|
{
|
|
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
|
|
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
|
|
Failure: failure.NewServerFailure("incorrect first event", true),
|
|
}},
|
|
}}, nil
|
|
}
|
|
|
|
// Just check that it can be decoded
|
|
var lcr int
|
|
s.NoError(payloads.Decode(startedEvent.GetWorkflowExecutionStartedEventAttributes().GetLastCompletionResult(), &lcr))
|
|
}
|
|
|
|
executions = append(executions, task.WorkflowExecution)
|
|
if len(executions) >= 3 {
|
|
return []*commandpb.Command{
|
|
{
|
|
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
|
|
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
|
|
Result: payloads.EncodeString("cron-test-result"),
|
|
}},
|
|
}}, nil
|
|
}
|
|
return []*commandpb.Command{
|
|
{
|
|
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
|
|
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
|
|
Failure: failure.NewServerFailure("cron-test-error", false),
|
|
}},
|
|
}}, nil
|
|
}
|
|
|
|
poller := &testcore.TaskPoller{
|
|
Client: env.FrontendClient(),
|
|
Namespace: env.Namespace().String(),
|
|
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
|
|
Identity: identity,
|
|
WorkflowTaskHandler: wtHandler,
|
|
Logger: env.Logger,
|
|
T: s.T(),
|
|
}
|
|
|
|
startFilter := &filterpb.StartTimeFilter{}
|
|
startFilter.EarliestTime = timestamppb.New(startWorkflowTS)
|
|
startFilter.LatestTime = timestamppb.New(time.Now().UTC())
|
|
|
|
// Sleep some time before checking the open executions.
|
|
// This will not cost extra time as the polling for first workflow task will be blocked for 3 seconds.
|
|
time.Sleep(2 * time.Second) //nolint:forbidigo
|
|
resp, err := env.FrontendClient().ListOpenWorkflowExecutions(s.Context(), &workflowservice.ListOpenWorkflowExecutionsRequest{
|
|
Namespace: env.Namespace().String(),
|
|
MaximumPageSize: 100,
|
|
StartTimeFilter: startFilter,
|
|
Filters: &workflowservice.ListOpenWorkflowExecutionsRequest_ExecutionFilter{ExecutionFilter: &filterpb.WorkflowExecutionFilter{
|
|
WorkflowId: id,
|
|
}},
|
|
})
|
|
s.NoError(err)
|
|
s.Len(resp.GetExecutions(), 1)
|
|
executionInfo := resp.GetExecutions()[0]
|
|
s.Equal(targetBackoffDuration, executionInfo.GetExecutionTime().AsTime().Sub(executionInfo.GetStartTime().AsTime()))
|
|
|
|
_, err = poller.PollAndProcessWorkflowTask()
|
|
s.NoError(err)
|
|
|
|
// Make sure the cron workflow start running at a proper time, in this case 3 seconds after the
|
|
// startWorkflowExecution request
|
|
backoffDuration := time.Now().UTC().Sub(startWorkflowTS)
|
|
s.Greater(backoffDuration, targetBackoffDuration)
|
|
s.Less(backoffDuration, targetBackoffDuration+backoffDurationTolerance)
|
|
|
|
_, err = poller.PollAndProcessWorkflowTask()
|
|
s.NoError(err)
|
|
|
|
_, err = poller.PollAndProcessWorkflowTask()
|
|
s.NoError(err)
|
|
|
|
s.Len(executions, 3)
|
|
|
|
_, terminateErr := env.FrontendClient().TerminateWorkflowExecution(s.Context(), &workflowservice.TerminateWorkflowExecutionRequest{
|
|
Namespace: env.Namespace().String(),
|
|
WorkflowExecution: &commonpb.WorkflowExecution{
|
|
WorkflowId: id,
|
|
},
|
|
})
|
|
s.NoError(terminateErr)
|
|
|
|
// first two should be failures
|
|
for i := range 2 {
|
|
events := env.GetHistory(env.Namespace().String(), executions[i])
|
|
s.EqualHistoryEvents(fmt.Sprintf(`
|
|
1 WorkflowExecutionStarted {"Memo":{"Fields":{"memoKey":{"Data":"\"memoVal\""}}},"SearchAttributes":{"IndexedFields":{"CustomKeywordField":{"Data":"\"keyword-value\"","Metadata":{"type":"Keyword"}}}}}
|
|
2 WorkflowTaskScheduled
|
|
3 WorkflowTaskStarted
|
|
4 WorkflowTaskCompleted
|
|
5 WorkflowExecutionFailed {"Failure":{"Message":"cron-test-error"},"NewExecutionRunId":"%s"}
|
|
`, executions[i+1].RunId), events)
|
|
}
|
|
|
|
// third should be completed
|
|
events := env.GetHistory(env.Namespace().String(), executions[2])
|
|
s.EqualHistoryEvents(`
|
|
1 WorkflowExecutionStarted {"Memo":{"Fields":{"memoKey":{"Data":"\"memoVal\""}}},"SearchAttributes":{"IndexedFields":{"CustomKeywordField":{"Data":"\"keyword-value\"","Metadata":{"type":"Keyword"}}}}}
|
|
2 WorkflowTaskScheduled
|
|
3 WorkflowTaskStarted
|
|
4 WorkflowTaskCompleted
|
|
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"cron-test-result\""}]}}
|
|
`, events)
|
|
|
|
startFilter.LatestTime = timestamppb.New(time.Now().UTC())
|
|
var closedExecutions []*workflowpb.WorkflowExecutionInfo
|
|
for range 10 {
|
|
resp, err := env.FrontendClient().ListClosedWorkflowExecutions(s.Context(), &workflowservice.ListClosedWorkflowExecutionsRequest{
|
|
Namespace: env.Namespace().String(),
|
|
MaximumPageSize: 100,
|
|
StartTimeFilter: startFilter,
|
|
Filters: &workflowservice.ListClosedWorkflowExecutionsRequest_ExecutionFilter{ExecutionFilter: &filterpb.WorkflowExecutionFilter{
|
|
WorkflowId: id,
|
|
}},
|
|
})
|
|
s.NoError(err)
|
|
if len(resp.GetExecutions()) == 4 {
|
|
closedExecutions = resp.GetExecutions()
|
|
break
|
|
}
|
|
time.Sleep(200 * time.Millisecond) //nolint:forbidigo
|
|
}
|
|
s.NotNil(closedExecutions)
|
|
dweResponse, err := env.FrontendClient().DescribeWorkflowExecution(s.Context(), &workflowservice.DescribeWorkflowExecutionRequest{
|
|
Namespace: env.Namespace().String(),
|
|
Execution: &commonpb.WorkflowExecution{
|
|
WorkflowId: id,
|
|
RunId: we.RunId,
|
|
},
|
|
})
|
|
s.NoError(err)
|
|
expectedExecutionTime := dweResponse.WorkflowExecutionInfo.GetStartTime().AsTime().Add(3 * time.Second)
|
|
s.Equal(expectedExecutionTime, dweResponse.WorkflowExecutionInfo.GetExecutionTime().AsTime())
|
|
|
|
sort.Slice(closedExecutions, func(i, j int) bool {
|
|
return closedExecutions[i].GetStartTime().AsTime().Before(closedExecutions[j].GetStartTime().AsTime())
|
|
})
|
|
lastExecution := closedExecutions[0]
|
|
for i := 1; i < 4; i++ {
|
|
executionInfo := closedExecutions[i]
|
|
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-executionTimeDiff).Round(time.Second).Seconds())%int(targetBackoffDuration.Seconds()),
|
|
"expected backoff %v-%v=%v should be multiplier of target backoff %v",
|
|
expectedBackoff.Seconds(),
|
|
executionTimeDiff.Seconds(),
|
|
(expectedBackoff - executionTimeDiff).Round(time.Second).Seconds(),
|
|
targetBackoffDuration.Seconds())
|
|
lastExecution = executionInfo
|
|
|
|
// TODO: Remove the describeWorkflowExecution call when firstRunID in WorkflowExecutionInfo
|
|
// is populated by Visibility api as well.
|
|
dweResponse, err := env.FrontendClient().DescribeWorkflowExecution(s.Context(), &workflowservice.DescribeWorkflowExecutionRequest{
|
|
Namespace: env.Namespace().String(),
|
|
Execution: executionInfo.GetExecution(),
|
|
})
|
|
s.NoError(err)
|
|
s.Equal(we.RunId, dweResponse.WorkflowExecutionInfo.GetFirstRunId())
|
|
}
|
|
}
|
|
|
|
func (s *CronTestClientSuite) TestCronWorkflowCompletionStates() {
|
|
// Run a cron workflow that completes in (almost) all the possible ways:
|
|
// Run 1: succeeds
|
|
// Run 2: fails
|
|
// Run 3: times out
|
|
// Run 4: succeeds
|
|
// Run 5: succeeds
|
|
// Run 6: terminated before it runs
|
|
|
|
// Continue-as-new is not tested (behavior is currently not correct)
|
|
|
|
env := testcore.NewEnv(s.T())
|
|
|
|
id := "functional-wf-cron-failed-test"
|
|
cronSchedule := "@every 3s"
|
|
|
|
targetBackoffDuration := 3 * time.Second
|
|
workflowRunTimeout := 5 * time.Second
|
|
tolerance := 500 * time.Millisecond
|
|
durationNear := func(value, target time.Duration) {
|
|
s.T().Helper()
|
|
s.Greater(value, target-tolerance)
|
|
s.Less(value, target+tolerance)
|
|
}
|
|
|
|
runIDs := make(map[string]bool)
|
|
wfCh := make(chan int)
|
|
|
|
workflowFn := func(ctx workflow.Context) (string, error) {
|
|
runIDs[workflow.GetInfo(ctx).WorkflowExecution.RunID] = true
|
|
iteration := len(runIDs)
|
|
wfCh <- iteration
|
|
|
|
var lcr string
|
|
switch iteration {
|
|
case 1:
|
|
s.False(workflow.HasLastCompletionResult(ctx))
|
|
s.NoError(workflow.GetLastError(ctx))
|
|
return "pass", nil
|
|
|
|
case 2:
|
|
s.True(workflow.HasLastCompletionResult(ctx))
|
|
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
|
|
s.Equal("pass", lcr)
|
|
s.NoError(workflow.GetLastError(ctx))
|
|
return "", errors.New("second error") //nolint:err113
|
|
|
|
case 3:
|
|
s.True(workflow.HasLastCompletionResult(ctx))
|
|
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
|
|
s.Equal("pass", lcr)
|
|
s.Error(workflow.GetLastError(ctx))
|
|
s.Equal("second error", workflow.GetLastError(ctx).Error())
|
|
s.NoError(workflow.Sleep(ctx, 10*time.Second)) // cause wft timeout
|
|
panic("should have been timed out on server already")
|
|
|
|
case 4:
|
|
s.True(workflow.HasLastCompletionResult(ctx))
|
|
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
|
|
s.Equal("pass", lcr)
|
|
s.Error(workflow.GetLastError(ctx))
|
|
s.Equal("workflow timeout (type: StartToClose)", workflow.GetLastError(ctx).Error())
|
|
return "pass again", nil
|
|
|
|
case 5:
|
|
s.True(workflow.HasLastCompletionResult(ctx))
|
|
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
|
|
s.Equal("pass again", lcr)
|
|
s.NoError(workflow.GetLastError(ctx))
|
|
return "final pass", nil
|
|
}
|
|
|
|
panic("shouldn't get here")
|
|
}
|
|
|
|
env.SdkWorker().RegisterWorkflow(workflowFn)
|
|
|
|
// 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
|
|
}
|
|
|
|
workflowOptions := sdkclient.StartWorkflowOptions{
|
|
ID: id,
|
|
TaskQueue: env.WorkerTaskQueue(),
|
|
WorkflowRunTimeout: workflowRunTimeout,
|
|
CronSchedule: cronSchedule,
|
|
}
|
|
ts := time.Now()
|
|
startTs := ts
|
|
_, err := env.SdkClient().ExecuteWorkflow(s.Context(), workflowOptions, workflowFn)
|
|
s.NoError(err)
|
|
|
|
// check execution and history of first run
|
|
exec := s.listOpenWorkflowExecutions(env, startTs, time.Now(), id, 1)[0]
|
|
firstRunID := exec.GetExecution().RunId
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, exec.GetStatus())
|
|
historyEvents := env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(fmt.Sprintf(`
|
|
1 WorkflowExecutionStarted {"ContinuedExecutionRunId":"","CronSchedule":"@every 3s","FirstExecutionRunId":"%s", "Initiator":3}`, firstRunID), historyEvents)
|
|
attrs1 := historyEvents[0].GetWorkflowExecutionStartedEventAttributes()
|
|
// not `"FirstWorkflowTaskBackoff":{"Nanos":0,"Seconds":3}` in the history above because DurationNear is not supported by EqualHistoryEvents.
|
|
durationNear(attrs1.FirstWorkflowTaskBackoff.AsDuration(), targetBackoffDuration)
|
|
|
|
// wait for first run
|
|
s.Equal(1, <-wfCh)
|
|
durationNear(time.Since(ts), targetBackoffDuration)
|
|
ts = time.Now()
|
|
|
|
// let first run finish, then check execution and history of second run
|
|
s.Eventually(
|
|
func() bool {
|
|
exec = s.listOpenWorkflowExecutions(env, startTs, time.Now(), id, 1)[0]
|
|
return exec.GetExecution().GetRunId() != firstRunID
|
|
},
|
|
targetBackoffDuration+tolerance,
|
|
250*time.Millisecond,
|
|
)
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, exec.GetStatus())
|
|
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(fmt.Sprintf(`
|
|
1 WorkflowExecutionStarted {"ContinuedExecutionRunId":"%s","CronSchedule":"@every 3s","FirstExecutionRunId":"%s", "Initiator":%d}`, firstRunID, firstRunID, enumspb.CONTINUE_AS_NEW_INITIATOR_CRON_SCHEDULE), historyEvents)
|
|
attrs2 := historyEvents[0].GetWorkflowExecutionStartedEventAttributes()
|
|
// not `"FirstWorkflowTaskBackoff":{"Nanos":0,"Seconds":3}` in the history above because DurationNear is not supported by EqualHistoryEvents.
|
|
durationNear(attrs2.FirstWorkflowTaskBackoff.AsDuration(), targetBackoffDuration)
|
|
|
|
// wait for second run
|
|
s.Equal(2, <-wfCh)
|
|
durationNear(time.Since(ts), targetBackoffDuration)
|
|
ts = time.Now()
|
|
|
|
// don't bother checking started events for subsequent runs, we covered the important parts already
|
|
|
|
// wait for third run
|
|
s.Equal(3, <-wfCh)
|
|
durationNear(time.Since(ts), targetBackoffDuration)
|
|
ts = time.Now()
|
|
|
|
// wait for fourth run (third one waits for timeout after 5s, so will run after 6s)
|
|
s.Equal(4, <-wfCh)
|
|
durationNear(time.Since(ts), 2*targetBackoffDuration)
|
|
ts = time.Now()
|
|
|
|
// wait for fifth run
|
|
s.Equal(5, <-wfCh)
|
|
durationNear(time.Since(ts), targetBackoffDuration)
|
|
|
|
// let fifth run finish and sixth get scheduled
|
|
_ = s.listClosedWorkflowExecutions(env, startTs, time.Now().Add(targetBackoffDuration), id, 5)
|
|
_ = s.listOpenWorkflowExecutions(env, startTs, time.Now().Add(targetBackoffDuration), id, 1)
|
|
// then terminate
|
|
s.NoError(env.SdkClient().TerminateWorkflow(s.Context(), id, "", "test is over"))
|
|
|
|
closedExecutions := s.listClosedWorkflowExecutions(env, startTs, time.Now(), id, 6)
|
|
|
|
exec = closedExecutions[5] // first: success
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, exec.GetStatus())
|
|
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(`
|
|
1 WorkflowExecutionStarted
|
|
2 WorkflowTaskScheduled
|
|
3 WorkflowTaskStarted
|
|
4 WorkflowTaskCompleted
|
|
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"pass\""}]}}`, historyEvents)
|
|
|
|
exec = closedExecutions[4] // second: fail
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_FAILED, exec.GetStatus())
|
|
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(`
|
|
1 WorkflowExecutionStarted
|
|
2 WorkflowTaskScheduled
|
|
3 WorkflowTaskStarted
|
|
4 WorkflowTaskCompleted
|
|
5 WorkflowExecutionFailed {"Failure":{"Message":"second error"}}`, historyEvents)
|
|
|
|
exec = closedExecutions[3] // third: timed out
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT, exec.GetStatus())
|
|
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(`
|
|
1 WorkflowExecutionStarted
|
|
2 WorkflowTaskScheduled
|
|
3 WorkflowTaskStarted
|
|
4 WorkflowTaskCompleted
|
|
5 TimerStarted
|
|
6 WorkflowExecutionTimedOut {"RetryState":5} // enumspb.RETRY_STATE_RETRY_POLICY_NOT_SET`, historyEvents)
|
|
|
|
exec = closedExecutions[2] // fourth: success
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, exec.GetStatus())
|
|
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(`
|
|
1 WorkflowExecutionStarted
|
|
2 WorkflowTaskScheduled
|
|
3 WorkflowTaskStarted
|
|
4 WorkflowTaskCompleted
|
|
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"pass again\""}]}}`, historyEvents)
|
|
|
|
exec = closedExecutions[1] // fifth: success
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, exec.GetStatus())
|
|
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(`
|
|
1 WorkflowExecutionStarted
|
|
2 WorkflowTaskScheduled
|
|
3 WorkflowTaskStarted
|
|
4 WorkflowTaskCompleted
|
|
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"final pass\""}]}}`, historyEvents)
|
|
|
|
exec = closedExecutions[0] // sixth: terminated
|
|
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, exec.GetStatus())
|
|
events := env.GetHistory(env.Namespace().String(), exec.GetExecution())
|
|
s.EqualHistoryEvents(`
|
|
1 WorkflowExecutionStarted
|
|
2 WorkflowExecutionTerminated {"Reason":"test is over"}`, events)
|
|
}
|
|
|
|
func (s *CronTestClientSuite) listOpenWorkflowExecutions(env *testcore.TestEnv, start, end time.Time, id string, expectedNumber int) []*workflowpb.WorkflowExecutionInfo {
|
|
s.T().Helper()
|
|
var resp *workflowservice.ListOpenWorkflowExecutionsResponse
|
|
s.Eventuallyf(
|
|
func() bool {
|
|
var err error
|
|
resp, err = env.SdkClient().ListOpenWorkflow(
|
|
s.Context(), &workflowservice.ListOpenWorkflowExecutionsRequest{
|
|
Namespace: env.Namespace().String(),
|
|
MaximumPageSize: int32(2 * expectedNumber),
|
|
StartTimeFilter: &filterpb.StartTimeFilter{
|
|
EarliestTime: timestamppb.New(start),
|
|
LatestTime: timestamppb.New(end),
|
|
},
|
|
Filters: &workflowservice.ListOpenWorkflowExecutionsRequest_ExecutionFilter{
|
|
ExecutionFilter: &filterpb.WorkflowExecutionFilter{
|
|
WorkflowId: id,
|
|
},
|
|
},
|
|
},
|
|
)
|
|
s.NoError(err)
|
|
return len(resp.GetExecutions()) == expectedNumber
|
|
},
|
|
testcore.WaitForESToSettle,
|
|
100*time.Millisecond,
|
|
"timeout expecting %d executions, found %d",
|
|
expectedNumber,
|
|
len(resp.GetExecutions()),
|
|
)
|
|
return resp.GetExecutions()
|
|
}
|
|
|
|
func (s *CronTestClientSuite) listClosedWorkflowExecutions(env *testcore.TestEnv, start, end time.Time, id string, expectedNumber int) []*workflowpb.WorkflowExecutionInfo {
|
|
s.T().Helper()
|
|
var resp *workflowservice.ListClosedWorkflowExecutionsResponse
|
|
s.Eventuallyf(
|
|
func() bool {
|
|
var err error
|
|
resp, err = env.SdkClient().ListClosedWorkflow(
|
|
s.Context(),
|
|
&workflowservice.ListClosedWorkflowExecutionsRequest{
|
|
Namespace: env.Namespace().String(),
|
|
MaximumPageSize: int32(2 * expectedNumber),
|
|
StartTimeFilter: &filterpb.StartTimeFilter{
|
|
EarliestTime: timestamppb.New(start),
|
|
LatestTime: timestamppb.New(end),
|
|
},
|
|
Filters: &workflowservice.ListClosedWorkflowExecutionsRequest_ExecutionFilter{
|
|
ExecutionFilter: &filterpb.WorkflowExecutionFilter{
|
|
WorkflowId: id,
|
|
},
|
|
},
|
|
},
|
|
)
|
|
s.NoError(err)
|
|
return len(resp.GetExecutions()) == expectedNumber
|
|
},
|
|
testcore.WaitForESToSettle,
|
|
100*time.Millisecond,
|
|
"timeout expecting %d executions, found %d",
|
|
expectedNumber,
|
|
len(resp.GetExecutions()),
|
|
)
|
|
return resp.GetExecutions()
|
|
}
|