Handle standby side workflow timeout & missing events (cadence-workfl…

…ow#1356)

* Standby logic shall try fetch history before discarding task if task pending for too long

client/admin/retryableClient.go

@@ -37,6 +37,15 @@ type retryableClient struct {
 	isRetryable backoff.IsRetryable
 }
 
+// NewRetryableClient creates a new instance of Client with retry policy
+func NewRetryableClient(client Client, policy backoff.RetryPolicy, isRetryable backoff.IsRetryable) Client {
+	return &retryableClient{
+		client:      client,
+		policy:      policy,
+		isRetryable: isRetryable,
+	}
+}
+
 func (c *retryableClient) DescribeHistoryHost(
 	ctx context.Context,
 	request *shared.DescribeHistoryHostRequest,

client/clientBean_mock.go

@@ -0,0 +1,116 @@
+// Copyright (c) 2017 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 client
+
+import (
+	"github.com/stretchr/testify/mock"
+	"github.com/uber/cadence/client/admin"
+	"github.com/uber/cadence/client/frontend"
+	"github.com/uber/cadence/client/history"
+	"github.com/uber/cadence/client/matching"
+)
+
+// MockClientBean is an autogenerated mock type for the MockClientBean type
+type MockClientBean struct {
+	mock.Mock
+}
+
+var _ Bean = (*MockClientBean)(nil)
+
+// GetHistoryClient provides a mock function with given fields:
+func (_m *MockClientBean) GetHistoryClient() history.Client {
+	ret := _m.Called()
+
+	var r0 history.Client
+	if rf, ok := ret.Get(0).(func() history.Client); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(history.Client)
+		}
+	}
+
+	return r0
+}
+
+// GetMatchingClient provides a mock function with given fields:
+func (_m *MockClientBean) GetMatchingClient() matching.Client {
+	ret := _m.Called()
+
+	var r0 matching.Client
+	if rf, ok := ret.Get(0).(func() matching.Client); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(matching.Client)
+		}
+	}
+
+	return r0
+}
+
+// GetFrontendClient provides a mock function with given fields:
+func (_m *MockClientBean) GetFrontendClient() frontend.Client {
+	ret := _m.Called()
+
+	var r0 frontend.Client
+	if rf, ok := ret.Get(0).(func() frontend.Client); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(frontend.Client)
+		}
+	}
+
+	return r0
+}
+
+// GetRemoteAdminClient provides a mock function with given fields: _a0
+func (_m *MockClientBean) GetRemoteAdminClient(_a0 string) admin.Client {
+	ret := _m.Called(_a0)
+
+	var r0 admin.Client
+	if rf, ok := ret.Get(0).(func(string) admin.Client); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(admin.Client)
+		}
+	}
+
+	return r0
+}
+
+// GetRemoteFrontendClient provides a mock function with given fields: _a0
+func (_m *MockClientBean) GetRemoteFrontendClient(_a0 string) frontend.Client {
+	ret := _m.Called(_a0)
+
+	var r0 frontend.Client
+	if rf, ok := ret.Get(0).(func(string) frontend.Client); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(frontend.Client)
+		}
+	}
+
+	return r0
+}

common/cluster/metadataTestBase.go

@@ -51,6 +51,22 @@ var (
 		TestCurrentClusterName:     TestCurrentClusterInitialFailoverVersion,
 		TestAlternativeClusterName: TestAlternativeClusterInitialFailoverVersion,
 	}
+	// TestAllClusterAddress is the same as above, juse convinent for test mocking
+	TestAllClusterAddress = map[string]config.Address{
+		TestCurrentClusterName:     config.Address{RPCName: common.FrontendServiceName, RPCAddress: TestCurrentClusterFrontendAddress},
+		TestAlternativeClusterName: config.Address{RPCName: common.FrontendServiceName, RPCAddress: TestAlternativeClusterFrontendAddress},
+	}
+
+	// TestSingleDCAllClusterNames is the all cluster names used for test
+	TestSingleDCAllClusterNames = []string{TestCurrentClusterName}
+	// TestSingleDCAllClusterFailoverVersions is the same as above, juse convinent for test mocking
+	TestSingleDCAllClusterFailoverVersions = map[string]int64{
+		TestCurrentClusterName: TestCurrentClusterInitialFailoverVersion,
+	}
+	// TestSingleDCAllClusterAddress is the same as above, juse convinent for test mocking
+	TestSingleDCAllClusterAddress = map[string]config.Address{
+		TestCurrentClusterName: config.Address{RPCName: common.FrontendServiceName, RPCAddress: TestCurrentClusterFrontendAddress},
+	}
 )
 
 // GetTestClusterMetadata return an cluster metadata instance, which is initialized
@@ -59,16 +75,27 @@ func GetTestClusterMetadata(enableGlobalDomain bool, isMasterCluster bool) Metad
 	if !isMasterCluster {
 		masterClusterName = TestAlternativeClusterName
 	}
+
+	if enableGlobalDomain {
+		return NewMetadata(
+			dynamicconfig.GetBoolPropertyFn(true),
+			TestFailoverVersionIncrement,
+			masterClusterName,
+			TestCurrentClusterName,
+			TestAllClusterFailoverVersions,
+			TestAllClusterAddress,
+			dynamicconfig.GetBoolPropertyFn(false),
+			"",
+		)
+	}
+
 	return NewMetadata(
-		dynamicconfig.GetBoolPropertyFn(enableGlobalDomain),
+		dynamicconfig.GetBoolPropertyFn(false),
 		TestFailoverVersionIncrement,
-		masterClusterName,
 		TestCurrentClusterName,
-		TestAllClusterFailoverVersions,
-		map[string]config.Address{
-			TestCurrentClusterName:     config.Address{RPCName: common.FrontendServiceName, RPCAddress: TestCurrentClusterFrontendAddress},
-			TestAlternativeClusterName: config.Address{RPCName: common.FrontendServiceName, RPCAddress: TestAlternativeClusterFrontendAddress},
-		},
+		TestCurrentClusterName,
+		TestSingleDCAllClusterFailoverVersions,
+		TestSingleDCAllClusterAddress,
 		dynamicconfig.GetBoolPropertyFn(false),
 		"",
 	)

common/service/service.go

@@ -89,7 +89,6 @@ type (
 		membershipMonitor      membership.Monitor
 		rpcFactory             common.RPCFactory
 		pprofInitializer       common.PProfInitializer
-		clientFactory          client.Factory
 		clientBean             client.Bean
 		numberOfHistoryShards  int
 		logger                 bark.Logger
@@ -192,10 +191,11 @@ func (h *serviceImpl) Start() {
 	}
 	h.hostInfo = hostInfo
 
-	h.clientFactory = client.NewRPCClientFactory(h.rpcFactory, h.membershipMonitor, h.metricsClient,
-		h.numberOfHistoryShards)
-
-	h.clientBean, err = client.NewClientBean(h.clientFactory, h.dispatcherProvider, h.clusterMetadata)
+	h.clientBean, err = client.NewClientBean(
+		client.NewRPCClientFactory(h.rpcFactory, h.membershipMonitor, h.metricsClient, h.numberOfHistoryShards),
+		h.dispatcherProvider,
+		h.clusterMetadata,
+	)
 	if err != nil {
 		h.logger.WithFields(bark.Fields{logging.TagErr: err}).Fatal("fail to initialize client bean")
 	}
@@ -237,10 +237,6 @@ func (h *serviceImpl) GetMetricsClient() metrics.Client {
 	return h.metricsClient
 }
 
-func (h *serviceImpl) GetClientFactory() client.Factory {
-	return h.clientFactory
-}
-
 func (h *serviceImpl) GetClientBean() client.Bean {
 	return h.clientBean
 }