#985 add exception for zmq connection error to identify TCP fallback

Author: PropzSaladaz

blockfinalize/client/BlockFinalizeDownloader.cpp

@@ -27,6 +27,7 @@
 #include "exceptions/ConnectionRefusedException.h"
 #include "exceptions/ExitRequestedException.h"
 #include "exceptions/DoNotHaveProposalYetException.h"
+#include "exceptions/ZMQTransportException.h"
 
 #include "abstracttcpserver/ConnectionStatus.h"
 
@@ -311,17 +312,14 @@ void BlockFinalizeDownloader::downloadFragment(
         return;
     }
 
-    // If node is not marked as TCP fallback, try ZMQ first, and if it fails, mark it as TCP fallback and retry using TCP.
+    // If node is not marked as TCP fallback, try ZMQ first. Downgrade to TCP only for
+    // explicit ZMQ transport failures, not for parsing, protocol, or local-state errors.
     try {
         downloadFragmentZMQ( _dstIndex, _fragmentIndex, header );
         clearTCPFallback( _dstIndex );
         return;
-    } catch ( DoNotHaveProposalYetException& ) {
-        throw;
-    } catch ( ExitRequestedException& ) {
-        throw;
-    } catch ( ... ) {
-        // peer is not usable via ZMQ, mark it as TCP fallback and retry using TCP
+    } catch ( const ZMQTransportException& e ) {
+        // peer is not usable via ZMQ transport right now, mark it as TCP fallback
         getNode()->getSockets()->getBulkDataZMQSockets()->closeDestinationSocket( _dstIndex );
         getSchain()->noteBlockFinalizeZmqClientFallbackToTcp();
         markTCPFallback( _dstIndex );

blockfinalize/client/BlockFinalizeDownloader.h

@@ -24,6 +24,7 @@
 
 #pragma once
 
+#include <exception>
 
 class CommittedBlockList;
 class ClientSocket;
@@ -129,7 +130,7 @@ class BlockFinalizeDownloader : public Agent {
 
     /**
      * Downloads block fragment from '_dstIndex' node using TCP connection. 
-     * This is the old interface, used only as a fallback when ZMQ download fails, 
+     * This is the old interface, used only as a fallback when ZMQ transport fails, 
      * and is marked in local cache to keep using TCP for this node for some time in the future.
      */
     void downloadFragmentTCP( schain_index _dstIndex, fragment_index _fragmentIndex,

exceptions/ZMQTransportException.h

@@ -0,0 +1,38 @@
+/*
+    Copyright (C) 2018-2019 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file ZMQTransportException.h
+    @author Sidnei Teixeira
+    @date 2026
+*/
+
+#pragma once
+
+#include "NetworkProtocolException.h"
+
+class ZMQTransportException : public NetworkProtocolException {
+    int zmqErrno = 0;
+
+public:
+    ZMQTransportException( const string& _message, int _zmqErrno, const string& _className )
+        : NetworkProtocolException( _message, _className ), zmqErrno( _zmqErrno ) {}
+
+    [[nodiscard]] int getZmqErrno() const {
+        return zmqErrno;
+    }
+};

network/ZMQErrorClassifier.h

@@ -0,0 +1,42 @@
+/*
+    Copyright (C) 2018-2019 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file ZMQErrorClassifier.h
+    @author Sidnei Teixeira
+    @date 2026
+*/
+
+#pragma once
+
+#include <cerrno>
+
+inline bool isZMQTransportErrorForTcpFallback( int _zmqErrno ) {
+    switch ( _zmqErrno ) {
+    case EAGAIN:
+    case ETIMEDOUT:
+    case ECONNRESET:
+    case ECONNREFUSED:
+    case ENETDOWN:
+    case ENETUNREACH:
+    case EHOSTUNREACH:
+    case EPIPE:
+        return true;
+    default:
+        return false;
+    }
+}

network/ZMQHeaderPayloadFrame.cpp

@@ -23,7 +23,9 @@
 #include "exceptions/ExitRequestedException.h"
 #include "exceptions/NetworkProtocolException.h"
 #include "exceptions/ParsingException.h"
+#include "exceptions/ZMQTransportException.h"
 #include "headers/Header.h"
+#include "network/ZMQErrorClassifier.h"
 #include "node/Node.h"
 #include "utils/Time.h"
 #include "ZMQHeaderPayloadFrame.h"
@@ -38,6 +40,17 @@ void maybeSimulateDelay( Schain& _sChain ) {
         usleep( simulatedDelay * 1000 );
     }
 }
+
+[[noreturn]] void throwZMQCommunicationException( const string& _message ) {
+    auto zmqErrno = zmq_errno();
+    auto errorMessage = _message + ":" + string( zmq_strerror( zmqErrno ) );
+
+    if ( isZMQTransportErrorForTcpFallback( zmqErrno ) ) {
+        BOOST_THROW_EXCEPTION( ZMQTransportException( errorMessage, zmqErrno, __CLASS_NAME__ ) );
+    }
+
+    BOOST_THROW_EXCEPTION( NetworkProtocolException( errorMessage, __CLASS_NAME__ ) );
+}
 }
 
 ptr< vector< uint8_t > > ZMQHeaderPayloadFrame::packMessage(
@@ -110,8 +123,7 @@ void ZMQHeaderPayloadFrame::sendFrame( Schain& _sChain, void* _socket, const ptr
 
     auto rc = zmq_send( _socket, _frame->data(), _frame->size(), 0 );
     if ( rc < 0 ) {
-        BOOST_THROW_EXCEPTION( NetworkProtocolException(
-            "ZMQ send failed:" + string( zmq_strerror( errno ) ), __CLASS_NAME__ ) );
+        throwZMQCommunicationException( "ZMQ send failed" );
     }
 }
 
@@ -134,8 +146,7 @@ void ZMQHeaderPayloadFrame::sendFrameToRoutingId( Schain& _sChain, void* _socket
     rc = zmq_msg_send( &msg, _socket, 0 );
     if ( rc < 0 ) {
         zmq_msg_close( &msg );
-        BOOST_THROW_EXCEPTION( NetworkProtocolException(
-            "ZMQ send failed:" + string( zmq_strerror( errno ) ), __CLASS_NAME__ ) );
+        throwZMQCommunicationException( "ZMQ send failed" );
     }
 
     rc = zmq_msg_close( &msg );
@@ -154,8 +165,9 @@ ptr< vector< uint8_t > > ZMQHeaderPayloadFrame::receiveFrame(
 
     rc = zmq_msg_recv( &msg, _socket, 0 );
     if ( rc < 0 ) {
+        auto zmqErrno = zmq_errno();
         zmq_msg_close( &msg );
-        if ( errno == EAGAIN ) {
+        if ( zmqErrno == EAGAIN ) {
             // check if timeout happened during exit
             _sChain.getNode()->exitCheck();
 
@@ -164,8 +176,12 @@ ptr< vector< uint8_t > > ZMQHeaderPayloadFrame::receiveFrame(
                 return nullptr;
             }
         }
-        BOOST_THROW_EXCEPTION( NetworkProtocolException(
-            string( _errorString ) + ":" + zmq_strerror( errno ), __CLASS_NAME__ ) );
+        auto errorMessage = string( _errorString ) + ":" + string( zmq_strerror( zmqErrno ) );
+        if ( isZMQTransportErrorForTcpFallback( zmqErrno ) ) {
+            BOOST_THROW_EXCEPTION(
+                ZMQTransportException( errorMessage, zmqErrno, __CLASS_NAME__ ) );
+        }
+        BOOST_THROW_EXCEPTION( NetworkProtocolException( errorMessage, __CLASS_NAME__ ) );
     }
 
     auto frame = make_shared< vector< uint8_t > >( rc );

tests/unit/network/ZMQErrorClassifierTests.cpp

@@ -0,0 +1,53 @@
+#include "thirdparty/catch.hpp"
+
+#include <cerrno>
+#include <exception>
+
+#include "exceptions/InvalidStateException.h"
+#include "exceptions/NetworkProtocolException.h"
+#include "exceptions/ParsingException.h"
+#include "exceptions/ZMQTransportException.h"
+#include "network/ZMQErrorClassifier.h"
+#include "zmq.h"
+
+CATCH_TEST_CASE( "ZMQ transport fallback classifier accepts transport errno values",
+    "[network][zmq][unit]" ) {
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( EAGAIN ) );
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( ETIMEDOUT ) );
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( ECONNRESET ) );
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( ECONNREFUSED ) );
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( ENETDOWN ) );
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( ENETUNREACH ) );
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( EHOSTUNREACH ) );
+    CATCH_REQUIRE( isZMQTransportErrorForTcpFallback( EPIPE ) );
+}
+
+CATCH_TEST_CASE( "ZMQ transport fallback classifier rejects non-transport errno values",
+    "[network][zmq][unit]" ) {
+    CATCH_REQUIRE_FALSE( isZMQTransportErrorForTcpFallback( EINVAL ) );
+    CATCH_REQUIRE_FALSE( isZMQTransportErrorForTcpFallback( EFSM ) );
+    CATCH_REQUIRE_FALSE( isZMQTransportErrorForTcpFallback( ETERM ) );
+}
+
+CATCH_TEST_CASE( "BlockFinalize TCP fallback only applies to ZMQ transport exceptions",
+    "[blockfinalize][network][unit]" ) {
+    ZMQTransportException transportException( "transport", EAGAIN, "Test" );
+    ParsingException parsingException( "parsing", "Test" );
+    NetworkProtocolException protocolException( "protocol", "Test" );
+    InvalidStateException invalidStateException( "state", "Test" );
+    auto catchesAsTransport = []( const auto& _exception ) {
+        try {
+            throw _exception;
+        } catch ( const ZMQTransportException& ) {
+            return true;
+        } catch ( ... ) {
+            return false;
+        }
+    };
+
+    CATCH_REQUIRE( transportException.getZmqErrno() == EAGAIN );
+    CATCH_REQUIRE( catchesAsTransport( transportException ) );
+    CATCH_REQUIRE_FALSE( catchesAsTransport( parsingException ) );
+    CATCH_REQUIRE_FALSE( catchesAsTransport( protocolException ) );
+    CATCH_REQUIRE_FALSE( catchesAsTransport( invalidStateException ) );
+}