Harden HTTP server against up-but-not-serving failure modes

src/mtconnect/configuration/async_context.hpp

@@ -71,7 +71,24 @@ namespace mtconnect::configuration {
       {
         for (int i = 0; i < m_threadCount; i++)
         {
-          m_workers.emplace_back(boost::thread([this]() { m_context.run(); }));
+          m_workers.emplace_back(boost::thread([this]() {
+            while (m_running)
+            {
+              try
+              {
+                m_context.run();
+                break;  // clean return: context stopped or out of work
+              }
+              catch (const std::exception &e)
+              {
+                LOG(error) << "Unhandled exception in io_context worker, resuming: " << e.what();
+              }
+              catch (...)
+              {
+                LOG(error) << "Unhandled unknown exception in io_context worker, resuming";
+              }
+            }
+          }));
         }
         auto &first = m_workers.front();
         while (m_running && !m_paused)

src/mtconnect/sink/rest_sink/server.cpp

@@ -49,6 +49,10 @@ namespace mtconnect::sink::rest_sink {
   using std::placeholders::_1;
   using std::placeholders::_2;
 
+  // Delay before re-arming the accept loop after a recoverable accept error so we
+  // don't hot-loop while a transient condition (e.g. fd exhaustion) clears.
+  static constexpr auto ACCEPT_BACKOFF = std::chrono::milliseconds(100);
+
   void Server::loadTlsCertificate()
   {
     if (HasOption(m_options, configuration::TlsCertificateChain) &&
@@ -117,19 +121,19 @@ namespace mtconnect::sink::rest_sink {
     if (ec)
     {
       fail(ec, "Cannot open server socket");
-      return;
+      throw std::runtime_error(std::string("Cannot open server socket: ") + ec.message());
     }
     m_acceptor.set_option(boost::asio::socket_base::reuse_address(true), ec);
     if (ec)
     {
       fail(ec, "Cannot set reuse address");
-      return;
+      throw std::runtime_error(std::string("Cannot set reuse address: ") + ec.message());
     }
     m_acceptor.bind(ep, ec);
     if (ec)
     {
       fail(ec, "Cannot bind to server address");
-      return;
+      throw std::runtime_error(std::string("Cannot bind to server address: ") + ec.message());
     }
     if (m_port == 0)
     {
@@ -140,7 +144,7 @@ namespace mtconnect::sink::rest_sink {
     if (ec)
     {
       fail(ec, "Cannot set listen queue length");
-      return;
+      throw std::runtime_error(std::string("Cannot set listen queue length: ") + ec.message());
     }
 
     m_listening = true;
@@ -177,13 +181,39 @@ namespace mtconnect::sink::rest_sink {
   {
     NAMED_SCOPE("Server::accept");
 
+    if (ec == boost::asio::error::operation_aborted)
+    {
+      // The acceptor was closed during shutdown, do not re-arm.
+      return;
+    }
+
     if (ec)
     {
       LOG(error) << ec.category().message(ec.value()) << ": " << ec.message();
 
       fail(ec, "Accept failed");
+
+      // The acceptor remains open and valid after a non-fatal accept error, so
+      // re-arm after a short backoff to avoid hot-looping while a transient
+      // condition (e.g. fd exhaustion) clears.
+      if (m_run)
+      {
+        m_acceptBackoff.expires_after(ACCEPT_BACKOFF);
+        m_acceptBackoff.async_wait([this](beast::error_code tec) {
+          if (!tec && m_run)
+          {
+            // Count only error-recovery re-arms (not the success-path re-arm or
+            // the operation_aborted shutdown path) for diagnostics / tests.
+            ++m_acceptErrorRearms;
+            m_acceptor.async_accept(net::make_strand(m_context),
+                                    beast::bind_front_handler(&Server::accept, this));
+          }
+        });
+      }
+      return;
     }
-    else if (m_run)
+
+    if (m_run)
     {
       auto dispatcher = [this](SessionPtr session, RequestPtr request) {
         if (!m_run)
@@ -226,7 +256,7 @@ namespace mtconnect::sink::rest_sink {
   // Report a failure
   void Server::fail(beast::error_code ec, char const *what)
   {
-    LOG(error) << " error: " << ec.message();
+    LOG(error) << what << " error: " << ec.message();
   }
 
   using namespace mtconnect::printer;

src/mtconnect/sink/rest_sink/server.hpp

@@ -20,10 +20,13 @@
 #include <boost/asio/connect.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/ssl.hpp>
+#include <boost/asio/steady_timer.hpp>
 #include <boost/beast/http/status.hpp>
 #include <boost/bind/bind.hpp>
 
+#include <atomic>
 #include <chrono>
+#include <cstdint>
 #include <cstdlib>
 #include <iostream>
 #include <mutex>
@@ -60,6 +63,7 @@ namespace mtconnect::sink::rest_sink {
         m_options(options),
         m_allowPuts(IsOptionSet(options, configuration::AllowPut)),
         m_acceptor(context),
+        m_acceptBackoff(context),
         m_sslContext(boost::asio::ssl::context::tls)
     {
       auto inter = GetOption<std::string>(options, configuration::ServerIp);
@@ -272,6 +276,10 @@ namespace mtconnect::sink::rest_sink {
     /// @return the error function
     ErrorFunction getErrorFunction() const { return m_errorFunction; }
 
+    /// @brief Number of times the accept loop re-armed after a recoverable accept error.
+    /// @return the cumulative re-arm count (useful for diagnostics / tests)
+    std::uint64_t getAcceptRearmCount() const { return m_acceptErrorRearms; }
+
     /// @name Only for testing
     ///@{
     /// @brief Callback for testing. Allows test to grab the last session dispatched.
@@ -318,6 +326,11 @@ namespace mtconnect::sink::rest_sink {
     std::optional<ParameterDocList> m_parameterDocumentation;
 
     boost::asio::ip::tcp::acceptor m_acceptor;
+    boost::asio::steady_timer m_acceptBackoff;
+    /// Cumulative count of accept-loop re-arms triggered by a recoverable accept
+    /// error (incremented only on the error-recovery path, not normal success or
+    /// shutdown). Exposed via getAcceptRearmCount() for diagnostics / tests.
+    std::atomic<std::uint64_t> m_acceptErrorRearms {0};
     boost::asio::ssl::context m_sslContext;
     bool m_tlsEnabled {false};
     bool m_tlsOnly {false};

test_package/CMakeLists.txt

@@ -289,6 +289,7 @@ add_agent_test(globals FALSE core)
 
 add_agent_test(config_parser FALSE configuration)
 add_agent_test(config FALSE configuration)
+add_agent_test(async_context FALSE configuration)
 
 add_agent_test(change_observer FALSE observation)
 add_agent_test(observation TRUE observation)

test_package/async_context_test.cpp

@@ -0,0 +1,70 @@
+//
+// Copyright Copyright 2009-2024, AMT – The Association For Manufacturing Technology (“AMT”)
+// All rights reserved.
+//
+//    Licensed under the Apache License, Version 2.0 (the "License");
+//    you may not use this file except in compliance with the License.
+//    You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//    Unless required by applicable law or agreed to in writing, software
+//    distributed under the License is distributed on an "AS IS" BASIS,
+//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//    See the License for the specific language governing permissions and
+//    limitations under the License.
+//
+
+// Ensure that gtest is the first header otherwise Windows raises an error
+#include <gtest/gtest.h>
+// Keep this comment to keep gtest.h above. (clang-format off/on is not working here!)
+
+#include <boost/asio/post.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <stdexcept>
+#include <thread>
+
+#include "mtconnect/configuration/async_context.hpp"
+
+using namespace std;
+using namespace mtconnect;
+using namespace mtconnect::configuration;
+
+// main
+int main(int argc, char* argv[])
+{
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+class AsyncContextTest : public testing::Test
+{
+};
+
+TEST_F(AsyncContextTest, worker_survives_a_thrown_handler)
+{
+  // Fix #3: a handler that throws must not kill the io_context worker. With a
+  // single worker thread handlers run FIFO on that one thread; without the fix
+  // the throwing handler tears down the worker and the second handler never
+  // runs. With the fix the worker catches, resumes run(), and processes it.
+  AsyncContext ctx;
+
+  boost::asio::post(ctx.get(), [] { throw std::runtime_error("boom"); });
+
+  std::atomic<bool> secondRan {false};
+  boost::asio::post(ctx.get(), [&] { secondRan = true; });
+
+  // start() blocks in its own run loop, so drive it on a separate thread.
+  std::thread t([&] { ctx.start(); });
+
+  // Poll up to ~2s so the test cannot hang even if the fix regresses.
+  for (int i = 0; i < 200 && !secondRan; ++i)
+    std::this_thread::sleep_for(10ms);
+
+  ctx.stop();
+  t.join();
+
+  EXPECT_TRUE(secondRan);
+}

test_package/http_server_test.cpp

@@ -678,6 +678,58 @@ TEST_F(HttpServerTest, additional_header_fields)
   ASSERT_EQ("https://foo.example", f2->second);
 }
 
+TEST_F(HttpServerTest, listen_throws_when_port_is_already_bound)
+{
+  // The fixture's server binds a random free port; capture the concrete port.
+  start();
+  auto port = m_server->getPort();
+
+  // A second server bound to the same concrete ip:port must fail to bind.
+  // SO_REUSEADDR is set, but two sockets concurrently listening on the same
+  // 127.0.0.1:port still yield EADDRINUSE on Linux, so this is deterministic.
+  //
+  // We call listen() directly rather than start(): start() catches the throw
+  // and calls std::exit(1), which would terminate the test runner.
+  asio::io_context otherContext;
+  Server other(otherContext,
+               ConfigOptions {{configuration::Port, static_cast<int>(port)},
+                              {configuration::ServerIp, "127.0.0.1"s}});
+  ASSERT_THROW(other.listen(), std::runtime_error);
+}
+
+TEST_F(HttpServerTest, accept_loop_rearms_after_recoverable_error)
+{
+  // Fix #1: on a recoverable accept error the accept loop must re-arm after a
+  // backoff instead of stopping permanently. getAcceptRearmCount() is a
+  // production seam that counts ONLY the error-recovery re-arm path, so a
+  // 0 -> 1 transition proves the fix actually re-armed the acceptor after the
+  // injected error (rather than the initial accept armed by start() simply
+  // serving a later connection, which would not distinguish fixed from broken).
+  //
+  // Injecting the error while the initial accept is still armed leaves two
+  // outstanding async_accept operations on the acceptor briefly. This is a
+  // supported Boost.Asio pattern (an acceptor may hold multiple concurrent
+  // pending accepts); both are cancelled cleanly (operation_aborted) when the
+  // server is destroyed in TearDown.
+  start();
+  ASSERT_EQ(0u, m_server->getAcceptRearmCount());
+
+  // Drive the recoverable-error branch directly with a fresh (unopened) socket;
+  // the error branch returns before touching the socket.
+  m_server->accept(boost::asio::error::connection_aborted, tcp::socket(m_context));
+
+  // Pump the io_context until the 100ms backoff timer fires and re-arms, with a
+  // bounded loop so a regression (no re-arm) fails fast instead of hanging.
+  for (int i = 0; i < 100 && m_server->getAcceptRearmCount() == 0; ++i)
+    m_context.run_for(20ms);
+
+  EXPECT_EQ(1u, m_server->getAcceptRearmCount());
+
+  // The server remains running and listening after recovering from the error.
+  EXPECT_TRUE(m_server->isRunning());
+  EXPECT_TRUE(m_server->isListening());
+}
+
 const string CertFile(TEST_RESOURCE_DIR "/user.crt");
 const string KeyFile {TEST_RESOURCE_DIR "/user.key"};
 const string DhFile {TEST_RESOURCE_DIR "/dh2048.pem"};