478
}
479
481
>
now := time.Now().Truncate(common.ScheduledTaskMinPrecision)
482
>
scheduledTasks := []tasks.Task{
483
>
tasks.NewFakeTask(
484
>
s.WorkflowKey,
485
>
fakeScheduledTaskCategory,
486
>
now.Add(time.Nanosecond*10),
487
>
),
488
>
tasks.NewFakeTask(
489
>
s.WorkflowKey,
490
>
fakeScheduledTaskCategory,
491
>
now.Add(time.Nanosecond*20),
492
>
),
493
>
}
494
>
scheduledTasks[0].SetTaskID(100)
495
>
scheduledTasks[1].SetTaskID(50)
496
>
497
>
err := s.ExecutionManager.AddHistoryTasks(s.Ctx, &p.AddHistoryTasksRequest{
498
>
ShardID: s.ShardID,
499
>
RangeID: s.RangeID,
500
>
NamespaceID: s.WorkflowKey.NamespaceID,
501
>
WorkflowID: s.WorkflowKey.WorkflowID,
502
>
ArchetypeID: chasm.WorkflowArchetypeID,
503
>
Tasks: map[tasks.Category][]tasks.Task{
504
>
fakeScheduledTaskCategory: scheduledTasks,
505
>
},
506
>
})
507
>
s.NoError(err)
508
>
509
>
// due to persistence layer precision loss,
510
>
// two tasks can be returned in either order,
511
>
// but must be ordered in terms of tasks.Key
512
>
loadedTasks := s.PaginateTasks(
513
>
fakeScheduledTaskCategory,
514
>
tasks.NewKey(now, 0),
515
>
tasks.NewKey(now.Add(time.Second), 0),
516
>
10,
517
>
)
518
>
s.Len(loadedTasks, 2)
519
>
s.Negative(loadedTasks[0].GetKey().CompareTo(loadedTasks[1].GetKey()))
520
>
521
>
err = s.ExecutionManager.RangeCompleteHistoryTasks(s.Ctx, &p.RangeCompleteHistoryTasksRequest{
522
>
ShardID: s.ShardID,
523
>
TaskCategory: fakeScheduledTaskCategory,
524
>
InclusiveMinTaskKey: tasks.NewKey(now, 0),
525
>
ExclusiveMaxTaskKey: tasks.NewKey(now.Add(time.Second), 0),
526
>
})
527
>
s.NoError(err)
528
>
529
>
response, err := s.ExecutionManager.GetHistoryTasks(s.Ctx, &p.GetHistoryTasksRequest{
530
>
ShardID: s.ShardID,
531
>
TaskCategory: fakeScheduledTaskCategory,
532
>
InclusiveMinTaskKey: tasks.NewKey(now, 0),
533
>
ExclusiveMaxTaskKey: tasks.NewKey(now.Add(time.Second), 0),
534
>
BatchSize: 10,
535
>
})
536
>
s.NoError(err)
537
>
s.Empty(response.Tasks)
538
>
}
539
540
func (s *ExecutionMutableStateTaskSuite) AddRandomTasks(