Stop add follower retries when the follower is already present

common/error_codes.go

@@ -32,6 +32,7 @@ const (
 	CodeInvalidSessionTimeout   codes.Code = 109
 	CodeNamespaceNotFound       codes.Code = 110
 	CodeNotificationsNotEnabled codes.Code = 111
+	CodeFollowerAlreadyPresent  codes.Code = 112
 )
 
 var (
@@ -47,4 +48,5 @@ var (
 	ErrorInvalidSessionTimeout   = status.Error(CodeInvalidSessionTimeout, "oxia: invalid session timeout")
 	ErrorNamespaceNotFound       = status.Error(CodeNamespaceNotFound, "oxia: namespace not found")
 	ErrorNotificationsNotEnabled = status.Error(CodeNotificationsNotEnabled, "oxia: notifications not enabled on namespace")
+	ErrorFollowerAlreadyPresent  = status.Error(CodeFollowerAlreadyPresent, "oxia: follower is already present")
 )

coordinator/impl/shard_controller.go

@@ -512,7 +512,7 @@ func (s *shardController) internalNewTermAndAddFollower(ctx context.Context, nod
 	if err = s.addFollower(*s.shardMetadata.Leader, node.Internal, &proto.EntryId{
 		Term:   fr.Term,
 		Offset: fr.Offset,
-	}); err != nil {
+	}); err != nil && status.Code(err) != common.CodeFollowerAlreadyPresent {
 		res <- err
 		return
 	}

server/leader_controller.go

@@ -376,7 +376,7 @@ func (lc *leaderController) AddFollower(req *proto.AddFollowerRequest) (*proto.A
 	}
 
 	if _, followerAlreadyPresent := lc.followers[req.FollowerName]; followerAlreadyPresent {
-		return nil, errors.Errorf("follower %s is already present", req.FollowerName)
+		return nil, errors.Wrapf(common.ErrorFollowerAlreadyPresent, "follower: %s", req.FollowerName)
 	}
 
 	if len(lc.followers) == int(lc.replicationFactor)-1 {

server/leader_controller_test.go

@@ -553,6 +553,59 @@ func TestLeaderController_AddFollower(t *testing.T) {
 	assert.NoError(t, walFactory.Close())
 }
 
+func TestLeaderController_AddFollowerRepeated(t *testing.T) {
+	var shard int64 = 1
+
+	kvFactory, err := kv.NewPebbleKVFactory(testKVOptions)
+	assert.NoError(t, err)
+	walFactory := newTestWalFactory(t)
+
+	lc, err := NewLeaderController(Config{}, common.DefaultNamespace, shard, newMockRpcClient(), walFactory, kvFactory)
+	assert.NoError(t, err)
+
+	_, err = lc.NewTerm(&proto.NewTermRequest{
+		Term:  5,
+		Shard: shard,
+	})
+	assert.NoError(t, err)
+
+	assert.EqualValues(t, 5, lc.Term())
+	assert.Equal(t, proto.ServingStatus_FENCED, lc.Status())
+
+	_, err = lc.BecomeLeader(context.Background(), &proto.BecomeLeaderRequest{
+		Shard:             shard,
+		Term:              5,
+		ReplicationFactor: 3,
+		FollowerMaps: map[string]*proto.EntryId{
+			"f1": InvalidEntryId,
+		},
+	})
+	assert.NoError(t, err)
+
+	// f1 is already connected
+	afRes, err := lc.AddFollower(&proto.AddFollowerRequest{
+		Shard:               shard,
+		Term:                5,
+		FollowerName:        "f1",
+		FollowerHeadEntryId: InvalidEntryId,
+	})
+	assert.Nil(t, afRes)
+	assert.Error(t, err)
+
+	_, err = lc.AddFollower(&proto.AddFollowerRequest{
+		Shard:               shard,
+		Term:                5,
+		FollowerName:        "f1",
+		FollowerHeadEntryId: InvalidEntryId,
+	})
+	assert.Error(t, err)
+	assert.Equal(t, common.CodeFollowerAlreadyPresent, status.Code(err))
+
+	assert.NoError(t, lc.Close())
+	assert.NoError(t, kvFactory.Close())
+	assert.NoError(t, walFactory.Close())
+}
+
 // When a follower is added after the initial leader election,
 // the leader should use the head-entry at the time of the
 // election instead of the current head-entry.