zmq: Log bind error at Error level, abort startup on init error #33727

pointers are nullable, so why wrap twice?

nullptr is already used to indicate that no ZMQ interface was initialized. This is not the same as failing to initialize a ZMQ interface.

I also don't like this double wrapping, it's error prone. It seems like this is currently necessary because Create is actually more like a MaybeCreate, with actual creation depending on gArgs. A more robust approach imo would be to take gArgs out of Create, let the caller decide if we actually want to create a CZMQNotificationInterface and pass the necessary parameters. With that change, Create should always return a value, so we can keep nullptr for the failure case?

One approach could be to carve out a GetNotifiers function:

<details> <summary>git diff on 759725ffed</summary>

diff --git a/src/init.cpp b/src/init.cpp
index cd3512f3a9..0582d7ed8f 100644
--- a/src/init.cpp
+++ b/src/init.cpp
@@ -1749,15 +1749,18 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     }
 
 #ifdef ENABLE_ZMQ
-    auto zmq_notification_interface = CZMQNotificationInterface::Create(
+    auto notifiers{CZMQNotificationInterface::GetNotifiers(
+        gArgs,
         [&chainman = node.chainman](std::vector<std::byte>& block, const CBlockIndex& index) {
             assert(chainman);
             return chainman->m_blockman.ReadRawBlock(block, WITH_LOCK(cs_main, return index.GetBlockPos()));
-        });
-    if (zmq_notification_interface.has_value()) {
-        g_zmq_notification_interface = std::move(zmq_notification_interface.value());
-    } else {
-        return InitError(Untranslated("Initializing ZMQ interface failed."));
+        })};
+    if (!notifiers.empty()) {
+        if (auto zmq_notification_interface{CZMQNotificationInterface::Create(std::move(notifiers))}) {
+            g_zmq_notification_interface = std::move(zmq_notification_interface);
+        } else {
+            return InitError(Untranslated("Initializing ZMQ interface failed."));
+        }
     }
 
     if (g_zmq_notification_interface) {
diff --git a/src/zmq/zmqnotificationinterface.cpp b/src/zmq/zmqnotificationinterface.cpp
index eae377f718..c1e45e33e5 100644
--- a/src/zmq/zmqnotificationinterface.cpp
+++ b/src/zmq/zmqnotificationinterface.cpp
@@ -40,7 +40,9 @@ std::list<const CZMQAbstractNotifier*> CZMQNotificationInterface::GetActiveNotif
     return result;
 }
 
-std::optional<std::unique_ptr<CZMQNotificationInterface>> CZMQNotificationInterface::Create(std::function<bool(std::vector<std::byte>&, const CBlockIndex&)> get_block_by_index)
+std::list<std::unique_ptr<CZMQAbstractNotifier>> CZMQNotificationInterface::GetNotifiers(
+    const ArgsManager& args,
+    std::function<bool(std::vector<std::byte>&, const CBlockIndex&)> get_block_by_index)
 {
     std::map<std::string, CZMQNotifierFactory> factories;
     factories["pubhashblock"] = CZMQAbstractNotifier::Create<CZMQPublishHashBlockNotifier>;
@@ -56,7 +58,7 @@ std::optional<std::unique_ptr<CZMQNotificationInterface>> CZMQNotificationInterf
     {
         std::string arg("-zmq" + entry.first);
         const auto& factory = entry.second;
-        for (std::string& address : gArgs.GetArgs(arg)) {
+        for (std::string& address : args.GetArgs(arg)) {
             // libzmq uses prefix "ipc://" for UNIX domain sockets
             if (address.starts_with(ADDR_PREFIX_UNIX)) {
                 address.replace(0, ADDR_PREFIX_UNIX.length(), ADDR_PREFIX_IPC);
@@ -65,24 +67,26 @@ std::optional<std::unique_ptr<CZMQNotificationInterface>> CZMQNotificationInterf
             std::unique_ptr<CZMQAbstractNotifier> notifier = factory();
             notifier->SetType(entry.first);
             notifier->SetAddress(address);
-            notifier->SetOutboundMessageHighWaterMark(static_cast<int>(gArgs.GetIntArg(arg + "hwm", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM)));
+            notifier->SetOutboundMessageHighWaterMark(static_cast<int>(args.GetIntArg(arg + "hwm", CZMQAbstractNotifier::DEFAULT_ZMQ_SNDHWM)));
             notifiers.push_back(std::move(notifier));
         }
     }
+    return notifiers;
+}
 
-    if (!notifiers.empty())
-    {
-        std::unique_ptr<CZMQNotificationInterface> notificationInterface(new CZMQNotificationInterface());
-        notificationInterface->notifiers = std::move(notifiers);
+std::unique_ptr<CZMQNotificationInterface> CZMQNotificationInterface::Create(
+    std::list<std::unique_ptr<CZMQAbstractNotifier>>&& notifiers)
+{
+    if (notifiers.empty()) return nullptr;
 
-        if (notificationInterface->Initialize()) {
-            return notificationInterface;
-        } else {
-            return std::nullopt;
-        }
-    }
+    std::unique_ptr<CZMQNotificationInterface> notificationInterface(new CZMQNotificationInterface());
+    notificationInterface->notifiers = std::move(notifiers);
 
-    return nullptr;
+    if (notificationInterface->Initialize()) {
+        return notificationInterface;
+    } else {
+        return nullptr;
+    }
 }
 
 // Called at startup to conditionally set up ZMQ socket(s)
diff --git a/src/zmq/zmqnotificationinterface.h b/src/zmq/zmqnotificationinterface.h
index 0abbbef428..f36d633814 100644
--- a/src/zmq/zmqnotificationinterface.h
+++ b/src/zmq/zmqnotificationinterface.h
@@ -14,6 +14,7 @@
 #include <memory>
 #include <vector>
 
+class ArgsManager;
 class CBlock;
 class CBlockIndex;
 class CZMQAbstractNotifier;
@@ -26,8 +27,11 @@ public:
 
     std::list<const CZMQAbstractNotifier*> GetActiveNotifiers() const;
 
-    // nullopt if initialization fails
-    static std::optional<std::unique_ptr<CZMQNotificationInterface>> Create(std::function<bool(std::vector<std::byte>&, const CBlockIndex&)> get_block_by_index);
+    static std::list<std::unique_ptr<CZMQAbstractNotifier>> GetNotifiers(
+        const ArgsManager& args,
+        std::function<bool(std::vector<std::byte>&, const CBlockIndex&)> get_block_by_index);
+    // nullptr if initialization fails
+    static std::unique_ptr<CZMQNotificationInterface> Create(std::list<std::unique_ptr<CZMQAbstractNotifier>>&& notifiers);
 
 protected:
     bool Initialize();

</details>

Edit: adding some unit tests around those functions could also be helpful.

Applied the patch, thanks

nit: line length (also in .h)

Error and Debug are different levels, so this name seems confusing to me. Would simplify to zmqError and zmqDebug, removing the level parameter.

Side note: I don't know much about zmq, but it seems like some of these debug messages should actually be warnings, e.g. when block gets failed to read from disk? Can/should probably be done in a separate pull, though.

I agree that zmqDebug/zmqError would be ideal, however zmqError is currently used to log at the Debug level - changing it to log at Error level while keeping the type signature the same seems likely to cause misuse.

This code isn't touched much, the last commit touching zmqError is from 2021. I don't think any authors or reviewers would assume zmqError means zmqDebug without looking at the underlying code.

On further inspection, it seems like none of these messages should be debug to begin with? They all look like warning or error to me? Finally, LogZmq{Error,Warning} would be a better name.

Suggested (untested) diff to address all:

<details> <summary>git diff on 05661ef06b</summary>

diff --git a/src/zmq/zmqnotificationinterface.cpp b/src/zmq/zmqnotificationinterface.cpp
index c1e45e33e5..85b8b995b9 100644
--- a/src/zmq/zmqnotificationinterface.cpp
+++ b/src/zmq/zmqnotificationinterface.cpp
@@ -103,7 +103,7 @@ bool CZMQNotificationInterface::Initialize()
 
     if (!pcontext)
     {
-        zmqErrorDebug("Unable to initialize context");
+        LogZmqError("Unable to initialize context");
         return false;
     }
 
diff --git a/src/zmq/zmqpublishnotifier.cpp b/src/zmq/zmqpublishnotifier.cpp
index 665f4a42f8..a7cbf48915 100644
--- a/src/zmq/zmqpublishnotifier.cpp
+++ b/src/zmq/zmqpublishnotifier.cpp
@@ -59,7 +59,7 @@ static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
         int rc = zmq_msg_init_size(&msg, size);
         if (rc != 0)
         {
-            zmqErrorDebug("Unable to initialize ZMQ msg");
+            LogZmqWarning("Unable to initialize ZMQ msg");
             va_end(args);
             return -1;
         }
@@ -72,7 +72,7 @@ static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
         rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0);
         if (rc == -1)
         {
-            zmqErrorDebug("Unable to send ZMQ msg");
+            LogZmqWarning("Unable to send ZMQ msg");
             zmq_msg_close(&msg);
             va_end(args);
             return -1;
@@ -114,7 +114,7 @@ bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
         psocket = zmq_socket(pcontext, ZMQ_PUB);
         if (!psocket)
         {
-            zmqErrorDebug("Failed to create socket");
+            LogZmqError("Failed to create socket");
             return false;
         }
 
@@ -123,7 +123,7 @@ bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
         int rc = zmq_setsockopt(psocket, ZMQ_SNDHWM, &outbound_message_high_water_mark, sizeof(outbound_message_high_water_mark));
         if (rc != 0)
         {
-            zmqErrorDebug("Failed to set outbound message high water mark");
+            LogZmqError("Failed to set outbound message high water mark");
             zmq_close(psocket);
             return false;
         }
@@ -131,7 +131,7 @@ bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
         const int so_keepalive_option {1};
         rc = zmq_setsockopt(psocket, ZMQ_TCP_KEEPALIVE, &so_keepalive_option, sizeof(so_keepalive_option));
         if (rc != 0) {
-            zmqErrorDebug("Failed to set SO_KEEPALIVE");
+            LogZmqError("Failed to set SO_KEEPALIVE");
             zmq_close(psocket);
             return false;
         }
@@ -140,7 +140,7 @@ bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
         const int enable_ipv6 { IsZMQAddressIPV6(address) ? 1 : 0};
         rc = zmq_setsockopt(psocket, ZMQ_IPV6, &enable_ipv6, sizeof(enable_ipv6));
         if (rc != 0) {
-            zmqErrorDebug("Failed to set ZMQ_IPV6");
+            LogZmqError("Failed to set ZMQ_IPV6");
             zmq_close(psocket);
             return false;
         }
@@ -148,7 +148,7 @@ bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
         rc = zmq_bind(psocket, address.c_str());
         if (rc != 0)
         {
-            zmqError(BCLog::Level::Error, "Failed to bind address");
+            LogZmqError("Failed to bind address");
             zmq_close(psocket);
             return false;
         }
@@ -245,7 +245,7 @@ bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
 
     std::vector<std::byte> block{};
     if (!m_get_block_by_index(block, *pindex)) {
-        zmqErrorDebug("Can't read block from disk");
+        LogZmqWarning("Can't read block from disk");
         return false;
     }
 
diff --git a/src/zmq/zmqutil.cpp b/src/zmq/zmqutil.cpp
index 86d0b5a2cf..e10817ce19 100644
--- a/src/zmq/zmqutil.cpp
+++ b/src/zmq/zmqutil.cpp
@@ -8,14 +8,18 @@
 #include <zmq.h>
 
 #include <cerrno>
-#include <string>
+#include <string_view>
 
-void zmqError(BCLog::Level level, const std::string& str)
+void LogZmqError(std::string_view context)
 {
-    LogPrintLevel(BCLog::ZMQ, level, "Error: %s, msg: %s\n", str, zmq_strerror(errno));
+    LogPrintLevel(BCLog::ZMQ, BCLog::Level::Error,
+                  "%s, zmq_errno: %d, msg: %s\n",
+                  context, errno, zmq_strerror(errno));
 }
 
-void zmqErrorDebug(const std::string& str)
+void LogZmqWarning(std::string_view context)
 {
-    zmqError(BCLog::Level::Debug, str);
+    LogPrintLevel(BCLog::ZMQ, BCLog::Level::Warning,
+                  "%s, zmq_errno: %d, msg: %s\n",
+                  context, errno, zmq_strerror(errno));
 }
diff --git a/src/zmq/zmqutil.h b/src/zmq/zmqutil.h
index 595e2c1d31..115254e5f4 100644
--- a/src/zmq/zmqutil.h
+++ b/src/zmq/zmqutil.h
@@ -8,12 +8,13 @@
 #include <logging.h>
 
 #include <string>
+#include <string_view>
 
-/** Log ZMQ error at the specified level */
-void zmqError(BCLog::Level level, const std::string& str);
+/** Log ZMQ failure with Error severity. Use for fatal errors that prevent ZMQ functionality. */
+void LogZmqError(std::string_view context);
 
-/** Log ZMQ error at Debug level */
-void zmqErrorDebug(const std::string& str);
+/** Log ZMQ failure with Warning severity. Use for non-fatal issues that may impact functionality. */
+void LogZmqWarning(std::string_view context);
 
 /** Prefix for unix domain socket addresses (which are local filesystem paths) */
 const std::string ADDR_PREFIX_IPC = "ipc://"; // used by libzmq, example "ipc:///root/path/to/file"

</details>

Instead of disabling the test, we should probably test that invalid arguments shut down the node, as per the new behaviour?

Done

Would be nice to assert the expected_msg here. Also, to more explicitly cover the issue in #33715 I'd expect to see a test where zmq is set to a valid tcp address but one that is already in use (you should be able to attempt a bind to node's p2p or rpc port)