connection_pool: recover instances after a cluster restart

Treat any instance error as an unhealthy state instead of an unhandled
exception. An instance which has not finished its bootstrap yet replies
with an error to `box.info`, and the `Response` constructor raises it
as a plain DatabaseError. `_get_new_state()` caught only a
NetworkError, so the error escaped the background refresh loop and
killed its thread.

Close the socket if a handshake fails: `is_closed()` only checks the
socket, so such a connection was reported as open and was never
authenticated again.

Closes #328

Author: r.inyakin

CHANGELOG.md

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Set upper bound for version of setuptools (PR #342).
 - Reduce idle CPU usage in `ConnectionPool` while waiting for
   queued requests (PR #336).
+- Recover `ConnectionPool` instances after a cluster restart
+  (PR #343).
 
 ## 1.2.0 - 2024-03-27
 

tarantool/connection.py

@@ -1155,6 +1155,7 @@ def connect(self):
             raise exc
         except Exception as exc:
             self.connected = False
+            self.close()
             raise NetworkError(exc) from exc
 
     def _recv(self, to_read):
@@ -1271,7 +1272,8 @@ def _send_request_wo_reconnect(self, request, on_push=None, on_push_ctx=None):
 
         :raise: :exc:`~AssertionError`,
             :exc:`~tarantool.error.SchemaError`,
-            :exc:`~tarantool.error.NetworkError`
+            :exc:`~tarantool.error.NetworkError`,
+            :exc:`~tarantool.error.DatabaseError`
 
         :meta private:
         """
@@ -1365,7 +1367,12 @@ def check():  # Check that connection is alive
             attempt += 1
         if self.transport == SSL_TRANSPORT:
             self.wrap_socket_ssl()
-        self.handshake()
+        try:
+            self.handshake()
+        except Exception:
+            self.connected = False
+            self.close()
+            raise
 
     def _send_request(self, request, on_push=None, on_push_ctx=None):
         """

tarantool/connection_pool.py

@@ -25,6 +25,7 @@
     PoolTolopogyError,
     PoolTolopogyWarning,
     ConfigurationError,
+    DatabaseError,
     NetworkError,
     warn
 )
@@ -575,7 +576,7 @@ def _get_new_state(self, unit):
 
         try:
             resp = conn.call('box.info')
-        except NetworkError as exc:
+        except DatabaseError as exc:
             msg = (f"Failed to get box.info for {unit.get_address()}, "
                    f"reason: {repr(exc)}")
             warn(msg, PoolTolopogyWarning)

test/suites/test_connection.py

@@ -3,6 +3,7 @@
 """
 # pylint: disable=missing-class-docstring,missing-function-docstring,duplicate-code
 
+import socket
 import sys
 import unittest
 import decimal
@@ -12,6 +13,7 @@
 
 import tarantool
 import tarantool.msgpack_ext.decimal as ext_decimal
+from tarantool.error import DatabaseError, NetworkError
 
 from .lib.skip import skip_or_run_decimal_test, skip_or_run_varbinary_test
 from .lib.tarantool_server import TarantoolServer
@@ -171,6 +173,29 @@ def my_unpacker_factory(_):
         resp = self.con.eval("return {1, 2, 3}")
         self.assertIsInstance(resp[0], tuple)
 
+    def test_failed_connect_closes_socket(self):
+        con = tarantool.Connection(self.srv.host, self.srv.args['primary'],
+                                   user='nosuchuser', password='wrongpassword',
+                                   connect_now=False)
+
+        with self.assertRaises(NetworkError):
+            con.connect()
+
+        self.assertTrue(con.is_closed())
+
+    def test_failed_handshake_on_reconnect_closes_socket(self):
+        self.con = tarantool.Connection(self.srv.host, self.srv.args['primary'],
+                                        user='test', password='test')
+        self.assertFalse(self.con.is_closed())
+
+        self.con.user = 'nosuchuser'
+        self.con._socket.shutdown(socket.SHUT_RDWR)  # pylint: disable=protected-access
+
+        with self.assertRaises(DatabaseError):
+            self.con.call('box.info')
+
+        self.assertTrue(self.con.is_closed())
+
     def tearDown(self):
         if self.con:
             self.con.close()

test/suites/test_pool.py

@@ -10,6 +10,7 @@
 import warnings
 
 import tarantool
+from tarantool.connection_pool import Status
 from tarantool.error import (
     ClusterConnectWarning,
     DatabaseError,
@@ -581,6 +582,52 @@ def test_16_is_closed(self):
 
         self.assertEqual(self.pool.is_closed(), True)
 
+    def test_17_instance_bootstrap_error_does_not_kill_refresh(self):
+        warnings.simplefilter('ignore', category=PoolTolopogyWarning)
+
+        self.set_cluster_ro([False, True, True, True, True])
+
+        self.pool = tarantool.ConnectionPool(
+            addrs=self.addrs,
+            user='test',
+            password='test',
+            refresh_delay=0.2)
+
+        self.pool.ping(mode=tarantool.Mode.RW)
+
+        unit = self.pool.pool[f"{self.addrs[0]['host']}:{self.addrs[0]['port']}"]
+
+        # Simulate an instance which is up, but has not finished its
+        # bootstrap yet: box.info fails with a plain DatabaseError
+        # instead of a NetworkError.
+        resp = self.servers[0].admin(r"""
+            rawset(_G, 'box_info_backup', box.info)
+            box.info = function()
+                box.error({code = 116,
+                           reason = "Instance bootstrap hasn't finished yet"})
+            end
+            return true
+        """)
+        assert_admin_success(resp)
+
+        def expect_instance_unhealthy_and_refresh_alive():
+            self.assertTrue(unit.thread.is_alive(),
+                            'refresh thread died on a DatabaseError')
+            self.assertEqual(unit.state.status, Status.UNHEALTHY)
+
+        self.retry(func=expect_instance_unhealthy_and_refresh_alive)
+
+        resp = self.servers[0].admin(r"""
+            box.info = box_info_backup
+            return true
+        """)
+        assert_admin_success(resp)
+
+        def expect_rw_request_succeed():
+            self.pool.ping(mode=tarantool.Mode.RW)
+
+        self.retry(func=expect_rw_request_succeed)
+
     def tearDown(self):
         if self.pool:
             self.pool.close()