Strip workflow cache out of history engine

common/deadlock/deadlock.go

@@ -42,7 +42,7 @@ import (
  "go.temporal.io/server/common/metrics"
  "go.temporal.io/server/common/namespace"
  "go.temporal.io/server/internal/goro"
- "go.temporal.io/server/service/history/shard"
+ "go.temporal.io/server/service/history/definition"
 )
 
 type (
@@ -57,7 +57,7 @@ type (
  // root pingables:
  NamespaceRegistry namespace.Registry
  ClusterMetadata cluster.Metadata
- ShardController shard.Controller `optional:"true"`
+ ShardController definition.ShardController `optional:"true"`
  }
 
  config struct {

service/history/api/activity_util.go

@@ -31,7 +31,7 @@ import (
  "go.temporal.io/api/serviceerror"
 
  tokenspb "go.temporal.io/server/api/token/v1"
- "go.temporal.io/server/service/history/workflow"
+ "go.temporal.io/server/service/history/definition"
 )
 
 func SetActivityTaskRunID(
@@ -64,7 +64,7 @@ func SetActivityTaskRunID(
 
 func GetActivityScheduledEventID(
  activityID string,
- mutableState workflow.MutableState,
+ mutableState definition.MutableState,
 ) (int64, error) {
 
  if activityID == "" {

service/history/api/consistency_checker.go

@@ -33,21 +33,19 @@ import (
 
  clockspb "go.temporal.io/server/api/clock/v1"
 
- "go.temporal.io/server/common/definition"
  "go.temporal.io/server/common/namespace"
  "go.temporal.io/server/common/persistence"
  "go.temporal.io/server/common/persistence/versionhistory"
  "go.temporal.io/server/service/history/consts"
- "go.temporal.io/server/service/history/shard"
+ "go.temporal.io/server/service/history/definition"
  "go.temporal.io/server/service/history/vclock"
- "go.temporal.io/server/service/history/workflow"
 )
 
 type (
- MutableStateConsistencyPredicate func(mutableState workflow.MutableState) bool
+ MutableStateConsistencyPredicate func(mutableState definition.MutableState) bool
 
  WorkflowConsistencyChecker interface {
- GetWorkflowCache() workflow.Cache
+ GetWorkflowCache() definition.WorkflowCache
  GetCurrentRunID(
  ctx context.Context,
  namespaceID string,
@@ -62,22 +60,22 @@ type (
  }
 
  WorkflowConsistencyCheckerImpl struct {
- shardContext shard.Context
- workflowCache workflow.Cache
+ shardContext definition.ShardContext
+ workflowCache definition.WorkflowCache
  }
 )
 
 func NewWorkflowConsistencyChecker(
- shardContext shard.Context,
- workflowCache workflow.Cache,
+ shardContext definition.ShardContext,
+ workflowCache definition.WorkflowCache,
 ) *WorkflowConsistencyCheckerImpl {
  return &WorkflowConsistencyCheckerImpl{
  shardContext: shardContext,
  workflowCache: workflowCache,
  }
 }
 
-func (c *WorkflowConsistencyCheckerImpl) GetWorkflowCache() workflow.Cache {
+func (c *WorkflowConsistencyCheckerImpl) GetWorkflowCache() definition.WorkflowCache {
  return c.workflowCache
 }
 
@@ -169,7 +167,7 @@ func (c *WorkflowConsistencyCheckerImpl) getWorkflowContextValidatedByClock(
  WorkflowId: workflowKey.WorkflowID,
  RunId: workflowKey.RunID,
  },
- workflow.CallerTypeAPI,
+ definition.CallerTypeAPI,
  )
  if err != nil {
  return nil, err
@@ -202,7 +200,7 @@ func (c *WorkflowConsistencyCheckerImpl) getWorkflowContextValidatedByCheck(
  WorkflowId: workflowKey.WorkflowID,
  RunId: workflowKey.RunID,
  },
- workflow.CallerTypeAPI,
+ definition.CallerTypeAPI,
  )
  if err != nil {
  return nil, err
@@ -318,7 +316,7 @@ func (c *WorkflowConsistencyCheckerImpl) getCurrentRunID(
 
 func assertShardOwnership(
  ctx context.Context,
- shardContext shard.Context,
+ shardContext definition.ShardContext,
  shardOwnershipAsserted *bool,
 ) error {
  if !*shardOwnershipAsserted {
@@ -329,13 +327,13 @@ func assertShardOwnership(
 }
 
 func BypassMutableStateConsistencyPredicate(
- mutableState workflow.MutableState,
+ mutableState definition.MutableState,
 ) bool {
  return true
 }
 
 func FailMutableStateConsistencyPredicate(
- mutableState workflow.MutableState,
+ mutableState definition.MutableState,
 ) bool {
  return false
 }
@@ -344,7 +342,7 @@ func HistoryEventConsistencyPredicate(
  eventID int64,
  eventVersion int64,
 ) MutableStateConsistencyPredicate {
- return func(mutableState workflow.MutableState) bool {
+ return func(mutableState definition.MutableState) bool {
  if eventVersion != 0 {
  _, err := versionhistory.FindFirstVersionHistoryIndexByVersionHistoryItem(
  mutableState.GetExecutionInfo().GetVersionHistories(),

service/history/api/consistency_checker_test.go

@@ -39,12 +39,10 @@ import (
 
  historyspb "go.temporal.io/server/api/history/v1"
  persistencespb "go.temporal.io/server/api/persistence/v1"
- "go.temporal.io/server/common/definition"
  "go.temporal.io/server/common/namespace"
  "go.temporal.io/server/common/persistence"
  "go.temporal.io/server/common/persistence/versionhistory"
- "go.temporal.io/server/service/history/shard"
- "go.temporal.io/server/service/history/workflow"
+ "go.temporal.io/server/service/history/definition"
 )
 
 type (
@@ -53,8 +51,8 @@ type (
  *require.Assertions
 
  controller *gomock.Controller
- shardContext *shard.MockContext
- workflowCache *workflow.MockCache
+ shardContext *definition.MockShardContext
+ workflowCache *definition.MockWorkflowCache
 
  shardID int32
  namespaceID string
@@ -81,8 +79,8 @@ func (s *workflowConsistencyCheckerSuite) SetupTest() {
  s.Assertions = require.New(s.T())
 
  s.controller = gomock.NewController(s.T())
- s.shardContext = shard.NewMockContext(s.controller)
- s.workflowCache = workflow.NewMockCache(s.controller)
+ s.shardContext = definition.NewMockShardContext(s.controller)
+ s.workflowCache = definition.NewMockWorkflowCache(s.controller)
 
  s.shardID = rand.Int31()
  s.namespaceID = uuid.New().String()
@@ -102,8 +100,8 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  ctx := context.Background()
  shardOwnershipAsserted := false
 
- wfContext := workflow.NewMockContext(s.controller)
- mutableState := workflow.NewMockMutableState(s.controller)
+ wfContext := definition.NewMockWorkflowContext(s.controller)
+ mutableState := definition.NewMockMutableState(s.controller)
  released := false
  releaseFn := func(err error) { released = true }
 
@@ -114,7 +112,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  WorkflowId: s.workflowID,
  RunId: s.currentRunID,
  },
- workflow.CallerTypeAPI,
+ definition.CallerTypeAPI,
  ).Return(wfContext, releaseFn, nil)
  wfContext.EXPECT().LoadMutableState(ctx).Return(mutableState, nil)
 
@@ -133,9 +131,9 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  ctx := context.Background()
  shardOwnershipAsserted := false
 
- wfContext := workflow.NewMockContext(s.controller)
- mutableState1 := workflow.NewMockMutableState(s.controller)
- mutableState2 := workflow.NewMockMutableState(s.controller)
+ wfContext := definition.NewMockWorkflowContext(s.controller)
+ mutableState1 := definition.NewMockMutableState(s.controller)
+ mutableState2 := definition.NewMockMutableState(s.controller)
  released := false
  releaseFn := func(err error) { released = true }
 
@@ -146,7 +144,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  WorkflowId: s.workflowID,
  RunId: s.currentRunID,
  },
- workflow.CallerTypeAPI,
+ definition.CallerTypeAPI,
  ).Return(wfContext, releaseFn, nil)
  gomock.InOrder(
  wfContext.EXPECT().LoadMutableState(ctx).Return(mutableState1, nil),
@@ -169,7 +167,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  ctx := context.Background()
  shardOwnershipAsserted := false
 
- wfContext := workflow.NewMockContext(s.controller)
+ wfContext := definition.NewMockWorkflowContext(s.controller)
  released := false
  releaseFn := func(err error) { released = true }
 
@@ -180,7 +178,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  WorkflowId: s.workflowID,
  RunId: s.currentRunID,
  },
- workflow.CallerTypeAPI,
+ definition.CallerTypeAPI,
  ).Return(wfContext, releaseFn, nil)
  wfContext.EXPECT().LoadMutableState(ctx).Return(nil, serviceerror.NewNotFound(""))
 
@@ -201,7 +199,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  ctx := context.Background()
  shardOwnershipAsserted := false
 
- wfContext := workflow.NewMockContext(s.controller)
+ wfContext := definition.NewMockWorkflowContext(s.controller)
  released := false
  releaseFn := func(err error) { released = true }
 
@@ -212,7 +210,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  WorkflowId: s.workflowID,
  RunId: s.currentRunID,
  },
- workflow.CallerTypeAPI,
+ definition.CallerTypeAPI,
  ).Return(wfContext, releaseFn, nil)
  wfContext.EXPECT().LoadMutableState(ctx).Return(nil, serviceerror.NewNotFound(""))
 
@@ -233,7 +231,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  ctx := context.Background()
  shardOwnershipAsserted := false
 
- wfContext := workflow.NewMockContext(s.controller)
+ wfContext := definition.NewMockWorkflowContext(s.controller)
  released := false
  releaseFn := func(err error) { released = true }
 
@@ -244,7 +242,7 @@ func (s *workflowConsistencyCheckerSuite) TestGetWorkflowContextValidatedByCheck
  WorkflowId: s.workflowID,
  RunId: s.currentRunID,
  },
- workflow.CallerTypeAPI,
+ definition.CallerTypeAPI,
  ).Return(wfContext, releaseFn, nil)
  wfContext.EXPECT().LoadMutableState(ctx).Return(nil, serviceerror.NewUnavailable(""))
 
@@ -412,7 +410,7 @@ func (s *workflowConsistencyCheckerSuite) TestHistoryEventConsistencyPredicate()
 
  for _, tc := range testCases {
  s.Run(tc.name, func() {
- mockMutableState := workflow.NewMockMutableState(s.controller)
+ mockMutableState := definition.NewMockMutableState(s.controller)
  mockMutableState.EXPECT().GetExecutionInfo().Return(&persistencespb.WorkflowExecutionInfo{
  VersionHistories: tc.versionHistories,
  })

service/history/api/create_workflow_util.go

@@ -36,13 +36,12 @@ import (
  "go.temporal.io/server/api/historyservice/v1"
  workflowspb "go.temporal.io/server/api/workflow/v1"
  "go.temporal.io/server/common"
- "go.temporal.io/server/common/definition"
  "go.temporal.io/server/common/log/tag"
  "go.temporal.io/server/common/metrics"
  "go.temporal.io/server/common/namespace"
  "go.temporal.io/server/common/primitives/timestamp"
  "go.temporal.io/server/common/rpc/interceptor"
- "go.temporal.io/server/service/history/shard"
+ "go.temporal.io/server/service/history/definition"
  "go.temporal.io/server/service/history/workflow"
 )
 
@@ -55,7 +54,7 @@ type (
 
 func NewWorkflowWithSignal(
  ctx context.Context,
- shard shard.Context,
+ shard definition.ShardContext,
  namespaceEntry *namespace.Namespace,
  workflowID string,
  runID string,
@@ -122,12 +121,12 @@ func NewWorkflowWithSignal(
 
 func CreateMutableState(
  ctx context.Context,
- shard shard.Context,
+ shard definition.ShardContext,
  namespaceEntry *namespace.Namespace,
  executionTimeout *time.Duration,
  runTimeout *time.Duration,
  runID string,
-) (workflow.MutableState, error) {
+) (definition.MutableState, error) {
  newMutableState := workflow.NewMutableState(
  shard,
  shard.GetEventsCache(),
@@ -142,7 +141,7 @@ func CreateMutableState(
 }
 
 func GenerateFirstWorkflowTask(
- mutableState workflow.MutableState,
+ mutableState definition.MutableState,
  parentInfo *workflowspb.ParentExecutionInfo,
  startEvent *historypb.HistoryEvent,
 ) error {
@@ -159,9 +158,9 @@ func GenerateFirstWorkflowTask(
 }
 
 func NewWorkflowVersionCheck(
- shard shard.Context,
+ shard definition.ShardContext,
  prevLastWriteVersion int64,
- newMutableState workflow.MutableState,
+ newMutableState definition.MutableState,
 ) error {
  if prevLastWriteVersion == common.EmptyVersion {
  return nil
@@ -181,7 +180,7 @@ func NewWorkflowVersionCheck(
 
 func ValidateStart(
  ctx context.Context,
- shard shard.Context,
+ shard definition.ShardContext,
  namespaceEntry *namespace.Namespace,
  workflowID string,
  workflowInputSize int,
@@ -229,7 +228,7 @@ func ValidateStart(
 func ValidateStartWorkflowExecutionRequest(
  ctx context.Context,
  request *workflowservice.StartWorkflowExecutionRequest,
- shard shard.Context,
+ shard definition.ShardContext,
  namespaceEntry *namespace.Namespace,
  operation string,
 ) error {
@@ -289,7 +288,7 @@ func ValidateStartWorkflowExecutionRequest(
 func OverrideStartWorkflowExecutionRequest(
  request *workflowservice.StartWorkflowExecutionRequest,
  operation string,
- shard shard.Context,
+ shard definition.ShardContext,
  metricsHandler metrics.MetricsHandler,
 ) {
  // workflow execution timeout is left as is

service/history/api/deleteworkflow/api.go

@@ -30,18 +30,17 @@ import (
  commonpb "go.temporal.io/api/common/v1"
 
  "go.temporal.io/server/api/historyservice/v1"
- "go.temporal.io/server/common/definition"
  "go.temporal.io/server/common/namespace"
  "go.temporal.io/server/service/history/api"
  "go.temporal.io/server/service/history/consts"
- "go.temporal.io/server/service/history/shard"
+ "go.temporal.io/server/service/history/definition"
  "go.temporal.io/server/service/history/workflow"
 )
 
 func Invoke(
  ctx context.Context,
  request *historyservice.DeleteWorkflowExecutionRequest,
- shard shard.Context,
+ shard definition.ShardContext,
  workflowConsistencyChecker api.WorkflowConsistencyChecker,
  workflowDeleteManager workflow.DeleteManager,
 ) (_ *historyservice.DeleteWorkflowExecutionResponse, retError error) {