test (1e728c16) · Commits · tuxist / libparitypp

include/paritypp/client.h

+2 −0

Original line number	Diff line number	Diff line
		@@ -134,6 +134,8 @@ private:
		struct quic_connection {
		std::unique_ptr<netplus::quic> socket;
		bool connected = false;
		// Track dead nodes to skip them quickly
		std::chrono::steady_clock::time_point dead_until{};
		};

		std::vector<quic_connection> connections_;

src/client.cpp

+20 −6

Original line number	Diff line number	Diff line
		@@ -68,6 +68,7 @@ void client::connect_to_node(size_t node_index) {
		nc.socket->pumpNetwork(MSG_DONTWAIT);
		if (nc.socket->getHandshakeDone()) {
		nc.connected = true;
		nc.dead_until = {}; // clear dead status on successful connect
		// Authenticate if credentials provided
		if (!creds_.client_name.empty()) {
		if (!authenticate_node(node_index)) {
		@@ -82,6 +83,7 @@ void client::connect_to_node(size_t node_index) {
		}
		nc.socket.reset();
		nc.connected = false;
		nc.dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);
		throw std::runtime_error("Failed to connect to node " + nodes_[node_index].address);
		}

		@@ -119,12 +121,20 @@ bool client::authenticate_node(size_t node_index) {
		}

		void client::ensure_connected(size_t node_index) {
		if (connections_[node_index].connected && connections_[node_index].socket) {
		auto& nc = connections_[node_index];
		if (nc.connected && nc.socket) {
		return;
		}
		// Skip node if recently dead (cooldown period)
		if (nc.dead_until > std::chrono::steady_clock::time_point{}) {
		if (std::chrono::steady_clock::now() < nc.dead_until) {
		throw std::runtime_error("node " + nodes_[node_index].address + " dead (cooldown)");
		}
		nc.dead_until = {};
		}
		// Reset broken state before reconnecting
		connections_[node_index].socket.reset();
		connections_[node_index].connected = false;
		nc.socket.reset();
		nc.connected = false;
		connect_to_node(node_index);
		}

		@@ -181,12 +191,12 @@ std::vector<uint8_t> client::send_recv(size_t node_index,
		}
		}
		conn.closeStream(stream);
		// No response after all attempts — mark connection as broken so
		// subsequent calls (e.g. get_cluster_status) fail fast instead of
		// retrying the expensive connect + send loop on a dead node.
		// No response after all attempts — mark connection as broken and dead
		// so subsequent calls fail fast instead of retrying.
		if (response.empty()) {
		connections_[node_index].connected = false;
		connections_[node_index].socket.reset();
		connections_[node_index].dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);
		}
		return response;
		}
		@@ -334,14 +344,17 @@ void client::store_stripe(uint64_t group_id, uint32_t stripe_index,
		pending[i].fail_reason = std::string("net: ") + e.what();
		connections_[i].socket.reset();
		connections_[i].connected = false;
		connections_[i].dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);
		} catch (const std::exception& e) {
		pending[i].fail_reason = std::string("exception: ") + e.what();
		connections_[i].socket.reset();
		connections_[i].connected = false;
		connections_[i].dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);
		} catch (...) {
		pending[i].fail_reason = "unknown exception";
		connections_[i].socket.reset();
		connections_[i].connected = false;
		connections_[i].dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);
		}
		};

		@@ -566,6 +579,7 @@ bool client::retrieve_stripe(uint64_t group_id, uint32_t stripe_index,
		} catch (...) {
		connections_[i].socket.reset();
		connections_[i].connected = false;
		connections_[i].dead_until = std::chrono::steady_clock::now() + std::chrono::seconds(10);
		}
		};