Enable the workflow ID cache in shadow mode for start workflow

service/history/handler.go

@@ -53,9 +53,14 @@ import (
 	"github.com/uber/cadence/service/history/resource"
 	"github.com/uber/cadence/service/history/shard"
 	"github.com/uber/cadence/service/history/task"
+	"github.com/uber/cadence/service/history/workflowcache"
 )
 
-const shardOwnershipTransferDelay = 5 * time.Second
+const (
+	shardOwnershipTransferDelay = 5 * time.Second
+	workflowIDCacheTTL          = 1 * time.Second
+	workflowIDCacheMaxCount     = 10_000
+)
 
 type (
 	// handlerImpl is an implementation for history service independent of wire protocol
@@ -73,6 +78,7 @@ type (
 		replicationTaskFetchers  replication.TaskFetchers
 		queueTaskProcessor       task.Processor
 		failoverCoordinator      failover.Coordinator
+		workflowIDCache          workflowcache.WFCache
 	}
 )
 
@@ -102,6 +108,16 @@ func NewHandler(
 		config:          config,
 		tokenSerializer: common.NewJSONTaskTokenSerializer(),
 		rateLimiter:     quotas.NewDynamicRateLimiter(config.RPS.AsFloat64()),
+		workflowIDCache: workflowcache.New(workflowcache.Params{
+			TTL:                    workflowIDCacheTTL,
+			ExternalLimiterFactory: quotas.NewSimpleDynamicRateLimiterFactory(config.WorkflowIDExternalRPS),
+			InternalLimiterFactory: quotas.NewSimpleDynamicRateLimiterFactory(config.WorkflowIDInternalRPS),
+			WorkflowIDCacheEnabled: config.WorkflowIDCacheEnabled,
+			MaxCount:               workflowIDCacheMaxCount,
+			DomainCache:            resource.GetDomainCache(),
+			Logger:                 resource.GetLogger(),
+			MetricsClient:          resource.GetMetricsClient(),
+		}),
 	}
 
 	// prevent us from trying to serve requests before shard controller is started and ready
@@ -681,6 +697,10 @@ func (h *handlerImpl) StartWorkflowExecution(
 
 	startRequest := wrappedRequest.StartRequest
 	workflowID := startRequest.GetWorkflowID()
+
+	// TODO: actually rate limit
+	h.workflowIDCache.AllowExternal(domainID, workflowID)
+
 	engine, err1 := h.controller.GetEngine(workflowID)
 	if err1 != nil {
 		return nil, h.error(err1, scope, domainID, workflowID, "")

service/history/handler_test.go

@@ -42,6 +42,13 @@ import (
 	"github.com/uber/cadence/service/history/engine"
 	"github.com/uber/cadence/service/history/resource"
 	"github.com/uber/cadence/service/history/shard"
+	"github.com/uber/cadence/service/history/workflowcache"
+)
+
+const (
+	testWorkflowID    = "test-workflow-id"
+	testWorkflowRunID = "test-workflow-run-id"
+	testDomainID      = "BF80C53A-ED56-4DD9-84EB-BE9AD4E45867"
 )
 
 type (
@@ -53,6 +60,7 @@ type (
 		mockResource        *resource.Test
 		mockShardController *shard.MockController
 		mockEngine          *engine.MockEngine
+		mockWFCache         *workflowcache.MockWFCache
 
 		handler *handlerImpl
 	}
@@ -72,9 +80,11 @@ func (s *handlerSuite) SetupTest() {
 	s.mockShardController = shard.NewMockController(s.controller)
 	s.mockEngine = engine.NewMockEngine(s.controller)
 	s.mockShardController.EXPECT().GetEngineForShard(gomock.Any()).Return(s.mockEngine, nil).AnyTimes()
+	s.mockWFCache = workflowcache.NewMockWFCache(s.controller)
 
 	s.handler = NewHandler(s.mockResource, config.NewForTest()).(*handlerImpl)
 	s.handler.controller = s.mockShardController
+	s.handler.workflowIDCache = s.mockWFCache
 	s.handler.startWG.Done()
 }
 
@@ -294,3 +304,36 @@ func TestCorrectUseOfErrorHandling(t *testing.T) {
 		})
 	}
 }
+
+func (s *handlerSuite) TestStartWorkflowExecution() {
+
+	request := &types.HistoryStartWorkflowExecutionRequest{
+		DomainUUID: testDomainID,
+		StartRequest: &types.StartWorkflowExecutionRequest{
+			WorkflowID: testWorkflowID,
+		},
+	}
+
+	expectedResponse := &types.StartWorkflowExecutionResponse{
+		RunID: testWorkflowRunID,
+	}
+
+	// We should _always_ see the startworkflowexecution call no matter if allow external is true or false
+	// as we are in shadow mode
+	tests := map[string]struct{ allowExternal bool }{
+		"allow external":    {allowExternal: true},
+		"disallow external": {allowExternal: false},
+	}
+
+	for name, test := range tests {
+		s.Run(name, func() {
+			s.mockWFCache.EXPECT().AllowExternal(gomock.Any(), gomock.Any()).Return(test.allowExternal).Times(1)
+			s.mockShardController.EXPECT().GetEngine(testWorkflowID).Return(s.mockEngine, nil).AnyTimes()
+			s.mockEngine.EXPECT().StartWorkflowExecution(gomock.Any(), gomock.Any()).Return(expectedResponse, nil).Times(1)
+
+			response, err := s.handler.StartWorkflowExecution(context.Background(), request)
+			s.Equal(expectedResponse, response)
+			s.Nil(err)
+		})
+	}
+}

service/history/workflowcache/cache.go

@@ -20,6 +20,8 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 // SOFTWARE.
 
+//go:generate mockgen -package=$GOPACKAGE -destination=cache_mock.go github.com/uber/cadence/service/history/workflowcache WFCache
+
 package workflowcache
 
 import (