mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Tim Deeb-SwihartandGitHub
95169a53c3
**What changed?** gogo/protobuf has been replaced with Google's official go compiler. **Why?** gogo/protobuf has been deprecated for some time and the community is moving on, building new tools (like vtproto) atop google's v2 compiler. **How did you test it?** `make test` **Potential risks** 1. The change from embedded gogo-generated-structs to google-generated-pointers-to-structs created a risk of nil pointer exceptions. I've fixed all the ones our tests found but it's possible there are more lurking in the new code. 2. This change may cause our performance to decrease. Certainly encoding/deconding of proto objects will become slower, but the overuse of pointers by the google compiler may negatively affect our overall performance. We'll need to keep an eye on the GC stats 3. This breaks the HTTP API. We will not support [shortand payload encoding](https://github.com/temporalio/proposals/blob/master/api/http-api.md#payload-formatting) in this first pass; that will come once this initial work is in testing. **Breaking changes for developers** - `*time.Time` in proto structs will now be [timestamppb.Timestamp](https://pkg.go.dev/google.golang.org/[email protected]/types/known/timestamppb#section-documentation) - `*time.Duration` will now be [durationpb.Duration](https://pkg.go.dev/google.golang.org/protobuf/types/known/durationpb) - V2-generated structs embed locks, so you cannot dereference them. `go vet` will scream at you about this. If you need a copy, use `proto.Clone`. - If the performance of this sucks then I will either update our code generator to add shallow-clone methods or hand-roll the ones we need - Proto enums will, when formatted to JSON, now be in `SCREAMING_SNAKE_CASE` rather than `PascalCase`. We decided (in discussion with the SDK team) that now was as good a time as any to rip the bandage off. - Proto objects, or objects embedding protos, cannot be compared using `reflect.DeepEqual` or _anything_ that uses it. This includes `testify` and `mock` equality testers! - You will need to use the `common/testing/protorequire`, `common/testing/protoassert`, or `common/testing/protomock` packages instead. I've implemented proto-compatible matchers and assertions there for all cases I've encountered - If you need `reflect.DeepEqual` for any reason you can use `go.temporal.io/api/temporalproto.DeepEqual` instead Note that history loading will not be impacted by the JSON changes: I rewrote history loading to dynamically fix incoming history JSON data (like all our other sdks); you can find this code in [my fork of our go API](https://github.com/tdeebswihart/temporal-api-go/blob/master/internal/temporalhistoryv1/load.go) alongside its tests. **🚨Sharp Edges Introduced🚨** Beware `*timestamppb.Timestamp.AsTime()`. If you need to extract a time value from a proto time (timestamppb) **always** make sure to check whether it's nil first. When the proto object is `nil` `AsTime()` will return a non-zero time at the proto epoch: UTC midnight on January 1, 1970. I've made this mistake multiple times during this transition and each time it's been a pain to debug **Is hotfix candidate?** No.
1592 lines
64 KiB
Go
1592 lines
64 KiB
Go
// The MIT License
|
|
//
|
|
// Copyright (c) 2020 Temporal Technologies Inc. All rights reserved.
|
|
//
|
|
// Copyright (c) 2020 Uber Technologies, Inc.
|
|
//
|
|
// Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
// of this software and associated documentation files (the "Software"), to deal
|
|
// in the Software without restriction, including without limitation the rights
|
|
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
// copies of the Software, and to permit persons to whom the Software is
|
|
// furnished to do so, subject to the following conditions:
|
|
//
|
|
// The above copyright notice and this permission notice shall be included in
|
|
// all copies or substantial portions of the Software.
|
|
//
|
|
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
// THE SOFTWARE.
|
|
|
|
package workflow
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/golang/mock/gomock"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/stretchr/testify/suite"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/common"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/primitives/timestamp"
|
|
"go.temporal.io/server/common/testing/protomock"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
"go.temporal.io/server/service/history/tests"
|
|
)
|
|
|
|
type (
|
|
timerSequenceSuite struct {
|
|
suite.Suite
|
|
*require.Assertions
|
|
|
|
controller *gomock.Controller
|
|
mockMutableState *MockMutableState
|
|
|
|
workflowKey definition.WorkflowKey
|
|
timerSequence *timerSequenceImpl
|
|
}
|
|
)
|
|
|
|
func TestTimerSequenceSuite(t *testing.T) {
|
|
s := new(timerSequenceSuite)
|
|
suite.Run(t, s)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) SetupSuite() {
|
|
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TearDownSuite() {
|
|
|
|
}
|
|
|
|
func (s *timerSequenceSuite) SetupTest() {
|
|
s.Assertions = require.New(s.T())
|
|
|
|
s.controller = gomock.NewController(s.T())
|
|
s.mockMutableState = NewMockMutableState(s.controller)
|
|
|
|
s.workflowKey = definition.NewWorkflowKey(
|
|
tests.NamespaceID.String(),
|
|
tests.WorkflowID,
|
|
tests.RunID,
|
|
)
|
|
s.mockMutableState.EXPECT().GetWorkflowKey().Return(s.workflowKey).AnyTimes()
|
|
s.timerSequence = NewTimerSequence(s.mockMutableState)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TearDownTest() {
|
|
s.controller.Finish()
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextUserTimer_AlreadyCreated_AfterWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
timerExpiry := timestamppb.New(now.Add(100))
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timerExpiry,
|
|
TaskStatus: TimerTaskStatusCreated,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{timerInfo.TimerId: timerInfo}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(timerExpiry.AsTime().Add(-1 * time.Second)),
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextUserTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextUserTimer_AlreadyCreated_BeforeWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
timerExpiry := timestamppb.New(now.Add(100))
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timerExpiry,
|
|
TaskStatus: TimerTaskStatusCreated,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{timerInfo.TimerId: timerInfo}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(timerExpiry.AsTime().Add(1 * time.Second)),
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextUserTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextUserTimer_AlreadyCreated_NoWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
timer1Expiry := timestamppb.New(now.Add(100))
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timer1Expiry,
|
|
TaskStatus: TimerTaskStatusCreated,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{timerInfo.TimerId: timerInfo}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: nil,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextUserTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextUserTimer_NotCreated_AfterWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
timerExpiry := timestamppb.New(now.Add(100))
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timerExpiry,
|
|
TaskStatus: TimerTaskStatusNone,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{timerInfo.TimerId: timerInfo}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(timerExpiry.AsTime().Add(-1 * time.Second)),
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextUserTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextUserTimer_NotCreated_BeforeWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
timerExpiry := timestamppb.New(now.Add(100))
|
|
currentVersion := int64(999)
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timerExpiry,
|
|
TaskStatus: TimerTaskStatusNone,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{timerInfo.TimerId: timerInfo}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
s.mockMutableState.EXPECT().GetUserTimerInfoByEventID(timerInfo.StartedEventId).Return(timerInfo, true)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(timerExpiry.AsTime().Add(1 * time.Second)),
|
|
})
|
|
|
|
var timerInfoUpdated = common.CloneProto(timerInfo) // make a copy
|
|
timerInfoUpdated.TaskStatus = TimerTaskStatusCreated
|
|
s.mockMutableState.EXPECT().UpdateUserTimer(protomock.Eq(timerInfoUpdated)).Return(nil)
|
|
s.mockMutableState.EXPECT().GetCurrentVersion().Return(currentVersion)
|
|
s.mockMutableState.EXPECT().AddTasks(&tasks.UserTimerTask{
|
|
// TaskID is set by shard
|
|
WorkflowKey: s.workflowKey,
|
|
VisibilityTimestamp: timerExpiry.AsTime(),
|
|
EventID: timerInfo.GetStartedEventId(),
|
|
Version: currentVersion,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextUserTimer()
|
|
s.NoError(err)
|
|
s.True(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextUserTimer_NotCreated_NoWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
timerExpiry := timestamppb.New(now.Add(100))
|
|
currentVersion := int64(999)
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timerExpiry,
|
|
TaskStatus: TimerTaskStatusNone,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{timerInfo.TimerId: timerInfo}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
s.mockMutableState.EXPECT().GetUserTimerInfoByEventID(timerInfo.StartedEventId).Return(timerInfo, true)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: nil,
|
|
})
|
|
|
|
var timerInfoUpdated = common.CloneProto(timerInfo) // make a copy
|
|
timerInfoUpdated.TaskStatus = TimerTaskStatusCreated
|
|
s.mockMutableState.EXPECT().UpdateUserTimer(protomock.Eq(timerInfoUpdated)).Return(nil)
|
|
s.mockMutableState.EXPECT().GetCurrentVersion().Return(currentVersion)
|
|
s.mockMutableState.EXPECT().AddTasks(&tasks.UserTimerTask{
|
|
// TaskID is set by shard
|
|
WorkflowKey: s.workflowKey,
|
|
VisibilityTimestamp: timerExpiry.AsTime(),
|
|
EventID: timerInfo.GetStartedEventId(),
|
|
Version: currentVersion,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextUserTimer()
|
|
s.NoError(err)
|
|
s.True(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_AlreadyCreated_AfterWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(now.Add(-2000 * time.Second)),
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_AlreadyCreated_BeforeWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(now.Add(2000 * time.Second)),
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_AlreadyCreated_NoWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: nil,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_NotCreated_AfterWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(now.Add(-2000 * time.Second)),
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_NotCreated_BeforeWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
currentVersion := int64(999)
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetActivityInfo(activityInfo.ScheduledEventId).Return(activityInfo, true)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(now.Add(2000 * time.Second)),
|
|
})
|
|
|
|
var activityInfoUpdated = common.CloneProto(activityInfo) // make a copy
|
|
activityInfoUpdated.TimerTaskStatus = TimerTaskStatusCreatedScheduleToStart
|
|
s.mockMutableState.EXPECT().UpdateActivity(protomock.Eq(activityInfoUpdated)).Return(nil)
|
|
s.mockMutableState.EXPECT().GetCurrentVersion().Return(currentVersion)
|
|
s.mockMutableState.EXPECT().AddTasks(&tasks.ActivityTimeoutTask{
|
|
// TaskID is set by shard
|
|
WorkflowKey: s.workflowKey,
|
|
VisibilityTimestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToStartTimeout.AsDuration()),
|
|
TimeoutType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Attempt: activityInfo.Attempt,
|
|
Version: currentVersion,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.True(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_NotCreated_NoWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
currentVersion := int64(999)
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetActivityInfo(activityInfo.ScheduledEventId).Return(activityInfo, true)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: nil,
|
|
})
|
|
|
|
var activityInfoUpdated = common.CloneProto(activityInfo) // make a copy
|
|
activityInfoUpdated.TimerTaskStatus = TimerTaskStatusCreatedScheduleToStart
|
|
s.mockMutableState.EXPECT().UpdateActivity(protomock.Eq(activityInfoUpdated)).Return(nil)
|
|
s.mockMutableState.EXPECT().GetCurrentVersion().Return(currentVersion)
|
|
s.mockMutableState.EXPECT().AddTasks(&tasks.ActivityTimeoutTask{
|
|
// TaskID is set by shard
|
|
WorkflowKey: s.workflowKey,
|
|
VisibilityTimestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToStartTimeout.AsDuration()),
|
|
TimeoutType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Attempt: activityInfo.Attempt,
|
|
Version: currentVersion,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.True(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_HeartbeatTimer_AfterWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(now.Add(-2000 * time.Second)),
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.False(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_HeartbeatTimer_BeforeWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
currentVersion := int64(999)
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetActivityInfo(activityInfo.ScheduledEventId).Return(activityInfo, true)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: timestamppb.New(now.Add(2000 * time.Second)),
|
|
})
|
|
|
|
taskVisibilityTimestamp := activityInfo.StartedTime.AsTime().Add(activityInfo.HeartbeatTimeout.AsDuration())
|
|
|
|
var activityInfoUpdated = common.CloneProto(activityInfo) // make a copy
|
|
activityInfoUpdated.TimerTaskStatus = TimerTaskStatusCreatedHeartbeat
|
|
s.mockMutableState.EXPECT().UpdateActivityWithTimerHeartbeat(protomock.Eq(activityInfoUpdated), taskVisibilityTimestamp).Return(nil)
|
|
s.mockMutableState.EXPECT().GetCurrentVersion().Return(currentVersion)
|
|
s.mockMutableState.EXPECT().AddTasks(&tasks.ActivityTimeoutTask{
|
|
// TaskID is set by shard
|
|
WorkflowKey: s.workflowKey,
|
|
VisibilityTimestamp: taskVisibilityTimestamp,
|
|
TimeoutType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Attempt: activityInfo.Attempt,
|
|
Version: currentVersion,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.True(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestCreateNextActivityTimer_HeartbeatTimer_NoWorkflowExpiry() {
|
|
now := time.Now().UTC()
|
|
currentVersion := int64(999)
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
s.mockMutableState.EXPECT().GetActivityInfo(activityInfo.ScheduledEventId).Return(activityInfo, true)
|
|
s.mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
|
|
WorkflowRunExpirationTime: nil,
|
|
})
|
|
|
|
taskVisibilityTimestamp := activityInfo.StartedTime.AsTime().Add(activityInfo.HeartbeatTimeout.AsDuration())
|
|
|
|
var activityInfoUpdated = common.CloneProto(activityInfo) // make a copy
|
|
activityInfoUpdated.TimerTaskStatus = TimerTaskStatusCreatedHeartbeat
|
|
s.mockMutableState.EXPECT().UpdateActivityWithTimerHeartbeat(protomock.Eq(activityInfoUpdated), taskVisibilityTimestamp).Return(nil)
|
|
s.mockMutableState.EXPECT().GetCurrentVersion().Return(currentVersion)
|
|
s.mockMutableState.EXPECT().AddTasks(&tasks.ActivityTimeoutTask{
|
|
// TaskID is set by shard
|
|
WorkflowKey: s.workflowKey,
|
|
VisibilityTimestamp: taskVisibilityTimestamp,
|
|
TimeoutType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Attempt: activityInfo.Attempt,
|
|
Version: currentVersion,
|
|
})
|
|
|
|
modified, err := s.timerSequence.CreateNextActivityTimer()
|
|
s.NoError(err)
|
|
s.True(modified)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortUserTimers_None() {
|
|
timerInfos := map[string]*persistencespb.TimerInfo{}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortUserTimers()
|
|
s.Empty(timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortUserTimers_One() {
|
|
now := time.Now().UTC()
|
|
timer1Expiry := timestamppb.New(now.Add(100))
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timer1Expiry,
|
|
TaskStatus: TimerTaskStatusCreated,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{timerInfo.TimerId: timerInfo}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortUserTimers()
|
|
s.Equal([]TimerSequenceID{{
|
|
EventID: timerInfo.GetStartedEventId(),
|
|
Timestamp: timer1Expiry.AsTime(),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: 1,
|
|
}}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortUserTimers_Multiple() {
|
|
now := time.Now().UTC()
|
|
timer1Expiry := timestamppb.New(now.Add(100))
|
|
timer2Expiry := timestamppb.New(now.Add(200))
|
|
timerInfo1 := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timer1Expiry,
|
|
TaskStatus: TimerTaskStatusCreated,
|
|
}
|
|
timerInfo2 := &persistencespb.TimerInfo{
|
|
Version: 1234,
|
|
TimerId: "other random timer ID",
|
|
StartedEventId: 4567,
|
|
ExpiryTime: timestamppb.New(now.Add(200)),
|
|
TaskStatus: TimerTaskStatusNone,
|
|
}
|
|
timerInfos := map[string]*persistencespb.TimerInfo{
|
|
timerInfo1.TimerId: timerInfo1,
|
|
timerInfo2.TimerId: timerInfo2,
|
|
}
|
|
s.mockMutableState.EXPECT().GetPendingTimerInfos().Return(timerInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortUserTimers()
|
|
s.Equal([]TimerSequenceID{
|
|
{
|
|
EventID: timerInfo1.GetStartedEventId(),
|
|
Timestamp: timer1Expiry.AsTime(),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: 1,
|
|
},
|
|
{
|
|
EventID: timerInfo2.GetStartedEventId(),
|
|
Timestamp: timer2Expiry.AsTime(),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: false,
|
|
Attempt: 1,
|
|
},
|
|
}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_None() {
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.Empty(timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_One_NotScheduled() {
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: common.EmptyEventID,
|
|
ScheduledTime: nil,
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.Empty(timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_One_Scheduled_NotStarted() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.Equal([]TimerSequenceID{
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToStartTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_One_Scheduled_Started_WithHeartbeatTimeout() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedStartToClose | TimerTaskStatusCreatedHeartbeat,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.Equal([]TimerSequenceID{
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.StartedTime.AsTime().Add(activityInfo.HeartbeatTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.StartedTime.AsTime().Add(activityInfo.StartToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_One_Scheduled_Started_WithoutHeartbeatTimeout() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedStartToClose,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.Equal([]TimerSequenceID{
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.StartedTime.AsTime().Add(activityInfo.StartToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_One_Scheduled_Started_Heartbeated_WithHeartbeatTimeout() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedStartToClose | TimerTaskStatusCreatedHeartbeat,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.Equal([]TimerSequenceID{
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.LastHeartbeatUpdateTime.AsTime().Add(activityInfo.HeartbeatTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.StartedTime.AsTime().Add(activityInfo.StartToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_One_Scheduled_Started_Heartbeated_WithoutHeartbeatTimeout() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose | TimerTaskStatusCreatedStartToClose,
|
|
Attempt: 12,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{activityInfo.ScheduledEventId: activityInfo}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.EqualValues([]TimerSequenceID{
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.StartedTime.AsTime().Add(activityInfo.StartToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: activityInfo.Attempt,
|
|
},
|
|
}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLoadAndSortActivityTimers_Multiple() {
|
|
now := time.Now().UTC()
|
|
activityInfo1 := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
activityInfo2 := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 2345,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "other random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(11),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1001),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(101),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(6),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(800 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 21,
|
|
}
|
|
activityInfos := map[int64]*persistencespb.ActivityInfo{
|
|
activityInfo1.ScheduledEventId: activityInfo1,
|
|
activityInfo2.ScheduledEventId: activityInfo2,
|
|
}
|
|
s.mockMutableState.EXPECT().GetPendingActivityInfos().Return(activityInfos)
|
|
|
|
timerSequenceIDs := s.timerSequence.LoadAndSortActivityTimers()
|
|
s.Equal([]TimerSequenceID{
|
|
{
|
|
EventID: activityInfo2.ScheduledEventId,
|
|
Timestamp: activityInfo2.ScheduledTime.AsTime().Add(activityInfo2.ScheduleToStartTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
|
|
TimerCreated: false,
|
|
Attempt: activityInfo2.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo1.ScheduledEventId,
|
|
Timestamp: activityInfo1.StartedTime.AsTime().Add(activityInfo1.StartToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: false,
|
|
Attempt: activityInfo1.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo1.ScheduledEventId,
|
|
Timestamp: activityInfo1.ScheduledTime.AsTime().Add(activityInfo1.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: false,
|
|
Attempt: activityInfo1.Attempt,
|
|
},
|
|
{
|
|
EventID: activityInfo2.ScheduledEventId,
|
|
Timestamp: activityInfo2.ScheduledTime.AsTime().Add(activityInfo2.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: false,
|
|
Attempt: activityInfo2.Attempt,
|
|
},
|
|
}, timerSequenceIDs)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetUserTimerTimeout() {
|
|
now := time.Now().UTC()
|
|
timerExpiry := timestamppb.New(now.Add(100))
|
|
timerInfo := &persistencespb.TimerInfo{
|
|
Version: 123,
|
|
TimerId: "some random timer ID",
|
|
StartedEventId: 456,
|
|
ExpiryTime: timerExpiry,
|
|
TaskStatus: TimerTaskStatusCreated,
|
|
}
|
|
|
|
expectedTimerSequence := &TimerSequenceID{
|
|
EventID: timerInfo.StartedEventId,
|
|
Timestamp: timerExpiry.AsTime(),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: 1,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getUserTimerTimeout(timerInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
|
|
timerInfo.TaskStatus = TimerTaskStatusNone
|
|
expectedTimerSequence.TimerCreated = false
|
|
timerSequence = s.timerSequence.getUserTimerTimeout(timerInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToStartTimeout_WithTimeout_NotScheduled() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: common.EmptyEventID,
|
|
ScheduledTime: nil,
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToStartTimeout_WithTimeout_Scheduled_NotStarted() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
|
|
expectedTimerSequence := &TimerSequenceID{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToStartTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
expectedTimerSequence.TimerCreated = false
|
|
timerSequence = s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToStartTimeout_WithTimeout_Scheduled_Started() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Second)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
timerSequence = s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToStartTimeout_WithoutTimeout_NotScheduled() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: common.EmptyEventID,
|
|
ScheduledTime: nil,
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(0),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToStartTimeout_WithoutTimeout_Scheduled_NotStarted() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(0),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
timerSequence = s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToStartTimeout_WithoutTimeout_Scheduled_Started() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Second)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(0),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToStart,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
timerSequence = s.timerSequence.getActivityScheduleToStartTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToCloseTimeout_WithTimeout_NotScheduled() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: common.EmptyEventID,
|
|
ScheduledTime: nil,
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToCloseTimeout_WithTimeout_Scheduled() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose,
|
|
Attempt: 12,
|
|
}
|
|
|
|
expectedTimerSequence := &TimerSequenceID{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.ScheduledTime.AsTime().Add(activityInfo.ScheduleToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToCloseTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
expectedTimerSequence.TimerCreated = false
|
|
timerSequence = s.timerSequence.getActivityScheduleToCloseTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToCloseTimeout_WithoutTimeout_NotScheduled() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: common.EmptyEventID,
|
|
ScheduledTime: nil,
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(0),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityScheduleToCloseTimeout_WithoutTimeout_Scheduled() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(0),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedScheduleToClose,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityScheduleToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
timerSequence = s.timerSequence.getActivityScheduleToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityStartToCloseTimeout_WithTimeout_NotStarted() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityStartToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityStartToCloseTimeout_WithTimeout_Started() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedStartToClose,
|
|
Attempt: 12,
|
|
}
|
|
|
|
expectedTimerSequence := &TimerSequenceID{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.StartedTime.AsTime().Add(activityInfo.StartToCloseTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_START_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityStartToCloseTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
expectedTimerSequence.TimerCreated = false
|
|
timerSequence = s.timerSequence.getActivityStartToCloseTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityStartToCloseTimeout_WithoutTimeout_NotStarted() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(0),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityStartToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityStartToCloseTimeout_WithoutTimeout_Started() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(0),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedStartToClose,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityStartToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
timerSequence = s.timerSequence.getActivityStartToCloseTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityHeartbeatTimeout_WithHeartbeat_NotStarted() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityHeartbeatTimeout_WithHeartbeat_Started_NoHeartbeat() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedHeartbeat,
|
|
Attempt: 12,
|
|
}
|
|
|
|
expectedTimerSequence := &TimerSequenceID{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.StartedTime.AsTime().Add(activityInfo.HeartbeatTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
expectedTimerSequence.TimerCreated = false
|
|
timerSequence = s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityHeartbeatTimeout_WithHeartbeat_Started_Heartbeated() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(1),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedHeartbeat,
|
|
Attempt: 12,
|
|
}
|
|
|
|
expectedTimerSequence := &TimerSequenceID{
|
|
EventID: activityInfo.ScheduledEventId,
|
|
Timestamp: activityInfo.LastHeartbeatUpdateTime.AsTime().Add(activityInfo.HeartbeatTimeout.AsDuration()),
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
expectedTimerSequence.TimerCreated = false
|
|
timerSequence = s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Equal(expectedTimerSequence, timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityHeartbeatTimeout_WithoutHeartbeat_NotStarted() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: common.EmptyEventID,
|
|
StartedTime: nil,
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusNone,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityHeartbeatTimeout_WithoutHeartbeat_Started_NoHeartbeat() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: nil,
|
|
TimerTaskStatus: TimerTaskStatusCreatedHeartbeat,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
timerSequence = s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestGetActivityHeartbeatTimeout_WithoutHeartbeat_Started_Heartbeated() {
|
|
now := time.Now().UTC()
|
|
activityInfo := &persistencespb.ActivityInfo{
|
|
Version: 123,
|
|
ScheduledEventId: 234,
|
|
ScheduledTime: timestamppb.New(now),
|
|
StartedEventId: 345,
|
|
StartedTime: timestamppb.New(now.Add(200 * time.Millisecond)),
|
|
ActivityId: "some random activity ID",
|
|
ScheduleToStartTimeout: timestamp.DurationFromSeconds(10),
|
|
ScheduleToCloseTimeout: timestamp.DurationFromSeconds(1000),
|
|
StartToCloseTimeout: timestamp.DurationFromSeconds(100),
|
|
HeartbeatTimeout: timestamp.DurationFromSeconds(0),
|
|
LastHeartbeatUpdateTime: timestamppb.New(now.Add(400 * time.Millisecond)),
|
|
TimerTaskStatus: TimerTaskStatusCreatedHeartbeat,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequence := s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
|
|
activityInfo.TimerTaskStatus = TimerTaskStatusNone
|
|
timerSequence = s.timerSequence.getActivityHeartbeatTimeout(activityInfo)
|
|
s.Empty(timerSequence)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestConversion() {
|
|
s.Equal(int32(TimerTaskStatusCreatedStartToClose), timerTypeToTimerMask(enumspb.TIMEOUT_TYPE_START_TO_CLOSE))
|
|
s.Equal(int32(TimerTaskStatusCreatedScheduleToStart), timerTypeToTimerMask(enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START))
|
|
s.Equal(int32(TimerTaskStatusCreatedScheduleToClose), timerTypeToTimerMask(enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE))
|
|
s.Equal(int32(TimerTaskStatusCreatedHeartbeat), timerTypeToTimerMask(enumspb.TIMEOUT_TYPE_HEARTBEAT))
|
|
|
|
s.Equal(TimerTaskStatusNone, 0)
|
|
s.Equal(TimerTaskStatusCreated, 1)
|
|
s.Equal(TimerTaskStatusCreatedStartToClose, 1)
|
|
s.Equal(TimerTaskStatusCreatedScheduleToStart, 2)
|
|
s.Equal(TimerTaskStatusCreatedScheduleToClose, 4)
|
|
s.Equal(TimerTaskStatusCreatedHeartbeat, 8)
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLess_CompareTime() {
|
|
now := time.Now().UTC()
|
|
timerSequenceID1 := TimerSequenceID{
|
|
EventID: 123,
|
|
Timestamp: now,
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequenceID2 := TimerSequenceID{
|
|
EventID: 123,
|
|
Timestamp: now.Add(time.Second),
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequenceIDs := TimerSequenceIDs([]TimerSequenceID{timerSequenceID1, timerSequenceID2})
|
|
s.True(timerSequenceIDs.Less(0, 1))
|
|
s.False(timerSequenceIDs.Less(1, 0))
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLess_CompareEventID() {
|
|
now := time.Now().UTC()
|
|
timerSequenceID1 := TimerSequenceID{
|
|
EventID: 122,
|
|
Timestamp: now,
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequenceID2 := TimerSequenceID{
|
|
EventID: 123,
|
|
Timestamp: now,
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequenceIDs := TimerSequenceIDs([]TimerSequenceID{timerSequenceID1, timerSequenceID2})
|
|
s.True(timerSequenceIDs.Less(0, 1))
|
|
s.False(timerSequenceIDs.Less(1, 0))
|
|
}
|
|
|
|
func (s *timerSequenceSuite) TestLess_CompareType() {
|
|
now := time.Now().UTC()
|
|
timerSequenceID1 := TimerSequenceID{
|
|
EventID: 123,
|
|
Timestamp: now,
|
|
TimerType: enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequenceID2 := TimerSequenceID{
|
|
EventID: 123,
|
|
Timestamp: now,
|
|
TimerType: enumspb.TIMEOUT_TYPE_HEARTBEAT,
|
|
TimerCreated: true,
|
|
Attempt: 12,
|
|
}
|
|
|
|
timerSequenceIDs := TimerSequenceIDs([]TimerSequenceID{timerSequenceID1, timerSequenceID2})
|
|
s.True(timerSequenceIDs.Less(0, 1))
|
|
s.False(timerSequenceIDs.Less(1, 0))
|
|
}
|