Pick simplify retry changes

common/persistence/client/factory.go

@@ -25,6 +25,8 @@
 package client
 
 import (
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/config"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
@@ -33,6 +35,10 @@ import (
 	"go.temporal.io/server/common/quotas"
 )
 
+var (
+	retryPolicy = common.CreatePersistenceClientRetryPolicy()
+)
+
 type (
 	// Factory defines the interface for any implementation that can vend
 	// persistence layer objects backed by a datastore. The actual datastore
@@ -123,6 +129,7 @@ func (f *factoryImpl) NewShardManager() (p.ShardManager, error) {
 	if f.metricsClient != nil {
 		result = p.NewShardPersistenceMetricsClient(result, f.metricsClient, f.logger)
 	}
+	result = p.NewShardPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
 }
 
@@ -140,6 +147,7 @@ func (f *factoryImpl) NewMetadataManager() (p.MetadataManager, error) {
 	if f.metricsClient != nil {
 		result = p.NewMetadataPersistenceMetricsClient(result, f.metricsClient, f.logger)
 	}
+	result = p.NewMetadataPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
 }
 
@@ -157,6 +165,7 @@ func (f *factoryImpl) NewClusterMetadataManager() (p.ClusterMetadataManager, err
 	if f.metricsClient != nil {
 		result = p.NewClusterMetadataPersistenceMetricsClient(result, f.metricsClient, f.logger)
 	}
+	result = p.NewClusterMetadataPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
 }
 
@@ -174,6 +183,7 @@ func (f *factoryImpl) NewExecutionManager() (p.ExecutionManager, error) {
 	if f.metricsClient != nil {
 		result = p.NewExecutionPersistenceMetricsClient(result, f.metricsClient, f.logger)
 	}
+	result = p.NewExecutionPersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return result, nil
 }
 
@@ -189,11 +199,20 @@ func (f *factoryImpl) NewNamespaceReplicationQueue() (p.NamespaceReplicationQueu
 	if f.metricsClient != nil {
 		result = p.NewQueuePersistenceMetricsClient(result, f.metricsClient, f.logger)
 	}
-
+	result = p.NewQueuePersistenceRetryableClient(result, retryPolicy, IsPersistenceTransientError)
 	return p.NewNamespaceReplicationQueue(result, f.serializer, f.clusterName, f.metricsClient, f.logger)
 }
 
 // Close closes this factory
 func (f *factoryImpl) Close() {
 	f.dataStoreFactory.Close()
 }
+
+func IsPersistenceTransientError(err error) bool {
+	switch err.(type) {
+	case *serviceerror.Unavailable:
+		return true
+	}
+
+	return false
+}

common/persistence/dataInterfaces.go

@@ -1159,6 +1159,16 @@ func (e *TransactionSizeLimitError) Error() string {
 	return e.Msg
 }
 
+func IsConflictErr(err error) bool {
+	switch err.(type) {
+	case *CurrentWorkflowConditionFailedError,
+		*WorkflowConditionFailedError,
+		*ConditionFailedError:
+		return true
+	}
+	return false
+}
+
 // UnixMilliseconds returns t as a Unix time, the number of milliseconds elapsed since January 1, 1970 UTC.
 // It should be used for all CQL timestamp.
 func UnixMilliseconds(t time.Time) int64 {