Immediately trigger error for non-recoverable error in client creation (oxia-db#295)

A non-recoverable error, such as trying to use a namespace that does not
exist, should trigger an immediate error in the client creation, instead
of having the client do multiple retries.

Author: merlimat

common/wait_group.go

@@ -62,7 +62,10 @@ func (g *waitGroup) Wait(ctx context.Context) error {
 }
 
 func (g *waitGroup) Done() {
-	g.responses <- nil
+	select {
+	case g.responses <- nil:
+	default:
+	}
 }
 
 func (g *waitGroup) Fail(err error) {

coordinator/impl/coordinator_e2e_test.go

@@ -293,6 +293,10 @@ func TestCoordinator_MultipleNamespaces(t *testing.T) {
 	assert.NoError(t, err)
 	defer clientNs1.Close()
 
+	clientNs3, err := oxia.NewSyncClient(sa1.Public, oxia.WithNamespace("my-ns-does-not-exist"))
+	assert.ErrorIs(t, err, common.ErrorNamespaceNotFound)
+	assert.Nil(t, clientNs3)
+
 	ctx := context.Background()
 
 	// Write in default ns

oxia/internal/shard_manager.go

@@ -38,7 +38,7 @@ type ShardManager interface {
 
 type shardManagerImpl struct {
 	sync.RWMutex
-	updatedCondition common.ConditionContext
+	updatedWg common.WaitGroup
 
 	shardStrategy  ShardStrategy
 	clientPool     common.ClientPool
@@ -63,7 +63,7 @@ func NewShardManager(shardStrategy ShardStrategy, clientPool common.ClientPool,
 		logger:         log.With().Str("component", "shardManager").Logger(),
 	}
 
-	sm.updatedCondition = common.NewConditionContext(sm)
+	sm.updatedWg = common.NewWaitGroup(1)
 	sm.ctx, sm.cancel = context.WithCancel(context.Background())
 
 	if err := sm.start(); err != nil {
@@ -80,7 +80,6 @@ func (s *shardManagerImpl) Close() error {
 
 func (s *shardManagerImpl) start() error {
 	s.Lock()
-	defer s.Unlock()
 
 	go common.DoWithLabels(map[string]string{
 		"oxia": "receive-shard-updates",
@@ -89,7 +88,8 @@ func (s *shardManagerImpl) start() error {
 	ctx, cancel := context.WithTimeout(s.ctx, s.requestTimeout)
 	defer cancel()
 
-	return s.updatedCondition.Wait(ctx)
+	s.Unlock()
+	return s.updatedWg.Wait(ctx)
 }
 
 func (s *shardManagerImpl) Get(key string) uint32 {
@@ -140,6 +140,10 @@ func (s *shardManagerImpl) receiveWithRecovery() {
 				s.logger.Debug().Err(err).Msg("Closed")
 				return nil
 			}
+
+			if !isErrorRetryable(err) {
+				return backoff.Permanent(err)
+			}
 			return err
 		},
 		backOff,
@@ -153,6 +157,7 @@ func (s *shardManagerImpl) receiveWithRecovery() {
 	)
 	if err != nil {
 		s.logger.Error().Err(err).Msg("Failed receiving shard assignments")
+		s.updatedWg.Fail(err)
 	}
 }
 
@@ -205,9 +210,19 @@ func (s *shardManagerImpl) update(updates []Shard) {
 		s.shards[update.Id] = update
 	}
 
-	s.updatedCondition.Signal()
+	s.updatedWg.Done()
 }
 
 func overlap(a HashRange, b HashRange) bool {
 	return !(a.MinInclusive > b.MaxInclusive || a.MaxInclusive < b.MinInclusive)
 }
+
+func isErrorRetryable(err error) bool {
+	switch status.Code(err) {
+	case common.CodeNamespaceNotFound:
+		return false
+
+	default:
+		return true
+	}
+}

server/assignment_dispatcher.go

@@ -73,6 +73,7 @@ func (s *shardAssignmentDispatcher) RegisterForUpdates(req *proto.ShardAssignmen
 	}
 
 	if _, ok := s.assignments.Namespaces[namespace]; !ok {
+		s.Unlock()
 		return common.ErrorNamespaceNotFound
 	}