go.temporal.io/server/tests/cron_test.go
642 LOC · 0 covered · 642 uncovered · 0 ranges · 0 concepts · 0 introducers · 0 tests
1
package tests
2
3
import (
4
"errors"
5
"fmt"
6
"sort"
7
"testing"
8
"time"
9
10
"github.com/google/uuid"
11
commandpb "go.temporal.io/api/command/v1"
12
commonpb "go.temporal.io/api/common/v1"
13
enumspb "go.temporal.io/api/enums/v1"
14
filterpb "go.temporal.io/api/filter/v1"
15
taskqueuepb "go.temporal.io/api/taskqueue/v1"
16
workflowpb "go.temporal.io/api/workflow/v1"
17
"go.temporal.io/api/workflowservice/v1"
18
sdkclient "go.temporal.io/sdk/client"
19
"go.temporal.io/sdk/workflow"
20
"go.temporal.io/server/common"
21
"go.temporal.io/server/common/failure"
22
"go.temporal.io/server/common/log/tag"
23
"go.temporal.io/server/common/payload"
24
"go.temporal.io/server/common/payloads"
25
"go.temporal.io/server/common/searchattribute/sadefs"
26
"go.temporal.io/server/common/testing/parallelsuite"
27
"go.temporal.io/server/tests/testcore"
28
"google.golang.org/protobuf/types/known/durationpb"
29
"google.golang.org/protobuf/types/known/timestamppb"
30
)
31
32
type CronTestSuite struct {
33
parallelsuite.Suite[*CronTestSuite]
34
}
35
36
type CronTestClientSuite struct {
37
parallelsuite.Suite[*CronTestClientSuite]
38
}
39
40
func TestCronTestSuite(t *testing.T) {
41
parallelsuite.Run(t, &CronTestSuite{})
42
}
43
44
func TestCronTestClientSuite(t *testing.T) {
45
parallelsuite.Run(t, &CronTestClientSuite{})
46
}
47
48
func (s *CronTestSuite) TestCronWorkflow_Failed_Infinite() {
49
env := testcore.NewEnv(s.T())
50
51
id := "functional-wf-cron-failed-infinite-test"
52
wt := "functional-wf-cron-failed-infinite-type"
53
tl := "functional-wf-cron-failed-infinite-taskqueue"
54
identity := "worker1"
55
cronSchedule := "@every 5s"
56
57
request := &workflowservice.StartWorkflowExecutionRequest{
58
RequestId: uuid.NewString(),
59
Namespace: env.Namespace().String(),
60
WorkflowId: id,
61
WorkflowType: &commonpb.WorkflowType{Name: wt},
62
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
63
Input: nil,
64
WorkflowRunTimeout: durationpb.New(5 * time.Second),
65
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
66
Identity: identity,
67
CronSchedule: cronSchedule, // minimum interval by standard spec is 1m (* * * * *, use non-standard descriptor for short interval for test
68
RetryPolicy: &commonpb.RetryPolicy{
69
MaximumAttempts: 2,
70
MaximumInterval: durationpb.New(1 * time.Second),
71
BackoffCoefficient: 1.2,
72
},
73
}
74
75
we, err0 := env.FrontendClient().StartWorkflowExecution(s.Context(), request)
76
s.NoError(err0)
77
78
env.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
79
80
respondFailed := false
81
seeRetry := false
82
wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
83
84
if !respondFailed {
85
respondFailed = true
86
87
return []*commandpb.Command{
88
{
89
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
90
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{
91
FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
92
Failure: failure.NewServerFailure("cron error for retry", false),
93
}},
94
}}, nil
95
}
96
97
startEvent := task.History.Events[0]
98
seeRetry = startEvent.GetWorkflowExecutionStartedEventAttributes().Initiator == enumspb.CONTINUE_AS_NEW_INITIATOR_RETRY
99
return []*commandpb.Command{
100
{
101
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
102
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{
103
CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
104
Result: nil,
105
}},
106
}}, nil
107
}
108
109
poller := &testcore.TaskPoller{
110
Client: env.FrontendClient(),
111
Namespace: env.Namespace().String(),
112
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
113
Identity: identity,
114
WorkflowTaskHandler: wtHandler,
115
Logger: env.Logger,
116
T: s.T(),
117
}
118
119
env.Logger.Info("Process first cron run which fails")
120
_, err := poller.PollAndProcessWorkflowTask(testcore.WithDumpHistory)
121
s.NoError(err)
122
123
env.Logger.Info("Process first cron run which completes")
124
_, err = poller.PollAndProcessWorkflowTask(testcore.WithDumpHistory)
125
s.NoError(err)
126
127
s.True(seeRetry)
128
}
129
130
func (s *CronTestSuite) TestCronWorkflow() {
131
env := testcore.NewEnv(s.T())
132
133
id := "functional-wf-cron-test"
134
wt := "functional-wf-cron-type"
135
tl := "functional-wf-cron-taskqueue"
136
identity := "worker1"
137
cronSchedule := "@every 3s"
138
139
targetBackoffDuration := time.Second * 3
140
backoffDurationTolerance := time.Millisecond * 500
141
142
memo := &commonpb.Memo{
143
Fields: map[string]*commonpb.Payload{"memoKey": payload.EncodeString("memoVal")},
144
}
145
searchAttr := &commonpb.SearchAttributes{
146
IndexedFields: map[string]*commonpb.Payload{
147
"CustomKeywordField": sadefs.MustEncodeValue("keyword-value", enumspb.INDEXED_VALUE_TYPE_KEYWORD),
148
},
149
}
150
151
request := &workflowservice.StartWorkflowExecutionRequest{
152
RequestId: uuid.NewString(),
153
Namespace: env.Namespace().String(),
154
WorkflowId: id,
155
WorkflowType: &commonpb.WorkflowType{Name: wt},
156
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
157
Input: nil,
158
WorkflowRunTimeout: durationpb.New(100 * time.Second),
159
WorkflowTaskTimeout: durationpb.New(1 * time.Second),
160
Identity: identity,
161
CronSchedule: cronSchedule, // minimum interval by standard spec is 1m (* * * * *, use non-standard descriptor for short interval for test
162
Memo: memo,
163
SearchAttributes: searchAttr,
164
}
165
166
// Because of rounding in GetBackoffForNextSchedule, we'll tend to stay aligned to whatever
167
// phase we start in relative to second boundaries, but drift slightly later within the second
168
// over time. If we cross a second boundary, one of our intervals will end up being 2s instead
169
// of 3s. To avoid this, wait until we can start early in the second.
170
for time.Now().Nanosecond()/int(time.Millisecond) > 150 {
171
time.Sleep(50 * time.Millisecond) //nolint:forbidigo
172
}
173
174
startWorkflowTS := time.Now().UTC()
175
we, err0 := env.FrontendClient().StartWorkflowExecution(s.Context(), request)
176
s.NoError(err0)
177
178
env.Logger.Info("StartWorkflowExecution", tag.WorkflowRunID(we.RunId))
179
180
var executions []*commonpb.WorkflowExecution
181
182
wtHandler := func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error) {
183
if task.PreviousStartedEventId == common.EmptyEventID {
184
startedEvent := task.History.Events[0]
185
if startedEvent.GetEventType() != enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED {
186
return []*commandpb.Command{
187
{
188
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
189
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
190
Failure: failure.NewServerFailure("incorrect first event", true),
191
}},
192
}}, nil
193
}
194
195
// Just check that it can be decoded
196
var lcr int
197
s.NoError(payloads.Decode(startedEvent.GetWorkflowExecutionStartedEventAttributes().GetLastCompletionResult(), &lcr))
198
}
199
200
executions = append(executions, task.WorkflowExecution)
201
if len(executions) >= 3 {
202
return []*commandpb.Command{
203
{
204
CommandType: enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION,
205
Attributes: &commandpb.Command_CompleteWorkflowExecutionCommandAttributes{CompleteWorkflowExecutionCommandAttributes: &commandpb.CompleteWorkflowExecutionCommandAttributes{
206
Result: payloads.EncodeString("cron-test-result"),
207
}},
208
}}, nil
209
}
210
return []*commandpb.Command{
211
{
212
CommandType: enumspb.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION,
213
Attributes: &commandpb.Command_FailWorkflowExecutionCommandAttributes{FailWorkflowExecutionCommandAttributes: &commandpb.FailWorkflowExecutionCommandAttributes{
214
Failure: failure.NewServerFailure("cron-test-error", false),
215
}},
216
}}, nil
217
}
218
219
poller := &testcore.TaskPoller{
220
Client: env.FrontendClient(),
221
Namespace: env.Namespace().String(),
222
TaskQueue: &taskqueuepb.TaskQueue{Name: tl, Kind: enumspb.TASK_QUEUE_KIND_NORMAL},
223
Identity: identity,
224
WorkflowTaskHandler: wtHandler,
225
Logger: env.Logger,
226
T: s.T(),
227
}
228
229
startFilter := &filterpb.StartTimeFilter{}
230
startFilter.EarliestTime = timestamppb.New(startWorkflowTS)
231
startFilter.LatestTime = timestamppb.New(time.Now().UTC())
232
233
// Sleep some time before checking the open executions.
234
// This will not cost extra time as the polling for first workflow task will be blocked for 3 seconds.
235
time.Sleep(2 * time.Second) //nolint:forbidigo
236
resp, err := env.FrontendClient().ListOpenWorkflowExecutions(s.Context(), &workflowservice.ListOpenWorkflowExecutionsRequest{
237
Namespace: env.Namespace().String(),
238
MaximumPageSize: 100,
239
StartTimeFilter: startFilter,
240
Filters: &workflowservice.ListOpenWorkflowExecutionsRequest_ExecutionFilter{ExecutionFilter: &filterpb.WorkflowExecutionFilter{
241
WorkflowId: id,
242
}},
243
})
244
s.NoError(err)
245
s.Len(resp.GetExecutions(), 1)
246
executionInfo := resp.GetExecutions()[0]
247
s.Equal(targetBackoffDuration, executionInfo.GetExecutionTime().AsTime().Sub(executionInfo.GetStartTime().AsTime()))
248
249
_, err = poller.PollAndProcessWorkflowTask()
250
s.NoError(err)
251
252
// Make sure the cron workflow start running at a proper time, in this case 3 seconds after the
253
// startWorkflowExecution request
254
backoffDuration := time.Now().UTC().Sub(startWorkflowTS)
255
s.Greater(backoffDuration, targetBackoffDuration)
256
s.Less(backoffDuration, targetBackoffDuration+backoffDurationTolerance)
257
258
_, err = poller.PollAndProcessWorkflowTask()
259
s.NoError(err)
260
261
_, err = poller.PollAndProcessWorkflowTask()
262
s.NoError(err)
263
264
s.Len(executions, 3)
265
266
_, terminateErr := env.FrontendClient().TerminateWorkflowExecution(s.Context(), &workflowservice.TerminateWorkflowExecutionRequest{
267
Namespace: env.Namespace().String(),
268
WorkflowExecution: &commonpb.WorkflowExecution{
269
WorkflowId: id,
270
},
271
})
272
s.NoError(terminateErr)
273
274
// first two should be failures
275
for i := range 2 {
276
events := env.GetHistory(env.Namespace().String(), executions[i])
277
s.EqualHistoryEvents(fmt.Sprintf(`
278
1 WorkflowExecutionStarted {"Memo":{"Fields":{"memoKey":{"Data":"\"memoVal\""}}},"SearchAttributes":{"IndexedFields":{"CustomKeywordField":{"Data":"\"keyword-value\"","Metadata":{"type":"Keyword"}}}}}
279
2 WorkflowTaskScheduled
280
3 WorkflowTaskStarted
281
4 WorkflowTaskCompleted
282
5 WorkflowExecutionFailed {"Failure":{"Message":"cron-test-error"},"NewExecutionRunId":"%s"}
283
`, executions[i+1].RunId), events)
284
}
285
286
// third should be completed
287
events := env.GetHistory(env.Namespace().String(), executions[2])
288
s.EqualHistoryEvents(`
289
1 WorkflowExecutionStarted {"Memo":{"Fields":{"memoKey":{"Data":"\"memoVal\""}}},"SearchAttributes":{"IndexedFields":{"CustomKeywordField":{"Data":"\"keyword-value\"","Metadata":{"type":"Keyword"}}}}}
290
2 WorkflowTaskScheduled
291
3 WorkflowTaskStarted
292
4 WorkflowTaskCompleted
293
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"cron-test-result\""}]}}
294
`, events)
295
296
startFilter.LatestTime = timestamppb.New(time.Now().UTC())
297
var closedExecutions []*workflowpb.WorkflowExecutionInfo
298
for range 10 {
299
resp, err := env.FrontendClient().ListClosedWorkflowExecutions(s.Context(), &workflowservice.ListClosedWorkflowExecutionsRequest{
300
Namespace: env.Namespace().String(),
301
MaximumPageSize: 100,
302
StartTimeFilter: startFilter,
303
Filters: &workflowservice.ListClosedWorkflowExecutionsRequest_ExecutionFilter{ExecutionFilter: &filterpb.WorkflowExecutionFilter{
304
WorkflowId: id,
305
}},
306
})
307
s.NoError(err)
308
if len(resp.GetExecutions()) == 4 {
309
closedExecutions = resp.GetExecutions()
310
break
311
}
312
time.Sleep(200 * time.Millisecond) //nolint:forbidigo
313
}
314
s.NotNil(closedExecutions)
315
dweResponse, err := env.FrontendClient().DescribeWorkflowExecution(s.Context(), &workflowservice.DescribeWorkflowExecutionRequest{
316
Namespace: env.Namespace().String(),
317
Execution: &commonpb.WorkflowExecution{
318
WorkflowId: id,
319
RunId: we.RunId,
320
},
321
})
322
s.NoError(err)
323
expectedExecutionTime := dweResponse.WorkflowExecutionInfo.GetStartTime().AsTime().Add(3 * time.Second)
324
s.Equal(expectedExecutionTime, dweResponse.WorkflowExecutionInfo.GetExecutionTime().AsTime())
325
326
sort.Slice(closedExecutions, func(i, j int) bool {
327
return closedExecutions[i].GetStartTime().AsTime().Before(closedExecutions[j].GetStartTime().AsTime())
328
})
329
lastExecution := closedExecutions[0]
330
for i := 1; i < 4; i++ {
331
executionInfo := closedExecutions[i]
332
expectedBackoff := executionInfo.GetExecutionTime().AsTime().Sub(lastExecution.GetExecutionTime().AsTime())
333
// The execution time calculated based on last execution close time.
334
// However, the current execution time is based on the current start time.
335
// This code is to remove the diff between current start time and last execution close time.
336
// TODO: Remove this line once we unify the time source
337
executionTimeDiff := executionInfo.GetStartTime().AsTime().Sub(lastExecution.GetCloseTime().AsTime())
338
// The backoff between any two executions should be a multiplier of the target backoff duration which is 3 in this test
339
s.Equal(
340
0,
341
int((expectedBackoff-executionTimeDiff).Round(time.Second).Seconds())%int(targetBackoffDuration.Seconds()),
342
"expected backoff %v-%v=%v should be multiplier of target backoff %v",
343
expectedBackoff.Seconds(),
344
executionTimeDiff.Seconds(),
345
(expectedBackoff - executionTimeDiff).Round(time.Second).Seconds(),
346
targetBackoffDuration.Seconds())
347
lastExecution = executionInfo
348
349
// TODO: Remove the describeWorkflowExecution call when firstRunID in WorkflowExecutionInfo
350
// is populated by Visibility api as well.
351
dweResponse, err := env.FrontendClient().DescribeWorkflowExecution(s.Context(), &workflowservice.DescribeWorkflowExecutionRequest{
352
Namespace: env.Namespace().String(),
353
Execution: executionInfo.GetExecution(),
354
})
355
s.NoError(err)
356
s.Equal(we.RunId, dweResponse.WorkflowExecutionInfo.GetFirstRunId())
357
}
358
}
359
360
func (s *CronTestClientSuite) TestCronWorkflowCompletionStates() {
361
// Run a cron workflow that completes in (almost) all the possible ways:
362
// Run 1: succeeds
363
// Run 2: fails
364
// Run 3: times out
365
// Run 4: succeeds
366
// Run 5: succeeds
367
// Run 6: terminated before it runs
368
369
// Continue-as-new is not tested (behavior is currently not correct)
370
371
env := testcore.NewEnv(s.T())
372
373
id := "functional-wf-cron-failed-test"
374
cronSchedule := "@every 3s"
375
376
targetBackoffDuration := 3 * time.Second
377
workflowRunTimeout := 5 * time.Second
378
tolerance := 500 * time.Millisecond
379
durationNear := func(value, target time.Duration) {
380
s.T().Helper()
381
s.Greater(value, target-tolerance)
382
s.Less(value, target+tolerance)
383
}
384
385
runIDs := make(map[string]bool)
386
wfCh := make(chan int)
387
388
workflowFn := func(ctx workflow.Context) (string, error) {
389
runIDs[workflow.GetInfo(ctx).WorkflowExecution.RunID] = true
390
iteration := len(runIDs)
391
wfCh <- iteration
392
393
var lcr string
394
switch iteration {
395
case 1:
396
s.False(workflow.HasLastCompletionResult(ctx))
397
s.NoError(workflow.GetLastError(ctx))
398
return "pass", nil
399
400
case 2:
401
s.True(workflow.HasLastCompletionResult(ctx))
402
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
403
s.Equal("pass", lcr)
404
s.NoError(workflow.GetLastError(ctx))
405
return "", errors.New("second error") //nolint:err113
406
407
case 3:
408
s.True(workflow.HasLastCompletionResult(ctx))
409
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
410
s.Equal("pass", lcr)
411
s.Error(workflow.GetLastError(ctx))
412
s.Equal("second error", workflow.GetLastError(ctx).Error())
413
s.NoError(workflow.Sleep(ctx, 10*time.Second)) // cause wft timeout
414
panic("should have been timed out on server already")
415
416
case 4:
417
s.True(workflow.HasLastCompletionResult(ctx))
418
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
419
s.Equal("pass", lcr)
420
s.Error(workflow.GetLastError(ctx))
421
s.Equal("workflow timeout (type: StartToClose)", workflow.GetLastError(ctx).Error())
422
return "pass again", nil
423
424
case 5:
425
s.True(workflow.HasLastCompletionResult(ctx))
426
s.NoError(workflow.GetLastCompletionResult(ctx, &lcr))
427
s.Equal("pass again", lcr)
428
s.NoError(workflow.GetLastError(ctx))
429
return "final pass", nil
430
}
431
432
panic("shouldn't get here")
433
}
434
435
env.SdkWorker().RegisterWorkflow(workflowFn)
436
437
// Because of rounding in GetBackoffForNextSchedule, we'll tend to stay aligned to whatever
438
// phase we start in relative to second boundaries, but drift slightly later within the second
439
// over time. If we cross a second boundary, one of our intervals will end up being 2s instead
440
// of 3s. To avoid this, wait until we can start early in the second.
441
for time.Now().Nanosecond()/int(time.Millisecond) > 150 {
442
time.Sleep(50 * time.Millisecond) //nolint:forbidigo
443
}
444
445
workflowOptions := sdkclient.StartWorkflowOptions{
446
ID: id,
447
TaskQueue: env.WorkerTaskQueue(),
448
WorkflowRunTimeout: workflowRunTimeout,
449
CronSchedule: cronSchedule,
450
}
451
ts := time.Now()
452
startTs := ts
453
_, err := env.SdkClient().ExecuteWorkflow(s.Context(), workflowOptions, workflowFn)
454
s.NoError(err)
455
456
// check execution and history of first run
457
exec := s.listOpenWorkflowExecutions(env, startTs, time.Now(), id, 1)[0]
458
firstRunID := exec.GetExecution().RunId
459
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, exec.GetStatus())
460
historyEvents := env.GetHistory(env.Namespace().String(), exec.GetExecution())
461
s.EqualHistoryEvents(fmt.Sprintf(`
462
1 WorkflowExecutionStarted {"ContinuedExecutionRunId":"","CronSchedule":"@every 3s","FirstExecutionRunId":"%s", "Initiator":3}`, firstRunID), historyEvents)
463
attrs1 := historyEvents[0].GetWorkflowExecutionStartedEventAttributes()
464
// not `"FirstWorkflowTaskBackoff":{"Nanos":0,"Seconds":3}` in the history above because DurationNear is not supported by EqualHistoryEvents.
465
durationNear(attrs1.FirstWorkflowTaskBackoff.AsDuration(), targetBackoffDuration)
466
467
// wait for first run
468
s.Equal(1, <-wfCh)
469
durationNear(time.Since(ts), targetBackoffDuration)
470
ts = time.Now()
471
472
// let first run finish, then check execution and history of second run
473
s.Eventually(
474
func() bool {
475
exec = s.listOpenWorkflowExecutions(env, startTs, time.Now(), id, 1)[0]
476
return exec.GetExecution().GetRunId() != firstRunID
477
},
478
targetBackoffDuration+tolerance,
479
250*time.Millisecond,
480
)
481
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING, exec.GetStatus())
482
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
483
s.EqualHistoryEvents(fmt.Sprintf(`
484
1 WorkflowExecutionStarted {"ContinuedExecutionRunId":"%s","CronSchedule":"@every 3s","FirstExecutionRunId":"%s", "Initiator":%d}`, firstRunID, firstRunID, enumspb.CONTINUE_AS_NEW_INITIATOR_CRON_SCHEDULE), historyEvents)
485
attrs2 := historyEvents[0].GetWorkflowExecutionStartedEventAttributes()
486
// not `"FirstWorkflowTaskBackoff":{"Nanos":0,"Seconds":3}` in the history above because DurationNear is not supported by EqualHistoryEvents.
487
durationNear(attrs2.FirstWorkflowTaskBackoff.AsDuration(), targetBackoffDuration)
488
489
// wait for second run
490
s.Equal(2, <-wfCh)
491
durationNear(time.Since(ts), targetBackoffDuration)
492
ts = time.Now()
493
494
// don't bother checking started events for subsequent runs, we covered the important parts already
495
496
// wait for third run
497
s.Equal(3, <-wfCh)
498
durationNear(time.Since(ts), targetBackoffDuration)
499
ts = time.Now()
500
501
// wait for fourth run (third one waits for timeout after 5s, so will run after 6s)
502
s.Equal(4, <-wfCh)
503
durationNear(time.Since(ts), 2*targetBackoffDuration)
504
ts = time.Now()
505
506
// wait for fifth run
507
s.Equal(5, <-wfCh)
508
durationNear(time.Since(ts), targetBackoffDuration)
509
510
// let fifth run finish and sixth get scheduled
511
_ = s.listClosedWorkflowExecutions(env, startTs, time.Now().Add(targetBackoffDuration), id, 5)
512
_ = s.listOpenWorkflowExecutions(env, startTs, time.Now().Add(targetBackoffDuration), id, 1)
513
// then terminate
514
s.NoError(env.SdkClient().TerminateWorkflow(s.Context(), id, "", "test is over"))
515
516
closedExecutions := s.listClosedWorkflowExecutions(env, startTs, time.Now(), id, 6)
517
518
exec = closedExecutions[5] // first: success
519
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, exec.GetStatus())
520
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
521
s.EqualHistoryEvents(`
522
1 WorkflowExecutionStarted
523
2 WorkflowTaskScheduled
524
3 WorkflowTaskStarted
525
4 WorkflowTaskCompleted
526
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"pass\""}]}}`, historyEvents)
527
528
exec = closedExecutions[4] // second: fail
529
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_FAILED, exec.GetStatus())
530
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
531
s.EqualHistoryEvents(`
532
1 WorkflowExecutionStarted
533
2 WorkflowTaskScheduled
534
3 WorkflowTaskStarted
535
4 WorkflowTaskCompleted
536
5 WorkflowExecutionFailed {"Failure":{"Message":"second error"}}`, historyEvents)
537
538
exec = closedExecutions[3] // third: timed out
539
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT, exec.GetStatus())
540
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
541
s.EqualHistoryEvents(`
542
1 WorkflowExecutionStarted
543
2 WorkflowTaskScheduled
544
3 WorkflowTaskStarted
545
4 WorkflowTaskCompleted
546
5 TimerStarted
547
6 WorkflowExecutionTimedOut {"RetryState":5} // enumspb.RETRY_STATE_RETRY_POLICY_NOT_SET`, historyEvents)
548
549
exec = closedExecutions[2] // fourth: success
550
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, exec.GetStatus())
551
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
552
s.EqualHistoryEvents(`
553
1 WorkflowExecutionStarted
554
2 WorkflowTaskScheduled
555
3 WorkflowTaskStarted
556
4 WorkflowTaskCompleted
557
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"pass again\""}]}}`, historyEvents)
558
559
exec = closedExecutions[1] // fifth: success
560
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, exec.GetStatus())
561
historyEvents = env.GetHistory(env.Namespace().String(), exec.GetExecution())
562
s.EqualHistoryEvents(`
563
1 WorkflowExecutionStarted
564
2 WorkflowTaskScheduled
565
3 WorkflowTaskStarted
566
4 WorkflowTaskCompleted
567
5 WorkflowExecutionCompleted {"Result":{"Payloads":[{"Data":"\"final pass\""}]}}`, historyEvents)
568
569
exec = closedExecutions[0] // sixth: terminated
570
s.Equal(enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, exec.GetStatus())
571
events := env.GetHistory(env.Namespace().String(), exec.GetExecution())
572
s.EqualHistoryEvents(`
573
1 WorkflowExecutionStarted
574
2 WorkflowExecutionTerminated {"Reason":"test is over"}`, events)
575
}
576
577
func (s *CronTestClientSuite) listOpenWorkflowExecutions(env *testcore.TestEnv, start, end time.Time, id string, expectedNumber int) []*workflowpb.WorkflowExecutionInfo {
578
s.T().Helper()
579
var resp *workflowservice.ListOpenWorkflowExecutionsResponse
580
s.Eventuallyf(
581
func() bool {
582
var err error
583
resp, err = env.SdkClient().ListOpenWorkflow(
584
s.Context(), &workflowservice.ListOpenWorkflowExecutionsRequest{
585
Namespace: env.Namespace().String(),
586
MaximumPageSize: int32(2 * expectedNumber),
587
StartTimeFilter: &filterpb.StartTimeFilter{
588
EarliestTime: timestamppb.New(start),
589
LatestTime: timestamppb.New(end),
590
},
591
Filters: &workflowservice.ListOpenWorkflowExecutionsRequest_ExecutionFilter{
592
ExecutionFilter: &filterpb.WorkflowExecutionFilter{
593
WorkflowId: id,
594
},
595
},
596
},
597
)
598
s.NoError(err)
599
return len(resp.GetExecutions()) == expectedNumber
600
},
601
testcore.WaitForESToSettle,
602
100*time.Millisecond,
603
"timeout expecting %d executions, found %d",
604
expectedNumber,
605
len(resp.GetExecutions()),
606
)
607
return resp.GetExecutions()
608
}
609
610
func (s *CronTestClientSuite) listClosedWorkflowExecutions(env *testcore.TestEnv, start, end time.Time, id string, expectedNumber int) []*workflowpb.WorkflowExecutionInfo {
611
s.T().Helper()
612
var resp *workflowservice.ListClosedWorkflowExecutionsResponse
613
s.Eventuallyf(
614
func() bool {
615
var err error
616
resp, err = env.SdkClient().ListClosedWorkflow(
617
s.Context(),
618
&workflowservice.ListClosedWorkflowExecutionsRequest{
619
Namespace: env.Namespace().String(),
620
MaximumPageSize: int32(2 * expectedNumber),
621
StartTimeFilter: &filterpb.StartTimeFilter{
622
EarliestTime: timestamppb.New(start),
623
LatestTime: timestamppb.New(end),
624
},
625
Filters: &workflowservice.ListClosedWorkflowExecutionsRequest_ExecutionFilter{
626
ExecutionFilter: &filterpb.WorkflowExecutionFilter{
627
WorkflowId: id,
628
},
629
},
630
},
631
)
632
s.NoError(err)
633
return len(resp.GetExecutions()) == expectedNumber
634
},
635
testcore.WaitForESToSettle,
636
100*time.Millisecond,
637
"timeout expecting %d executions, found %d",
638
expectedNumber,
639
len(resp.GetExecutions()),
640
)
641
return resp.GetExecutions()
642
}