From 3a4f1e4444dd47a68e20075b7aec776df43fe4b3 Mon Sep 17 00:00:00 2001 From: "r.inyakin" Date: Mon, 20 Jul 2026 16:29:55 +0300 Subject: [PATCH] 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 --- CHANGELOG.md | 2 ++ tarantool/connection.py | 11 ++++++-- tarantool/connection_pool.py | 3 ++- test/suites/test_connection.py | 25 ++++++++++++++++++ test/suites/test_pool.py | 47 ++++++++++++++++++++++++++++++++++ 5 files changed, 85 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a35b004..181ef7a2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/tarantool/connection.py b/tarantool/connection.py index 682b0700..2572ea47 100644 --- a/tarantool/connection.py +++ b/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): """ diff --git a/tarantool/connection_pool.py b/tarantool/connection_pool.py index 1734402a..9e335c37 100644 --- a/tarantool/connection_pool.py +++ b/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) diff --git a/test/suites/test_connection.py b/test/suites/test_connection.py index bc66a59f..c6129a30 100644 --- a/test/suites/test_connection.py +++ b/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() diff --git a/test/suites/test_pool.py b/test/suites/test_pool.py index e15bcbbf..a8a421db 100644 --- a/test/suites/test_pool.py +++ b/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()