[coordinator] Prevent TChannel leaks (#4026)

src/dbnode/client/connection_pool.go

@@ -206,6 +206,10 @@ func (p *connPool) connectEvery(interval time.Duration, stutter time.Duration) {
 				if p.status == statusOpen {
 					p.pool = append(p.pool, conn{channel, client})
 					p.poolLen = int64(len(p.pool))
+				} else {
+					// NB(antanas): just being defensive.
+					// It's likely a corner case and happens only during server shutdown.
+					channel.Close()
 				}
 				p.Unlock()
 			}()

src/dbnode/client/connection_pool_test.go

@@ -46,10 +46,26 @@ var (
 )
 
 type noopPooledChannel struct {
-	address string
+	address    string
+	closeCount int32
+}
+
+func asNoopPooledChannel(c Channel) *noopPooledChannel {
+	cc, ok := c.(*noopPooledChannel)
+	if !ok {
+		panic("not a noopPooledChannel")
+	}
+	return cc
+}
+
+func (c *noopPooledChannel) CloseCount() int {
+	return int(atomic.LoadInt32(&c.closeCount))
+}
+
+func (c *noopPooledChannel) Close() {
+	atomic.AddInt32(&c.closeCount, 1)
 }
 
-func (c *noopPooledChannel) Close() {}
 func (c *noopPooledChannel) GetSubChannel(
 	serviceName string,
 	opts ...tchannel.SubChannelOption,

src/dbnode/client/session.go

@@ -785,6 +785,7 @@ func (s *session) DedicatedConnection(
 		}
 
 		if err := s.healthCheckNewConnFn(client, s.opts, opts.BootstrappedNodesOnly); err != nil {
+			channel.Close()
 			multiErr = multiErr.Add(err)
 			return
 		}

src/dbnode/client/session_test.go

@@ -445,25 +445,35 @@ func TestDedicatedConnection(t *testing.T) {
 
 	_, ch, err := s.DedicatedConnection(shardID, DedicatedConnectionOptions{})
 	require.NoError(t, err)
-	assert.Equal(t, &noopPooledChannel{"remote1"}, ch)
+	assert.Equal(t, "remote1", asNoopPooledChannel(ch).address)
 
 	_, ch2, err := s.DedicatedConnection(shardID, DedicatedConnectionOptions{ShardStateFilter: shard.Available})
 	require.NoError(t, err)
-	assert.Equal(t, &noopPooledChannel{"remote2"}, ch2)
+	assert.Equal(t, "remote2", asNoopPooledChannel(ch2).address)
 
 	s.healthCheckNewConnFn = testHealthCheck(nil, true)
 	_, ch3, err := s.DedicatedConnection(shardID, DedicatedConnectionOptions{BootstrappedNodesOnly: true})
 	require.NoError(t, err)
-	assert.Equal(t, &noopPooledChannel{"remote1"}, ch3)
+	assert.Equal(t, "remote1", asNoopPooledChannel(ch3).address)
 
 	healthErr := errors.New("unhealthy")
 	s.healthCheckNewConnFn = testHealthCheck(healthErr, false)
 
+	var channels []*noopPooledChannel
+	s.opts = NewOptions().SetNewConnectionFn(func(_ string, _ string, _ Options) (Channel, rpc.TChanNode, error) {
+		c := &noopPooledChannel{"test", 0}
+		channels = append(channels, c)
+		return c, nil, nil
+	})
 	_, _, err = s.DedicatedConnection(shardID, DedicatedConnectionOptions{})
 	require.NotNil(t, err)
 	multiErr, ok := err.(xerror.MultiError) // nolint: errorlint
 	assert.True(t, ok, "expecting MultiError")
 	assert.True(t, multiErr.Contains(healthErr))
+	// 2 because of 2 remote hosts failing health check
+	assert.Len(t, channels, 2)
+	assert.Equal(t, 1, channels[0].CloseCount())
+	assert.Equal(t, 1, channels[1].CloseCount())
 }
 
 func testSessionClusterConnectConsistencyLevel(
@@ -621,9 +631,9 @@ func testHealthCheck(err error, bootstrappedNodesOnly bool) func(rpc.TChanNode,
 }
 
 func noopNewConnection(
-	channelName string,
+	_ string,
 	addr string,
-	opts Options,
+	_ Options,
 ) (Channel, rpc.TChanNode, error) {
-	return &noopPooledChannel{addr}, nil, nil
+	return &noopPooledChannel{addr, 0}, nil, nil
 }