fix(shard-distributor): add error handling in namespace refresh loop (cadence-workflow#7519)

<!-- Describe what has changed in this PR -->
**What changed?**
* Error handling was added to `namespaceRefreshLoop`


<!-- Tell your future self why have you made these changes -->
**Why?**
etcd `Watch` method may close `WatchChannel` in case of non-recoverable
errors like `ErrCompaction`, according to [the
doc](https://github.com/etcd-io/etcd/blob/main/client/v3/watch.go#L65-L67).
`namespaceRefreshLoop` is used to update internal cache. However, the
code didn't catch errors in case of a closed channel or `WatchResponse`
containing errors. It led to a busy loop, because the loop is never
broken in case of these cases, and may cause a high CPU problem observed
in the staging environment.

<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
**How did you test it?**
* Unit test
* Local run with canary
* Integration test

<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
**Potential risks**
* There is an interval of 150ms between retries that may cause a cache
invalidation for 150ms.

<!-- Is it notable for release? e.g. schema updates, configuration or
data migration required? If so, please mention it, and also update
CHANGELOG.md -->
**Release notes**

<!-- Is there any documentation updates should be made for config,
https://cadenceworkflow.io/docs/operation-guide/setup/ ? If so, please
open an PR in https://github.com/cadence-workflow/cadence-docs -->
**Documentation Changes**

Author: arzonus

service/sharddistributor/store/etcd/executorstore/etcdstore.go

@@ -57,7 +57,7 @@ type ExecutorStoreParams struct {
 
 // NewStore creates a new etcd-backed store and provides it to the fx application.
 func NewStore(p ExecutorStoreParams) (store.Store, error) {
-	shardCache := shardcache.NewShardToExecutorCache(p.Cfg.Prefix, p.Client, p.Logger)
+	shardCache := shardcache.NewShardToExecutorCache(p.Cfg.Prefix, p.Client, p.Logger, p.TimeSource)
 
 	timeSource := p.TimeSource
 	if timeSource == nil {

service/sharddistributor/store/etcd/executorstore/etcdstore_test.go

@@ -12,6 +12,7 @@ import (
 	"go.uber.org/fx/fxtest"
 	"gopkg.in/yaml.v2"
 
+	"github.com/uber/cadence/common/clock"
 	"github.com/uber/cadence/common/config"
 	"github.com/uber/cadence/common/log/testlogger"
 	"github.com/uber/cadence/common/types"
@@ -713,10 +714,11 @@ func createStore(t *testing.T, tc *testhelper.StoreTestCluster) store.Store {
 	require.NoError(t, err)
 
 	store, err := NewStore(ExecutorStoreParams{
-		Client:    tc.Client,
-		Cfg:       etcdConfig,
-		Lifecycle: fxtest.NewLifecycle(t),
-		Logger:    testlogger.New(t),
+		Client:     tc.Client,
+		Cfg:        etcdConfig,
+		Lifecycle:  fxtest.NewLifecycle(t),
+		Logger:     testlogger.New(t),
+		TimeSource: clock.NewRealTimeSource(),
 	})
 	require.NoError(t, err)
 	return store

service/sharddistributor/store/etcd/executorstore/shardcache/namespaceshardcache.go

@@ -5,9 +5,12 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
 	clientv3 "go.etcd.io/etcd/client/v3"
 
+	"github.com/uber/cadence/common/backoff"
+	"github.com/uber/cadence/common/clock"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/tag"
 	"github.com/uber/cadence/service/sharddistributor/store"
@@ -17,47 +20,49 @@ import (
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/executorstore/common"
 )
 
+const (
+	// RetryInterval for watch failures is between 50ms to 150ms
+	namespaceRefreshLoopWatchJitterCoeff   = 0.5
+	namespaceRefreshLoopWatchRetryInterval = 100 * time.Millisecond
+)
+
 type namespaceShardToExecutor struct {
 	sync.RWMutex
 
-	shardToExecutor     map[string]*store.ShardOwner   // shardID -> shardOwner
-	shardOwners         map[string]*store.ShardOwner   // executorID -> shardOwner
-	executorState       map[*store.ShardOwner][]string // executor -> shardIDs
-	executorRevision    map[string]int64
-	namespace           string
-	etcdPrefix          string
-	changeUpdateChannel clientv3.WatchChan
-	stopCh              chan struct{}
-	logger              log.Logger
-	client              etcdclient.Client
-	pubSub              *executorStatePubSub
+	shardToExecutor  map[string]*store.ShardOwner   // shardID -> shardOwner
+	shardOwners      map[string]*store.ShardOwner   // executorID -> shardOwner
+	executorState    map[*store.ShardOwner][]string // executor -> shardIDs
+	executorRevision map[string]int64
+	namespace        string
+	etcdPrefix       string
+	stopCh           chan struct{}
+	logger           log.Logger
+	client           etcdclient.Client
+	timeSource       clock.TimeSource
+	pubSub           *executorStatePubSub
 }
 
-func newNamespaceShardToExecutor(etcdPrefix, namespace string, client etcdclient.Client, stopCh chan struct{}, logger log.Logger) (*namespaceShardToExecutor, error) {
-	// Start listening
-	watchPrefix := etcdkeys.BuildExecutorsPrefix(etcdPrefix, namespace)
-	watchChan := client.Watch(context.Background(), watchPrefix, clientv3.WithPrefix(), clientv3.WithPrevKV())
-
+func newNamespaceShardToExecutor(etcdPrefix, namespace string, client etcdclient.Client, stopCh chan struct{}, logger log.Logger, timeSource clock.TimeSource) (*namespaceShardToExecutor, error) {
 	return &namespaceShardToExecutor{
-		shardToExecutor:     make(map[string]*store.ShardOwner),
-		executorState:       make(map[*store.ShardOwner][]string),
-		executorRevision:    make(map[string]int64),
-		shardOwners:         make(map[string]*store.ShardOwner),
-		namespace:           namespace,
-		etcdPrefix:          etcdPrefix,
-		changeUpdateChannel: watchChan,
-		stopCh:              stopCh,
-		logger:              logger,
-		client:              client,
-		pubSub:              newExecutorStatePubSub(logger, namespace),
+		shardToExecutor:  make(map[string]*store.ShardOwner),
+		executorState:    make(map[*store.ShardOwner][]string),
+		executorRevision: make(map[string]int64),
+		shardOwners:      make(map[string]*store.ShardOwner),
+		namespace:        namespace,
+		etcdPrefix:       etcdPrefix,
+		stopCh:           stopCh,
+		logger:           logger.WithTags(tag.ShardNamespace(namespace)),
+		client:           client,
+		timeSource:       timeSource,
+		pubSub:           newExecutorStatePubSub(logger, namespace),
 	}, nil
 }
 
 func (n *namespaceShardToExecutor) Start(wg *sync.WaitGroup) {
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		n.nameSpaceRefreashLoop()
+		n.namespaceRefreshLoop()
 	}()
 }
 
@@ -106,22 +111,56 @@ func (n *namespaceShardToExecutor) Subscribe(ctx context.Context) (<-chan map[*s
 
 		select {
 		case <-ctx.Done():
-			n.logger.Warn("context finnished before initial state was sent", tag.ShardNamespace(n.namespace))
+			n.logger.Warn("context finished before initial state was sent")
 		case subCh <- initialState:
-			n.logger.Info("initial state sent to subscriber", tag.ShardNamespace(n.namespace), tag.Value(initialState))
+			n.logger.Info("initial state sent to subscriber", tag.Value(initialState))
 		}
 
 	}()
 
 	return subCh, unSub
 }
 
-func (n *namespaceShardToExecutor) nameSpaceRefreashLoop() {
+func (n *namespaceShardToExecutor) namespaceRefreshLoop() {
+	for {
+		if err := n.watch(); err != nil {
+			n.logger.Error("error watching in namespaceRefreshLoop, retrying...", tag.Error(err))
+			n.timeSource.Sleep(backoff.JitDuration(
+				namespaceRefreshLoopWatchRetryInterval,
+				namespaceRefreshLoopWatchJitterCoeff,
+			))
+			continue
+		}
+
+		n.logger.Info("namespaceRefreshLoop is exiting")
+		return
+	}
+}
+
+func (n *namespaceShardToExecutor) watch() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	watchChan := n.client.Watch(
+		// WithRequireLeader ensures that the etcd cluster has a leader
+		clientv3.WithRequireLeader(ctx),
+		etcdkeys.BuildExecutorsPrefix(n.etcdPrefix, n.namespace),
+		clientv3.WithPrefix(), clientv3.WithPrevKV(),
+	)
+
 	for {
 		select {
 		case <-n.stopCh:
-			return
-		case watchResp := <-n.changeUpdateChannel:
+			return nil
+
+		case watchResp, ok := <-watchChan:
+			if err := watchResp.Err(); err != nil {
+				return fmt.Errorf("watch response: %w", err)
+			}
+			if !ok {
+				return fmt.Errorf("watch channel closed")
+			}
+
 			shouldRefresh := false
 			for _, event := range watchResp.Events {
 				_, keyType, keyErr := etcdkeys.ParseExecutorKey(n.etcdPrefix, n.namespace, string(event.Kv.Key))
@@ -134,13 +173,13 @@ func (n *namespaceShardToExecutor) nameSpaceRefreashLoop() {
 					break
 				}
 			}
+
 			if shouldRefresh {
 				err := n.refresh(context.Background())
 				if err != nil {
-					n.logger.Error("failed to refresh namespace shard to executor", tag.ShardNamespace(n.namespace), tag.Error(err))
+					n.logger.Error("failed to refresh namespace shard to executor", tag.Error(err))
 				}
 			}
-
 		}
 	}
 }

service/sharddistributor/store/etcd/executorstore/shardcache/namespaceshardcache_test.go

@@ -4,16 +4,21 @@ import (
 	"context"
 	"encoding/json"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.uber.org/goleak"
+	"go.uber.org/mock/gomock"
 
+	"github.com/uber/cadence/common/clock"
 	"github.com/uber/cadence/common/log/testlogger"
 	"github.com/uber/cadence/common/types"
 	"github.com/uber/cadence/service/sharddistributor/store"
+	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdclient"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdkeys"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/etcdtypes"
 	"github.com/uber/cadence/service/sharddistributor/store/etcd/testhelper"
@@ -32,7 +37,7 @@ func TestNamespaceShardToExecutor_Lifecycle(t *testing.T) {
 	})
 
 	// Start the cache
-	namespaceShardToExecutor, err := newNamespaceShardToExecutor(testCluster.EtcdPrefix, testCluster.Namespace, testCluster.Client, stopCh, logger)
+	namespaceShardToExecutor, err := newNamespaceShardToExecutor(testCluster.EtcdPrefix, testCluster.Namespace, testCluster.Client, stopCh, logger, clock.NewRealTimeSource())
 	assert.NoError(t, err)
 	namespaceShardToExecutor.Start(&sync.WaitGroup{})
 	time.Sleep(50 * time.Millisecond)
@@ -84,12 +89,13 @@ func TestNamespaceShardToExecutor_Subscribe(t *testing.T) {
 	})
 
 	// Start the cache
-	namespaceShardToExecutor, err := newNamespaceShardToExecutor(testCluster.EtcdPrefix, testCluster.Namespace, testCluster.Client, stopCh, logger)
+	namespaceShardToExecutor, err := newNamespaceShardToExecutor(testCluster.EtcdPrefix, testCluster.Namespace, testCluster.Client, stopCh, logger, clock.NewRealTimeSource())
 	assert.NoError(t, err)
 	namespaceShardToExecutor.Start(&sync.WaitGroup{})
 
 	// Refresh the cache to get the initial state
-	namespaceShardToExecutor.refresh(context.Background())
+	err = namespaceShardToExecutor.refresh(context.Background())
+	require.NoError(t, err)
 
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
@@ -133,6 +139,125 @@ func TestNamespaceShardToExecutor_Subscribe(t *testing.T) {
 	wg.Wait()
 }
 
+func TestNamespaceShardToExecutor_watch_watchChanErrors(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	logger := testlogger.New(t)
+	mockClient := etcdclient.NewMockClient(ctrl)
+	stopCh := make(chan struct{})
+	testPrefix := "/test-prefix"
+	testNamespace := "test-namespace"
+
+	// Mock the Watch call to return our watch channel
+	watchChan := make(chan clientv3.WatchResponse)
+	mockClient.EXPECT().
+		Watch(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(watchChan).
+		AnyTimes()
+
+	e, err := newNamespaceShardToExecutor(testPrefix, testNamespace, mockClient, stopCh, logger, clock.NewRealTimeSource())
+	require.NoError(t, err)
+
+	// Test Case #1
+	// Test received compact revision error from watch channel
+	{
+		go func() {
+			watchChan <- clientv3.WatchResponse{
+				CompactRevision: 100,
+			}
+		}()
+
+		err = e.watch()
+		require.Error(t, err)
+		assert.ErrorContains(t, err, "etcdserver: mvcc: required revision has been compacted")
+	}
+
+	// Test Case #2
+	// Test closed watch channel
+	{
+		close(watchChan)
+		err = e.watch()
+		require.Error(t, err)
+		assert.ErrorContains(t, err, "watch channel closed")
+	}
+}
+
+func TestNamespaceShardToExecutor_namespaceRefreshLoop_watchError(t *testing.T) {
+	defer goleak.VerifyNone(t)
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	logger := testlogger.New(t)
+	mockClient := etcdclient.NewMockClient(ctrl)
+	timeSource := clock.NewMockedTimeSource()
+	stopCh := make(chan struct{})
+	testPrefix := "/test-prefix"
+	testNamespace := "test-namespace"
+
+	// mock for first watch call that receives error
+	watchChanRcvErr := make(chan clientv3.WatchResponse)
+	mockClient.EXPECT().
+		Watch(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(watchChanRcvErr)
+
+	// mock for second watch call that receives closed channel
+	watchChanClosed := make(chan clientv3.WatchResponse)
+	mockClient.EXPECT().
+		Watch(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(watchChanClosed)
+
+	// mock for third watch call that will be used when stopCh is closed
+	mockClient.EXPECT().
+		Watch(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(make(chan clientv3.WatchResponse))
+
+	e, err := newNamespaceShardToExecutor(testPrefix, testNamespace, mockClient, stopCh, logger, timeSource)
+	require.NoError(t, err)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	finished := atomic.Bool{}
+
+	go func() {
+		defer wg.Done()
+		e.namespaceRefreshLoop()
+		finished.Store(true)
+	}()
+
+	// Test Case #1: watchChan receives error
+	{
+		// Sends a response containing compact revision to simulate error
+		watchChanRcvErr <- clientv3.WatchResponse{
+			CompactRevision: 100,
+		}
+
+		timeSource.BlockUntil(1)
+		require.False(t, finished.Load(), "namespaceRefreshLoop should not exit on watch error")
+	}
+
+	// Test Case #2: watchChan is closed
+	{
+		timeSource.Advance(2 * namespaceRefreshLoopWatchRetryInterval)
+
+		// Sends a response containing compact revision to simulate error
+		close(watchChanClosed)
+
+		timeSource.BlockUntil(1)
+		require.False(t, finished.Load(), "namespaceRefreshLoop should not exit on watch error")
+	}
+
+	// Test Case #3: stopCh is closed
+	{
+		timeSource.Advance(2 * namespaceRefreshLoopWatchRetryInterval)
+
+		close(stopCh)
+		wg.Wait()
+		require.True(t, finished.Load(), "namespaceRefreshLoop should exit on watch error")
+	}
+}
+
 // setupExecutorWithShards creates an executor in etcd with assigned shards and metadata
 func setupExecutorWithShards(t *testing.T, testCluster *testhelper.StoreTestCluster, executorID string, shards []string, metadata map[string]string) {
 	// Create assigned state