diff --git a/cmd/tools/rpcwrappers/main.go b/cmd/tools/rpcwrappers/main.go index aba845726c4..635c2a73813 100644 --- a/cmd/tools/rpcwrappers/main.go +++ b/cmd/tools/rpcwrappers/main.go @@ -30,12 +30,12 @@ import ( "io" "os" "reflect" + "slices" "strings" "text/template" "go.temporal.io/api/taskqueue/v1" "go.temporal.io/api/workflowservice/v1" - "golang.org/x/exp/slices" "go.temporal.io/server/api/adminservice/v1" "go.temporal.io/server/api/historyservice/v1" diff --git a/common/headers/version_checker.go b/common/headers/version_checker.go index cd1da41b4ca..b509797c6d0 100644 --- a/common/headers/version_checker.go +++ b/common/headers/version_checker.go @@ -27,10 +27,10 @@ package headers import ( "context" "fmt" + "slices" "strings" "github.com/blang/semver/v4" - "golang.org/x/exp/slices" "google.golang.org/grpc/metadata" "go.temporal.io/api/serviceerror" diff --git a/common/membership/ringpop/service_resolver.go b/common/membership/ringpop/service_resolver.go index 907d0b99876..ef88879655f 100644 --- a/common/membership/ringpop/service_resolver.go +++ b/common/membership/ringpop/service_resolver.go @@ -28,6 +28,7 @@ import ( "errors" "math" "net" + "slices" "strconv" "sync" "sync/atomic" @@ -35,7 +36,6 @@ import ( "github.com/temporalio/ringpop-go" "github.com/temporalio/tchannel-go" - "golang.org/x/exp/slices" "github.com/dgryski/go-farm" "github.com/temporalio/ringpop-go/events" diff --git a/common/metrics/metrics_test.go b/common/metrics/metrics_test.go index 70f91fdcd59..23935c167f4 100644 --- a/common/metrics/metrics_test.go +++ b/common/metrics/metrics_test.go @@ -26,12 +26,12 @@ package metrics_test import ( "fmt" + "slices" "time" "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/metrics/metricstest" "golang.org/x/exp/maps" - "golang.org/x/exp/slices" ) var ( diff --git a/common/persistence/client/quotas_test.go b/common/persistence/client/quotas_test.go index 0ee29f139e4..216c9f925ba 100644 --- a/common/persistence/client/quotas_test.go +++ b/common/persistence/client/quotas_test.go @@ -26,6 +26,7 @@ package client import ( "reflect" + "slices" "testing" "time" @@ -34,7 +35,6 @@ import ( "go.temporal.io/server/common/headers" "go.temporal.io/server/common/quotas" "go.temporal.io/server/common/testing/temporalapi" - "golang.org/x/exp/slices" "go.temporal.io/api/workflowservice/v1" ) diff --git a/service/frontend/configs/quotas_test.go b/service/frontend/configs/quotas_test.go index 011ba00c975..3685b808148 100644 --- a/service/frontend/configs/quotas_test.go +++ b/service/frontend/configs/quotas_test.go @@ -26,13 +26,13 @@ package configs import ( "reflect" + "slices" "testing" "time" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "go.temporal.io/api/workflowservice/v1" - "golang.org/x/exp/slices" "go.temporal.io/server/common/headers" "go.temporal.io/server/common/quotas" diff --git a/service/frontend/namespace_handler_test.go b/service/frontend/namespace_handler_test.go index 1bb664957e1..e55d91dcdea 100644 --- a/service/frontend/namespace_handler_test.go +++ b/service/frontend/namespace_handler_test.go @@ -26,12 +26,12 @@ package frontend import ( "context" + "slices" "strings" "testing" "time" "go.temporal.io/api/serviceerror" - "golang.org/x/exp/slices" "github.com/golang/mock/gomock" "github.com/pborman/uuid" diff --git a/service/history/configs/quotas_test.go b/service/history/configs/quotas_test.go index dda8414ae69..e32d576955f 100644 --- a/service/history/configs/quotas_test.go +++ b/service/history/configs/quotas_test.go @@ -26,12 +26,12 @@ package configs import ( "reflect" + "slices" "testing" "time" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" - "golang.org/x/exp/slices" "go.temporal.io/server/api/historyservice/v1" "go.temporal.io/server/common/headers" diff --git a/service/history/ndc/workflow_state_replicator.go b/service/history/ndc/workflow_state_replicator.go index 002b35273d7..c2b2c527e8c 100644 --- a/service/history/ndc/workflow_state_replicator.go +++ b/service/history/ndc/workflow_state_replicator.go @@ -29,13 +29,13 @@ package ndc import ( "context" "fmt" + "slices" "sort" "time" commonpb "go.temporal.io/api/common/v1" historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" - "golang.org/x/exp/slices" "go.temporal.io/server/api/adminservice/v1" enumsspb "go.temporal.io/server/api/enums/v1" diff --git a/service/history/queues/action_pending_task_count.go b/service/history/queues/action_pending_task_count.go index 3506ce94293..e58cb5600fd 100644 --- a/service/history/queues/action_pending_task_count.go +++ b/service/history/queues/action_pending_task_count.go @@ -25,10 +25,9 @@ package queues import ( + "slices" "time" - "golang.org/x/exp/slices" - "go.temporal.io/server/common/collection" ) diff --git a/service/history/queues/action_slice_count.go b/service/history/queues/action_slice_count.go index b1d8d27b1b1..8c37a8396ce 100644 --- a/service/history/queues/action_slice_count.go +++ b/service/history/queues/action_slice_count.go @@ -25,7 +25,7 @@ package queues import ( - "golang.org/x/exp/slices" + "slices" "go.temporal.io/server/service/history/tasks" ) diff --git a/service/history/queues/queue_scheduled_test.go b/service/history/queues/queue_scheduled_test.go index 0ff1ba7644f..3e9e3487f90 100644 --- a/service/history/queues/queue_scheduled_test.go +++ b/service/history/queues/queue_scheduled_test.go @@ -29,13 +29,13 @@ import ( "errors" "math" "math/rand" + "slices" "testing" "time" gomock "github.com/golang/mock/gomock" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" - "golang.org/x/exp/slices" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/common/cluster" diff --git a/service/history/queues/scope_test.go b/service/history/queues/scope_test.go index a21a8736834..6ca76c3cacd 100644 --- a/service/history/queues/scope_test.go +++ b/service/history/queues/scope_test.go @@ -26,6 +26,7 @@ package queues import ( "math/rand" + "slices" "testing" "time" @@ -33,7 +34,6 @@ import ( "github.com/pborman/uuid" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" - "golang.org/x/exp/slices" enumsspb "go.temporal.io/server/api/enums/v1" "go.temporal.io/server/common/predicates" diff --git a/service/history/queues/slice_test.go b/service/history/queues/slice_test.go index f690eaffec5..8bdacc640e3 100644 --- a/service/history/queues/slice_test.go +++ b/service/history/queues/slice_test.go @@ -27,6 +27,7 @@ package queues import ( "errors" "math/rand" + "slices" "testing" "time" @@ -34,7 +35,6 @@ import ( "github.com/pborman/uuid" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" - "golang.org/x/exp/slices" enumsspb "go.temporal.io/server/api/enums/v1" "go.temporal.io/server/common/clock" diff --git a/service/history/queues/test_util.go b/service/history/queues/test_util.go index 39380a99eda..67695fbd4e2 100644 --- a/service/history/queues/test_util.go +++ b/service/history/queues/test_util.go @@ -27,10 +27,9 @@ package queues import ( "math" "math/rand" + "slices" "time" - "golang.org/x/exp/slices" - "go.temporal.io/server/common/predicates" "go.temporal.io/server/service/history/tasks" ) diff --git a/service/history/tasks/predicates_test.go b/service/history/tasks/predicates_test.go index 4e6ca9590aa..480e49f50be 100644 --- a/service/history/tasks/predicates_test.go +++ b/service/history/tasks/predicates_test.go @@ -25,6 +25,7 @@ package tasks import ( + "slices" "testing" "github.com/golang/mock/gomock" @@ -32,7 +33,6 @@ import ( "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" "golang.org/x/exp/rand" - "golang.org/x/exp/slices" enumsspb "go.temporal.io/server/api/enums/v1" "go.temporal.io/server/api/persistence/v1" diff --git a/service/history/workflow/mutable_state_impl.go b/service/history/workflow/mutable_state_impl.go index d203aa4e39b..3cbca1e807c 100644 --- a/service/history/workflow/mutable_state_impl.go +++ b/service/history/workflow/mutable_state_impl.go @@ -29,6 +29,7 @@ import ( "context" "fmt" "math/rand" + "slices" "time" "github.com/nexus-rpc/sdk-go/nexus" @@ -43,7 +44,6 @@ import ( updatepb "go.temporal.io/api/update/v1" workflowpb "go.temporal.io/api/workflow/v1" "go.temporal.io/api/workflowservice/v1" - "golang.org/x/exp/slices" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" @@ -97,12 +97,10 @@ const ( int64SizeBytes = 8 ) -var ( - // Scheduled tasks with timestamp after this will not be created. - // Those tasks are too far in the future and pratically never fire and just consume storage space. - // NOTE: this value is less than timer.MaxAllowedTimer so that no capped timers will be created. - maxScheduledTaskDuration = time.Hour * 24 * 365 * 99 -) +// Scheduled tasks with timestamp after this will not be created. +// Those tasks are too far in the future and pratically never fire and just consume storage space. +// NOTE: this value is less than timer.MaxAllowedTimer so that no capped timers will be created. +var maxScheduledTaskDuration = time.Hour * 24 * 365 * 99 var ( // ErrWorkflowFinished indicates trying to mutate mutable state after workflow finished @@ -350,7 +348,6 @@ func NewMutableStateFromDB( dbRecord *persistencespb.WorkflowMutableState, dbRecordVersion int64, ) (*MutableStateImpl, error) { - // startTime will be overridden by DB record startTime := time.Time{} mutableState := NewMutableState( @@ -466,7 +463,6 @@ func NewSanitizedMutableState( lastFirstEventTxnID int64, lastWriteVersion int64, ) (*MutableStateImpl, error) { - mutableState, err := NewMutableStateFromDB(shard, eventsCache, logger, namespaceEntry, mutableStateRecord, 1) if err != nil { return nil, err @@ -690,7 +686,6 @@ func (ms *MutableStateImpl) SetHistoryTree( func (ms *MutableStateImpl) SetCurrentBranchToken( branchToken []byte, ) error { - currentVersionHistory, err := versionhistory.GetCurrentVersionHistory(ms.executionInfo.VersionHistories) if err != nil { return err @@ -738,7 +733,6 @@ func (ms *MutableStateImpl) UpdateCurrentVersion( version int64, forceUpdate bool, ) error { - if ms.transitionHistoryEnabled && len(ms.executionInfo.TransitionHistory) != 0 { // this make sure current version >= last write version lastVersionedTransition := ms.executionInfo.TransitionHistory[len(ms.executionInfo.TransitionHistory)-1] @@ -1058,7 +1052,6 @@ func (ms *MutableStateImpl) GetActivityScheduledEvent( ctx context.Context, scheduledEventID int64, ) (*historypb.HistoryEvent, error) { - ai, ok := ms.pendingActivityInfoIDs[scheduledEventID] if !ok { return nil, ErrMissingActivityInfo @@ -1097,7 +1090,6 @@ func (ms *MutableStateImpl) GetActivityScheduledEvent( func (ms *MutableStateImpl) GetActivityInfo( scheduledEventID int64, ) (*persistencespb.ActivityInfo, bool) { - ai, ok := ms.pendingActivityInfoIDs[scheduledEventID] return ai, ok } @@ -1119,7 +1111,6 @@ func (ms *MutableStateImpl) GetActivityInfoWithTimerHeartbeat( func (ms *MutableStateImpl) GetActivityByActivityID( activityID string, ) (*persistencespb.ActivityInfo, bool) { - eventID, ok := ms.pendingActivityIDToEventID[activityID] if !ok { return nil, false @@ -1148,7 +1139,6 @@ func (ms *MutableStateImpl) GetActivityType( func (ms *MutableStateImpl) GetChildExecutionInfo( initiatedEventID int64, ) (*persistencespb.ChildExecutionInfo, bool) { - ci, ok := ms.pendingChildExecutionInfoIDs[initiatedEventID] return ci, ok } @@ -1159,7 +1149,6 @@ func (ms *MutableStateImpl) GetChildExecutionInitiatedEvent( ctx context.Context, initiatedEventID int64, ) (*historypb.HistoryEvent, error) { - ci, ok := ms.pendingChildExecutionInfoIDs[initiatedEventID] if !ok { return nil, ErrMissingChildWorkflowInfo @@ -1198,7 +1187,6 @@ func (ms *MutableStateImpl) GetChildExecutionInitiatedEvent( func (ms *MutableStateImpl) GetRequestCancelInfo( initiatedEventID int64, ) (*persistencespb.RequestCancelInfo, bool) { - ri, ok := ms.pendingRequestCancelInfoIDs[initiatedEventID] return ri, ok } @@ -1244,7 +1232,6 @@ func (ms *MutableStateImpl) GetRequesteCancelExternalInitiatedEvent( func (ms *MutableStateImpl) GetRetryBackoffDuration( failure *failurepb.Failure, ) (time.Duration, enumspb.RetryState) { - info := ms.executionInfo if !info.HasRetryPolicy { return backoff.NoBackoff, enumspb.RETRY_STATE_RETRY_POLICY_NOT_SET @@ -1275,7 +1262,6 @@ func (ms *MutableStateImpl) GetCronBackoffDuration() time.Duration { func (ms *MutableStateImpl) GetSignalInfo( initiatedEventID int64, ) (*persistencespb.SignalInfo, bool) { - ri, ok := ms.pendingSignalInfoIDs[initiatedEventID] return ri, ok } @@ -1431,7 +1417,6 @@ func (ms *MutableStateImpl) GetWorkflowExecutionDuration(ctx context.Context) (t func (ms *MutableStateImpl) GetStartEvent( ctx context.Context, ) (*historypb.HistoryEvent, error) { - currentBranchToken, err := ms.GetCurrentBranchToken() if err != nil { return nil, err @@ -1487,7 +1472,6 @@ func (ms *MutableStateImpl) GetFirstRunID( func (ms *MutableStateImpl) DeletePendingChildExecution( initiatedEventID int64, ) error { - if prev, ok := ms.pendingChildExecutionInfoIDs[initiatedEventID]; ok { ms.approximateSize -= prev.Size() + int64SizeBytes delete(ms.pendingChildExecutionInfoIDs, initiatedEventID) @@ -1509,7 +1493,6 @@ func (ms *MutableStateImpl) DeletePendingChildExecution( func (ms *MutableStateImpl) DeletePendingRequestCancel( initiatedEventID int64, ) error { - if prev, ok := ms.pendingRequestCancelInfoIDs[initiatedEventID]; ok { ms.approximateSize -= prev.Size() + int64SizeBytes delete(ms.pendingRequestCancelInfoIDs, initiatedEventID) @@ -1531,7 +1514,6 @@ func (ms *MutableStateImpl) DeletePendingRequestCancel( func (ms *MutableStateImpl) DeletePendingSignal( initiatedEventID int64, ) error { - if prev, ok := ms.pendingSignalInfoIDs[initiatedEventID]; ok { ms.approximateSize -= prev.Size() + int64SizeBytes delete(ms.pendingSignalInfoIDs, initiatedEventID) @@ -1634,7 +1616,6 @@ func (ms *MutableStateImpl) UpdateActivityInfo( func (ms *MutableStateImpl) UpdateActivity( ai *persistencespb.ActivityInfo, ) error { - prev, ok := ms.pendingActivityInfoIDs[ai.ScheduledEventId] if !ok { ms.logError( @@ -1655,7 +1636,6 @@ func (ms *MutableStateImpl) UpdateActivityWithTimerHeartbeat( ai *persistencespb.ActivityInfo, timerTimeoutVisibility time.Time, ) error { - err := ms.UpdateActivity(ai) if err != nil { return err @@ -1669,7 +1649,6 @@ func (ms *MutableStateImpl) UpdateActivityWithTimerHeartbeat( func (ms *MutableStateImpl) DeleteActivity( scheduledEventID int64, ) error { - if activityInfo, ok := ms.pendingActivityInfoIDs[scheduledEventID]; ok { delete(ms.pendingActivityInfoIDs, scheduledEventID) delete(ms.pendingActivityTimerHeartbeats, scheduledEventID) @@ -1704,7 +1683,6 @@ func (ms *MutableStateImpl) DeleteActivity( func (ms *MutableStateImpl) GetUserTimerInfo( timerID string, ) (*persistencespb.TimerInfo, bool) { - timerInfo, ok := ms.pendingTimerInfoIDs[timerID] return timerInfo, ok } @@ -1713,7 +1691,6 @@ func (ms *MutableStateImpl) GetUserTimerInfo( func (ms *MutableStateImpl) GetUserTimerInfoByEventID( startEventID int64, ) (*persistencespb.TimerInfo, bool) { - timerID, ok := ms.pendingTimerEventIDToID[startEventID] if !ok { return nil, false @@ -1725,7 +1702,6 @@ func (ms *MutableStateImpl) GetUserTimerInfoByEventID( func (ms *MutableStateImpl) UpdateUserTimer( ti *persistencespb.TimerInfo, ) error { - timerID, ok := ms.pendingTimerEventIDToID[ti.GetStartedEventId()] if !ok { ms.logError( @@ -1752,7 +1728,6 @@ func (ms *MutableStateImpl) UpdateUserTimer( func (ms *MutableStateImpl) DeleteUserTimer( timerID string, ) error { - if timerInfo, ok := ms.pendingTimerInfoIDs[timerID]; ok { delete(ms.pendingTimerInfoIDs, timerID) ms.approximateSize -= timerInfo.Size() + len(timerID) @@ -1928,7 +1903,6 @@ func (ms *MutableStateImpl) IsWorkflowCloseAttempted() bool { func (ms *MutableStateImpl) IsSignalRequested( requestID string, ) bool { - if _, ok := ms.pendingSignalRequestedIDs[requestID]; ok { return true } @@ -1936,7 +1910,6 @@ func (ms *MutableStateImpl) IsSignalRequested( } func (ms *MutableStateImpl) IsWorkflowPendingOnWorkflowTaskBackoff() bool { - workflowTaskBackoff := timestamp.TimeValue(ms.executionInfo.GetExecutionTime()).After(timestamp.TimeValue(ms.executionInfo.GetStartTime())) if workflowTaskBackoff && !ms.HadOrHasWorkflowTask() { return true @@ -1957,7 +1930,6 @@ func (ms *MutableStateImpl) GetApproximatePersistedSize() int { func (ms *MutableStateImpl) AddSignalRequested( requestID string, ) { - if ms.pendingSignalRequestedIDs == nil { ms.pendingSignalRequestedIDs = make(map[string]struct{}) } @@ -1972,7 +1944,6 @@ func (ms *MutableStateImpl) AddSignalRequested( func (ms *MutableStateImpl) DeleteSignalRequested( requestID string, ) { - delete(ms.pendingSignalRequestedIDs, requestID) delete(ms.updateSignalRequestedIDs, requestID) ms.deleteSignalRequestedIDs[requestID] = struct{}{} @@ -1987,7 +1958,6 @@ func (ms *MutableStateImpl) addWorkflowExecutionStartedEventForContinueAsNew( firstRunID string, rootExecutionInfo *workflowspb.RootExecutionInfo, ) (*historypb.HistoryEvent, error) { - previousExecutionInfo := previousExecutionState.GetExecutionInfo() taskQueue := previousExecutionInfo.TaskQueue if command.TaskQueue != nil { @@ -2124,7 +2094,6 @@ func (ms *MutableStateImpl) AddWorkflowExecutionStartedEvent( execution *commonpb.WorkflowExecution, startRequest *historyservice.StartWorkflowExecutionRequest, ) (*historypb.HistoryEvent, error) { - return ms.AddWorkflowExecutionStartedEventWithOptions( execution, startRequest, @@ -2141,7 +2110,6 @@ func (ms *MutableStateImpl) AddWorkflowExecutionStartedEventWithOptions( prevRunID string, firstRunID string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowStarted if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -2195,7 +2163,6 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionStartedEvent( requestID string, startEvent *historypb.HistoryEvent, ) error { - if ms.executionInfo.NamespaceId != ms.namespaceEntry.ID().String() { return serviceerror.NewInternal(fmt.Sprintf("applying conflicting namespace ID: %v != %v", ms.executionInfo.NamespaceId, ms.namespaceEntry.ID().String())) @@ -2830,7 +2797,6 @@ func (ms *MutableStateImpl) AddActivityTaskScheduledEvent( command *commandpb.ScheduleActivityTaskCommandAttributes, bypassTaskGeneration bool, ) (*historypb.HistoryEvent, *persistencespb.ActivityInfo, error) { - opTag := tag.WorkflowActionActivityTaskScheduled if err := ms.checkMutability(opTag); err != nil { return nil, nil, err @@ -2862,7 +2828,6 @@ func (ms *MutableStateImpl) ApplyActivityTaskScheduledEvent( firstEventID int64, event *historypb.HistoryEvent, ) (*persistencespb.ActivityInfo, error) { - attributes := event.GetActivityTaskScheduledEventAttributes() scheduledEventID := event.GetEventId() @@ -3088,7 +3053,6 @@ func (ms *MutableStateImpl) AddActivityTaskCompletedEvent( startedEventID int64, request *workflowservice.RespondActivityTaskCompletedRequest, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionActivityTaskCompleted if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3123,7 +3087,6 @@ func (ms *MutableStateImpl) AddActivityTaskCompletedEvent( func (ms *MutableStateImpl) ApplyActivityTaskCompletedEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetActivityTaskCompletedEventAttributes() scheduledEventID := attributes.GetScheduledEventId() @@ -3138,7 +3101,6 @@ func (ms *MutableStateImpl) AddActivityTaskFailedEvent( identity string, versioningStamp *commonpb.WorkerVersionStamp, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionActivityTaskFailed if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3174,7 +3136,6 @@ func (ms *MutableStateImpl) AddActivityTaskFailedEvent( func (ms *MutableStateImpl) ApplyActivityTaskFailedEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetActivityTaskFailedEventAttributes() scheduledEventID := attributes.GetScheduledEventId() @@ -3187,7 +3148,6 @@ func (ms *MutableStateImpl) AddActivityTaskTimedOutEvent( timeoutFailure *failurepb.Failure, retryState enumspb.RetryState, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionActivityTaskTimedOut if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3228,7 +3188,6 @@ func (ms *MutableStateImpl) AddActivityTaskTimedOutEvent( func (ms *MutableStateImpl) ApplyActivityTaskTimedOutEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetActivityTaskTimedOutEventAttributes() scheduledEventID := attributes.GetScheduledEventId() @@ -3240,7 +3199,6 @@ func (ms *MutableStateImpl) AddActivityTaskCancelRequestedEvent( scheduledEventID int64, _ string, ) (*historypb.HistoryEvent, *persistencespb.ActivityInfo, error) { - opTag := tag.WorkflowActionActivityTaskCancelRequested if err := ms.checkMutability(opTag); err != nil { return nil, nil, err @@ -3284,7 +3242,6 @@ func (ms *MutableStateImpl) AddActivityTaskCancelRequestedEvent( func (ms *MutableStateImpl) ApplyActivityTaskCancelRequestedEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetActivityTaskCancelRequestedEventAttributes() scheduledEventID := attributes.GetScheduledEventId() ai, ok := ms.GetActivityInfo(scheduledEventID) @@ -3318,7 +3275,6 @@ func (ms *MutableStateImpl) AddActivityTaskCanceledEvent( details *commonpb.Payloads, identity string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionActivityTaskCanceled if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3364,7 +3320,6 @@ func (ms *MutableStateImpl) AddActivityTaskCanceledEvent( func (ms *MutableStateImpl) ApplyActivityTaskCanceledEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetActivityTaskCanceledEventAttributes() scheduledEventID := attributes.GetScheduledEventId() @@ -3376,7 +3331,6 @@ func (ms *MutableStateImpl) AddCompletedWorkflowEvent( command *commandpb.CompleteWorkflowExecutionCommandAttributes, newExecutionRunID string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowCompleted if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3400,7 +3354,6 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionCompletedEvent( firstEventID int64, event *historypb.HistoryEvent, ) error { - if err := ms.UpdateWorkflowStateStatus( enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, @@ -3421,7 +3374,6 @@ func (ms *MutableStateImpl) AddFailWorkflowEvent( command *commandpb.FailWorkflowExecutionCommandAttributes, newExecutionRunID string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowFailed if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3445,7 +3397,6 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionFailedEvent( firstEventID int64, event *historypb.HistoryEvent, ) error { - if err := ms.UpdateWorkflowStateStatus( enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_FAILED, @@ -3465,7 +3416,6 @@ func (ms *MutableStateImpl) AddTimeoutWorkflowEvent( retryState enumspb.RetryState, newExecutionRunID string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowTimeout if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3489,7 +3439,6 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionTimedoutEvent( firstEventID int64, event *historypb.HistoryEvent, ) error { - if err := ms.UpdateWorkflowStateStatus( enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT, @@ -3507,7 +3456,6 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionTimedoutEvent( func (ms *MutableStateImpl) AddWorkflowExecutionCancelRequestedEvent( request *historyservice.RequestCancelWorkflowExecutionRequest, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowCancelRequested if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3537,7 +3485,6 @@ func (ms *MutableStateImpl) AddWorkflowExecutionCancelRequestedEvent( func (ms *MutableStateImpl) ApplyWorkflowExecutionCancelRequestedEvent( _ *historypb.HistoryEvent, ) error { - ms.executionInfo.CancelRequested = true return nil } @@ -3546,7 +3493,6 @@ func (ms *MutableStateImpl) AddWorkflowExecutionCanceledEvent( workflowTaskCompletedEventID int64, command *commandpb.CancelWorkflowExecutionCommandAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowCanceled if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3590,7 +3536,6 @@ func (ms *MutableStateImpl) AddRequestCancelExternalWorkflowExecutionInitiatedEv command *commandpb.RequestCancelExternalWorkflowExecutionCommandAttributes, targetNamespaceID namespace.ID, ) (*historypb.HistoryEvent, *persistencespb.RequestCancelInfo, error) { - opTag := tag.WorkflowActionExternalWorkflowCancelInitiated if err := ms.checkMutability(opTag); err != nil { return nil, nil, err @@ -3615,7 +3560,6 @@ func (ms *MutableStateImpl) ApplyRequestCancelExternalWorkflowExecutionInitiated event *historypb.HistoryEvent, cancelRequestID string, ) (*persistencespb.RequestCancelInfo, error) { - // TODO: Evaluate if we need cancelRequestID also part of history event initiatedEventID := event.GetEventId() rci := &persistencespb.RequestCancelInfo{ @@ -3641,7 +3585,6 @@ func (ms *MutableStateImpl) AddExternalWorkflowExecutionCancelRequested( workflowID string, runID string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionExternalWorkflowCancelRequested if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3672,7 +3615,6 @@ func (ms *MutableStateImpl) AddExternalWorkflowExecutionCancelRequested( func (ms *MutableStateImpl) ApplyExternalWorkflowExecutionCancelRequested( event *historypb.HistoryEvent, ) error { - initiatedID := event.GetExternalWorkflowExecutionCancelRequestedEventAttributes().GetInitiatedEventId() return ms.DeletePendingRequestCancel(initiatedID) @@ -3686,7 +3628,6 @@ func (ms *MutableStateImpl) AddRequestCancelExternalWorkflowExecutionFailedEvent runID string, cause enumspb.CancelExternalWorkflowExecutionFailedCause, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionExternalWorkflowCancelFailed if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3719,7 +3660,6 @@ func (ms *MutableStateImpl) AddRequestCancelExternalWorkflowExecutionFailedEvent func (ms *MutableStateImpl) ApplyRequestCancelExternalWorkflowExecutionFailedEvent( event *historypb.HistoryEvent, ) error { - initiatedID := event.GetRequestCancelExternalWorkflowExecutionFailedEventAttributes().GetInitiatedEventId() return ms.DeletePendingRequestCancel(initiatedID) @@ -3731,7 +3671,6 @@ func (ms *MutableStateImpl) AddSignalExternalWorkflowExecutionInitiatedEvent( command *commandpb.SignalExternalWorkflowExecutionCommandAttributes, targetNamespaceID namespace.ID, ) (*historypb.HistoryEvent, *persistencespb.SignalInfo, error) { - opTag := tag.WorkflowActionExternalWorkflowSignalInitiated if err := ms.checkMutability(opTag); err != nil { return nil, nil, err @@ -3756,7 +3695,6 @@ func (ms *MutableStateImpl) ApplySignalExternalWorkflowExecutionInitiatedEvent( event *historypb.HistoryEvent, signalRequestID string, ) (*persistencespb.SignalInfo, error) { - // TODO: Consider also writing signalRequestID to history event initiatedEventID := event.GetEventId() si := &persistencespb.SignalInfo{ @@ -3779,7 +3717,6 @@ func (ms *MutableStateImpl) AddUpsertWorkflowSearchAttributesEvent( workflowTaskCompletedEventID int64, command *commandpb.UpsertWorkflowSearchAttributesCommandAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionUpsertWorkflowSearchAttributes if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3842,7 +3779,6 @@ func (ms *MutableStateImpl) AddExternalWorkflowExecutionSignaled( runID string, control string, // TODO this field is probably deprecated ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionExternalWorkflowSignalRequested if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3874,7 +3810,6 @@ func (ms *MutableStateImpl) AddExternalWorkflowExecutionSignaled( func (ms *MutableStateImpl) ApplyExternalWorkflowExecutionSignaled( event *historypb.HistoryEvent, ) error { - initiatedID := event.GetExternalWorkflowExecutionSignaledEventAttributes().GetInitiatedEventId() return ms.DeletePendingSignal(initiatedID) @@ -3889,7 +3824,6 @@ func (ms *MutableStateImpl) AddSignalExternalWorkflowExecutionFailedEvent( control string, // TODO this field is probably deprecated cause enumspb.SignalExternalWorkflowExecutionFailedCause, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionExternalWorkflowSignalFailed if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -3923,7 +3857,6 @@ func (ms *MutableStateImpl) AddSignalExternalWorkflowExecutionFailedEvent( func (ms *MutableStateImpl) ApplySignalExternalWorkflowExecutionFailedEvent( event *historypb.HistoryEvent, ) error { - initiatedID := event.GetSignalExternalWorkflowExecutionFailedEventAttributes().GetInitiatedEventId() return ms.DeletePendingSignal(initiatedID) @@ -3933,7 +3866,6 @@ func (ms *MutableStateImpl) AddTimerStartedEvent( workflowTaskCompletedEventID int64, command *commandpb.StartTimerCommandAttributes, ) (*historypb.HistoryEvent, *persistencespb.TimerInfo, error) { - opTag := tag.WorkflowActionTimerStarted if err := ms.checkMutability(opTag); err != nil { return nil, nil, err @@ -3960,7 +3892,6 @@ func (ms *MutableStateImpl) AddTimerStartedEvent( func (ms *MutableStateImpl) ApplyTimerStartedEvent( event *historypb.HistoryEvent, ) (*persistencespb.TimerInfo, error) { - attributes := event.GetTimerStartedEventAttributes() timerID := attributes.GetTimerId() @@ -3988,7 +3919,6 @@ func (ms *MutableStateImpl) ApplyTimerStartedEvent( func (ms *MutableStateImpl) AddTimerFiredEvent( timerID string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionTimerFired if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4014,7 +3944,6 @@ func (ms *MutableStateImpl) AddTimerFiredEvent( func (ms *MutableStateImpl) ApplyTimerFiredEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetTimerFiredEventAttributes() timerID := attributes.GetTimerId() @@ -4026,7 +3955,6 @@ func (ms *MutableStateImpl) AddTimerCanceledEvent( command *commandpb.CancelTimerCommandAttributes, identity string, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionTimerCanceled if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4070,7 +3998,6 @@ func (ms *MutableStateImpl) AddTimerCanceledEvent( func (ms *MutableStateImpl) ApplyTimerCanceledEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetTimerCanceledEventAttributes() timerID := attributes.GetTimerId() @@ -4081,7 +4008,6 @@ func (ms *MutableStateImpl) AddRecordMarkerEvent( workflowTaskCompletedEventID int64, command *commandpb.RecordMarkerCommandAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowRecordMarker if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4097,7 +4023,6 @@ func (ms *MutableStateImpl) AddWorkflowExecutionTerminatedEvent( identity string, deleteAfterTerminate bool, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionWorkflowTerminated if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4268,7 +4193,6 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionTerminatedEvent( firstEventID int64, event *historypb.HistoryEvent, ) error { - if err := ms.UpdateWorkflowStateStatus( enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED, @@ -4330,7 +4254,6 @@ func (ms *MutableStateImpl) AddWorkflowExecutionSignaledEvent( func (ms *MutableStateImpl) ApplyWorkflowExecutionSignaled( _ *historypb.HistoryEvent, ) error { - // Increment signal count in mutable state for this workflow execution ms.executionInfo.SignalCount++ return nil @@ -4343,7 +4266,6 @@ func (ms *MutableStateImpl) AddContinueAsNewEvent( parentNamespace namespace.Name, command *commandpb.ContinueAsNewWorkflowExecutionCommandAttributes, ) (*historypb.HistoryEvent, MutableState, error) { - opTag := tag.WorkflowActionWorkflowContinueAsNew if err := ms.checkMutability(opTag); err != nil { return nil, nil, err @@ -4469,7 +4391,6 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionContinuedAsNewEvent( firstEventID int64, continueAsNewEvent *historypb.HistoryEvent, ) error { - if err := ms.UpdateWorkflowStateStatus( enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW, @@ -4490,7 +4411,6 @@ func (ms *MutableStateImpl) AddStartChildWorkflowExecutionInitiatedEvent( command *commandpb.StartChildWorkflowExecutionCommandAttributes, targetNamespaceID namespace.ID, ) (*historypb.HistoryEvent, *persistencespb.ChildExecutionInfo, error) { - opTag := tag.WorkflowActionChildWorkflowInitiated if err := ms.checkMutability(opTag); err != nil { return nil, nil, err @@ -4515,7 +4435,6 @@ func (ms *MutableStateImpl) ApplyStartChildWorkflowExecutionInitiatedEvent( event *historypb.HistoryEvent, createRequestID string, ) (*persistencespb.ChildExecutionInfo, error) { - initiatedEventID := event.GetEventId() attributes := event.GetStartChildWorkflowExecutionInitiatedEventAttributes() ci := &persistencespb.ChildExecutionInfo{ @@ -4547,7 +4466,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionStartedEvent( header *commonpb.Header, clock *clockspb.VectorClock, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionChildWorkflowStarted if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4581,7 +4499,6 @@ func (ms *MutableStateImpl) ApplyChildWorkflowExecutionStartedEvent( event *historypb.HistoryEvent, clock *clockspb.VectorClock, ) error { - attributes := event.GetChildWorkflowExecutionStartedEventAttributes() initiatedID := attributes.GetInitiatedEventId() @@ -4610,7 +4527,6 @@ func (ms *MutableStateImpl) AddStartChildWorkflowExecutionFailedEvent( cause enumspb.StartChildWorkflowExecutionFailedCause, initiatedEventAttributes *historypb.StartChildWorkflowExecutionInitiatedEventAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionChildWorkflowInitiationFailed if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4645,7 +4561,6 @@ func (ms *MutableStateImpl) AddStartChildWorkflowExecutionFailedEvent( func (ms *MutableStateImpl) ApplyStartChildWorkflowExecutionFailedEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetStartChildWorkflowExecutionFailedEventAttributes() initiatedID := attributes.GetInitiatedEventId() @@ -4657,7 +4572,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionCompletedEvent( childExecution *commonpb.WorkflowExecution, attributes *historypb.WorkflowExecutionCompletedEventAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionChildWorkflowCompleted if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4695,7 +4609,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionCompletedEvent( func (ms *MutableStateImpl) ApplyChildWorkflowExecutionCompletedEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetChildWorkflowExecutionCompletedEventAttributes() initiatedID := attributes.GetInitiatedEventId() @@ -4707,7 +4620,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionFailedEvent( childExecution *commonpb.WorkflowExecution, attributes *historypb.WorkflowExecutionFailedEventAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionChildWorkflowFailed if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4746,7 +4658,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionFailedEvent( func (ms *MutableStateImpl) ApplyChildWorkflowExecutionFailedEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetChildWorkflowExecutionFailedEventAttributes() initiatedID := attributes.GetInitiatedEventId() @@ -4758,7 +4669,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionCanceledEvent( childExecution *commonpb.WorkflowExecution, attributes *historypb.WorkflowExecutionCanceledEventAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionChildWorkflowCanceled if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4796,7 +4706,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionCanceledEvent( func (ms *MutableStateImpl) ApplyChildWorkflowExecutionCanceledEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetChildWorkflowExecutionCanceledEventAttributes() initiatedID := attributes.GetInitiatedEventId() @@ -4808,7 +4717,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionTerminatedEvent( childExecution *commonpb.WorkflowExecution, _ *historypb.WorkflowExecutionTerminatedEventAttributes, // TODO this field is not used at all ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionChildWorkflowTerminated if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4845,7 +4753,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionTerminatedEvent( func (ms *MutableStateImpl) ApplyChildWorkflowExecutionTerminatedEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetChildWorkflowExecutionTerminatedEventAttributes() initiatedID := attributes.GetInitiatedEventId() @@ -4857,7 +4764,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionTimedOutEvent( childExecution *commonpb.WorkflowExecution, attributes *historypb.WorkflowExecutionTimedOutEventAttributes, ) (*historypb.HistoryEvent, error) { - opTag := tag.WorkflowActionChildWorkflowTimedOut if err := ms.checkMutability(opTag); err != nil { return nil, err @@ -4895,7 +4801,6 @@ func (ms *MutableStateImpl) AddChildWorkflowExecutionTimedOutEvent( func (ms *MutableStateImpl) ApplyChildWorkflowExecutionTimedOutEvent( event *historypb.HistoryEvent, ) error { - attributes := event.GetChildWorkflowExecutionTimedOutEventAttributes() initiatedID := attributes.GetInitiatedEventId() @@ -4906,7 +4811,6 @@ func (ms *MutableStateImpl) RetryActivity( ai *persistencespb.ActivityInfo, failure *failurepb.Failure, ) (enumspb.RetryState, error) { - opTag := tag.WorkflowActionActivityTaskRetry if err := ms.checkMutability(opTag); err != nil { return enumspb.RETRY_STATE_INTERNAL_SERVER_ERROR, err @@ -5046,7 +4950,6 @@ func (ms *MutableStateImpl) processCloseCallbacks() error { func (ms *MutableStateImpl) AddTasks( newTasks ...tasks.Task, ) { - now := ms.timeSource.Now() for _, task := range newTasks { category := task.GetCategory() @@ -5069,7 +4972,6 @@ func (ms *MutableStateImpl) SetUpdateCondition( nextEventIDInDB int64, dbRecordVersion int64, ) { - ms.nextEventIDInDB = nextEventIDInDB ms.dbRecordVersion = dbRecordVersion } @@ -5112,7 +5014,6 @@ func (ms *MutableStateImpl) UpdateWorkflowStateStatus( state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus, ) error { - ms.executionStateUpdated = true ms.visibilityUpdated = true // workflow status & state change triggers visibility change as well return setStateStatus(ms.executionState, state, status) @@ -5158,7 +5059,6 @@ func (ms *MutableStateImpl) StartTransaction( func (ms *MutableStateImpl) CloseTransactionAsMutation( transactionPolicy TransactionPolicy, ) (*persistence.WorkflowMutation, []*persistence.WorkflowEvents, error) { - result, err := ms.closeTransaction(transactionPolicy) if err != nil { return nil, nil, err @@ -5201,7 +5101,6 @@ func (ms *MutableStateImpl) CloseTransactionAsMutation( func (ms *MutableStateImpl) CloseTransactionAsSnapshot( transactionPolicy TransactionPolicy, ) (*persistence.WorkflowSnapshot, []*persistence.WorkflowEvents, error) { - result, err := ms.closeTransaction(transactionPolicy) if err != nil { return nil, nil, err @@ -5412,7 +5311,6 @@ func (ms *MutableStateImpl) closeTransactionUpdateTransitionHistory( transactionPolicy TransactionPolicy, workflowEventsSeq []*persistence.WorkflowEvents, ) error { - if len(workflowEventsSeq) > 0 { lastEvents := workflowEventsSeq[len(workflowEventsSeq)-1].Events lastEvent := lastEvents[len(lastEvents)-1] @@ -5538,7 +5436,6 @@ func (ms *MutableStateImpl) closeTransactionPrepareReplicationTasks( eventBatches [][]*historypb.HistoryEvent, clearBufferEvents bool, ) error { - if ms.config.ReplicationMultipleBatches() { if err := ms.eventsToReplicationTask(transactionPolicy, eventBatches); err != nil { return err @@ -5620,7 +5517,6 @@ func (ms *MutableStateImpl) cleanupTransaction() error { func (ms *MutableStateImpl) closeTransactionPrepareEvents( transactionPolicy TransactionPolicy, ) ([]*persistence.WorkflowEvents, [][]*historypb.HistoryEvent, []*historypb.HistoryEvent, bool, error) { - currentBranchToken, err := ms.GetCurrentBranchToken() if err != nil { return nil, nil, nil, false, err @@ -5773,7 +5669,6 @@ func (ms *MutableStateImpl) updateWithLastWriteEvent( lastEvent *historypb.HistoryEvent, transactionPolicy TransactionPolicy, ) error { - if transactionPolicy == TransactionPolicyPassive { // already handled in mutable state. return nil @@ -5800,7 +5695,6 @@ func (ms *MutableStateImpl) validateNoEventsAfterWorkflowFinish( transactionPolicy TransactionPolicy, workflowEventSeq []*persistence.WorkflowEvents, ) error { - if transactionPolicy == TransactionPolicyPassive || len(workflowEventSeq) == 0 { return nil @@ -5862,7 +5756,6 @@ func (ms *MutableStateImpl) startTransactionHandleNamespaceMigration( } func (ms *MutableStateImpl) startTransactionHandleWorkflowTaskFailover() (bool, error) { - if !ms.IsWorkflowExecutionRunning() { return false, nil } @@ -5924,7 +5817,7 @@ func (ms *MutableStateImpl) startTransactionHandleWorkflowTaskFailover() (bool, } // handle case 1 & 2 - var flushBufferVersion = lastWriteVersion + flushBufferVersion := lastWriteVersion // handle case 3 if lastWriteCluster != currentCluster && currentVersionCluster == currentCluster { @@ -5995,7 +5888,6 @@ func (ms *MutableStateImpl) BufferSizeAcceptable() bool { func (ms *MutableStateImpl) closeTransactionHandleBufferedEventsLimit( transactionPolicy TransactionPolicy, ) error { - if transactionPolicy == TransactionPolicyPassive || !ms.IsWorkflowExecutionRunning() { return nil @@ -6027,7 +5919,6 @@ func (ms *MutableStateImpl) closeTransactionHandleBufferedEventsLimit( func (ms *MutableStateImpl) closeTransactionHandleWorkflowResetTask( transactionPolicy TransactionPolicy, ) error { - if transactionPolicy == TransactionPolicyPassive || !ms.IsWorkflowExecutionRunning() { return nil @@ -6134,7 +6025,6 @@ func (ms *MutableStateImpl) generateReplicationTask() bool { func (ms *MutableStateImpl) checkMutability( actionTag tag.ZapTag, ) error { - if !ms.IsWorkflowExecutionRunning() { ms.logWarn( mutableStateInvalidHistoryActionMsg, @@ -6186,7 +6076,6 @@ func (ms *MutableStateImpl) shouldInvalidateCheckum() bool { func (ms *MutableStateImpl) createInternalServerError( actionTag tag.ZapTag, ) error { - return serviceerror.NewInternal(actionTag.Field().String + " operation failed") } diff --git a/service/history/workflow/retry.go b/service/history/workflow/retry.go index 3b48550c1fa..70cff669ccf 100644 --- a/service/history/workflow/retry.go +++ b/service/history/workflow/retry.go @@ -27,10 +27,10 @@ package workflow import ( "context" "math" + "slices" "time" "github.com/pborman/uuid" - "golang.org/x/exp/slices" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" diff --git a/service/matching/configs/quotas_test.go b/service/matching/configs/quotas_test.go index d906e60bc6d..e9d7b8da9fc 100644 --- a/service/matching/configs/quotas_test.go +++ b/service/matching/configs/quotas_test.go @@ -26,12 +26,12 @@ package configs import ( "reflect" + "slices" "testing" "time" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" - "golang.org/x/exp/slices" "go.temporal.io/server/api/matchingservice/v1" "go.temporal.io/server/common/headers" diff --git a/service/matching/version_sets.go b/service/matching/version_sets.go index d1e69d1c5f1..e4fb2b3ccc2 100644 --- a/service/matching/version_sets.go +++ b/service/matching/version_sets.go @@ -25,16 +25,15 @@ package matching import ( - "fmt" - "crypto/sha256" "encoding/base64" + "fmt" + "slices" commonpb "go.temporal.io/api/common/v1" "go.temporal.io/api/serviceerror" taskqueuepb "go.temporal.io/api/taskqueue/v1" "go.temporal.io/api/workflowservice/v1" - "golang.org/x/exp/slices" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/common" @@ -44,10 +43,8 @@ import ( "go.temporal.io/server/common/worker_versioning" ) -var ( - // Error used to signal that a queue has no versioning data. This shouldn't escape matching. - errEmptyVersioningData = serviceerror.NewInternal("versioning data is empty") -) +// Error used to signal that a queue has no versioning data. This shouldn't escape matching. +var errEmptyVersioningData = serviceerror.NewInternal("versioning data is empty") // ToBuildIdOrderingResponse transforms the internal VersioningData representation to public representation. // If maxSets is given, the last sets up to maxSets will be returned. diff --git a/service/worker/batcher/activities_test.go b/service/worker/batcher/activities_test.go index cce789805d1..7d913a80ebd 100644 --- a/service/worker/batcher/activities_test.go +++ b/service/worker/batcher/activities_test.go @@ -26,6 +26,7 @@ package batcher import ( "context" + "slices" "testing" "time" "unicode" @@ -39,7 +40,6 @@ import ( "go.temporal.io/api/workflowservice/v1" "go.temporal.io/api/workflowservicemock/v1" "go.temporal.io/sdk/testsuite" - "golang.org/x/exp/slices" "go.temporal.io/server/common/log" "go.temporal.io/server/common/primitives/timestamp" @@ -59,6 +59,7 @@ func (s *activitiesSuite) SetupTest() { s.mockFrontendClient = workflowservicemock.NewMockWorkflowServiceClient(s.controller) } + func TestActivitiesSuite(t *testing.T) { suite.Run(t, new(activitiesSuite)) } @@ -208,7 +209,7 @@ func (s *activitiesSuite) TestGetResetPoint() { { name: "not found", points: []*workflowpb.ResetPointInfo{ - &workflowpb.ResetPointInfo{ + { BuildId: "build1", RunId: "run1", FirstWorkflowTaskCompletedId: 123, @@ -221,7 +222,7 @@ func (s *activitiesSuite) TestGetResetPoint() { { name: "found", points: []*workflowpb.ResetPointInfo{ - &workflowpb.ResetPointInfo{ + { BuildId: "build1", RunId: "run1", FirstWorkflowTaskCompletedId: 123, @@ -234,7 +235,7 @@ func (s *activitiesSuite) TestGetResetPoint() { { name: "not resettable", points: []*workflowpb.ResetPointInfo{ - &workflowpb.ResetPointInfo{ + { BuildId: "build1", RunId: "run1", FirstWorkflowTaskCompletedId: 123, @@ -247,7 +248,7 @@ func (s *activitiesSuite) TestGetResetPoint() { { name: "from another run", points: []*workflowpb.ResetPointInfo{ - &workflowpb.ResetPointInfo{ + { BuildId: "build1", RunId: "run0", FirstWorkflowTaskCompletedId: 34, @@ -261,7 +262,7 @@ func (s *activitiesSuite) TestGetResetPoint() { { name: "from another run but not allowed", points: []*workflowpb.ResetPointInfo{ - &workflowpb.ResetPointInfo{ + { BuildId: "build1", RunId: "run0", FirstWorkflowTaskCompletedId: 34, @@ -275,7 +276,7 @@ func (s *activitiesSuite) TestGetResetPoint() { { name: "expired", points: []*workflowpb.ResetPointInfo{ - &workflowpb.ResetPointInfo{ + { BuildId: "build1", RunId: "run1", FirstWorkflowTaskCompletedId: 123, diff --git a/service/worker/scheduler/workflow.go b/service/worker/scheduler/workflow.go index f627f5c2311..4f6b74a460c 100644 --- a/service/worker/scheduler/workflow.go +++ b/service/worker/scheduler/workflow.go @@ -29,10 +29,10 @@ import ( "cmp" "errors" "fmt" + "slices" "time" "github.com/google/uuid" - "golang.org/x/exp/slices" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" @@ -1236,7 +1236,8 @@ func (s *scheduler) processBuffer() bool { } result, err := s.startWorkflow(start, req) metricsWithTag := s.metrics.WithTags(map[string]string{ - metrics.ScheduleActionTypeTag: metrics.ScheduleActionStartWorkflow}) + metrics.ScheduleActionTypeTag: metrics.ScheduleActionStartWorkflow, + }) if err != nil { s.logger.Error("Failed to start workflow", "error", err) if !isUserScheduleError(err) { diff --git a/temporal/server_impl.go b/temporal/server_impl.go index 0e576395b61..eaaec21d970 100644 --- a/temporal/server_impl.go +++ b/temporal/server_impl.go @@ -28,9 +28,9 @@ import ( "cmp" "context" "fmt" + "slices" "go.uber.org/multierr" - "golang.org/x/exp/slices" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/config" diff --git a/temporal/server_options.go b/temporal/server_options.go index 1714b91b0ae..421cb813401 100644 --- a/temporal/server_options.go +++ b/temporal/server_options.go @@ -27,8 +27,8 @@ package temporal import ( "fmt" "net/http" + "slices" - "golang.org/x/exp/slices" "google.golang.org/grpc" "go.temporal.io/server/client" diff --git a/tools/tdbg/dlq_v2_service.go b/tools/tdbg/dlq_v2_service.go index e6e00439468..295bc792365 100644 --- a/tools/tdbg/dlq_v2_service.go +++ b/tools/tdbg/dlq_v2_service.go @@ -29,6 +29,7 @@ import ( "encoding/json" "fmt" "io" + "slices" "sort" "strconv" "strings" @@ -36,7 +37,6 @@ import ( "github.com/urfave/cli/v2" commonpb "go.temporal.io/api/common/v1" "go.uber.org/multierr" - "golang.org/x/exp/slices" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto"