[BUG] Stop embedded HTTP server writes from raising SIGPIPE

CHANGELOG.md

@@ -19,6 +19,9 @@ Increment the:
   namespaces
   [#4303](https://github.com/open-telemetry/opentelemetry-cpp/pull/4303)
 
+* [BUG] Stop embedded HTTP server writes from raising SIGPIPE
+  [#4294](https://github.com/open-telemetry/opentelemetry-cpp/pull/4294)
+
 * [CODE HEALTH] Move metrics storage test fixtures into anonymous namespace
   [#4286](https://github.com/open-telemetry/opentelemetry-cpp/pull/4286)
 

ext/include/opentelemetry/ext/http/server/socket_tools.h

@@ -280,7 +280,7 @@ struct Socket
 
   Socket(Type sock = Invalid) : m_sock(sock) {}
 
-  Socket(int af, int type, int proto) : m_sock(::socket(af, type, proto)) {}
+  Socket(int af, int type, int proto) : m_sock(::socket(af, type, proto)) { suppressSigPipe(); }
 
   operator Socket::Type() const { return m_sock; }
 
@@ -345,6 +345,35 @@ struct Socket
     m_sock = Invalid;
   }
 
+  /**
+   * Stop writes to a socket whose peer has gone away from raising SIGPIPE, whose default
+   * disposition would terminate the host process. The guarantee is MSG_NOSIGNAL, which send()
+   * passes per call on every platform that defines it, which is modern Linux, macOS and the BSDs,
+   * and every platform this repo builds on. Where the platform also offers SO_NOSIGPIPE this
+   * additionally sets it at the socket level as a best-effort second layer, covering a write to
+   * the descriptor that does not go through send(). Windows has no SIGPIPE. Done per socket rather
+   * than by changing the process signal disposition, which belongs to the embedding program rather
+   * than to this library.
+   */
+  void suppressSigPipe()
+  {
+#ifdef SO_NOSIGPIPE
+    if (m_sock != Invalid)
+    {
+      int value = 1;
+      if (::setsockopt(m_sock, SOL_SOCKET, SO_NOSIGPIPE, reinterpret_cast<char *>(&value),
+                       sizeof(value)) != 0)
+      {
+        // Best effort: MSG_NOSIGNAL already guards send() on every platform this repo builds on,
+        // so a failure here is logged where a logger is configured rather than failing the socket.
+        // This wrapper does not own the descriptor's lifetime and does not promise to protect a
+        // platform whose only defence would be SO_NOSIGPIPE.
+        LOG_WARN("Socket: cannot set SO_NOSIGPIPE (errno %d)", errno);
+      }
+    }
+#endif
+  }
+
   int recv(void *buffer, unsigned size)
   {
     assert(m_sock != Invalid);
@@ -355,7 +384,12 @@ struct Socket
   int send(void const *buffer, unsigned size)
   {
     assert(m_sock != Invalid);
-    return static_cast<int>(::send(m_sock, reinterpret_cast<char const *>(buffer), size, 0));
+#ifdef MSG_NOSIGNAL
+    int const flags = MSG_NOSIGNAL;
+#else
+    int const flags = 0;
+#endif
+    return static_cast<int>(::send(m_sock, reinterpret_cast<char const *>(buffer), size, flags));
   }
 
   bool bind(SocketAddr const &addr)
@@ -390,6 +424,7 @@ struct Socket
     socklen_t addrlen = sizeof(caddr);
 #endif
     csock = ::accept(m_sock, caddr, &addrlen);
+    csock.suppressSigPipe();
     return !csock.invalid();
   }
 

ext/test/http/BUILD

@@ -3,6 +3,24 @@
 
 load("@rules_cc//cc:cc_test.bzl", "cc_test")
 
+cc_test(
+    name = "socket_sigpipe_test",
+    srcs = [
+        "socket_sigpipe_test.cc",
+    ],
+    # socket_tools.h instantiates a WsaInitializer global on Windows (WSAStartup), so the Windows
+    # build links ws2_32 even though the SIGPIPE test body itself is compiled out there.
+    linkopts = select({
+        "//bazel:windows": ["-DEFAULTLIB:Ws2_32.lib"],
+        "//conditions:default": [],
+    }),
+    tags = ["test"],
+    deps = [
+        "//ext:headers",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "curl_http_test",
     srcs = [

ext/test/http/CMakeLists.txt

@@ -14,6 +14,22 @@ if(WITH_HTTP_CLIENT_CURL)
     TEST_LIST ${FILENAME})
 endif()
 
+# The SIGPIPE suppression is POSIX-only (Windows has no such signal), so the
+# CMake target is built only on POSIX; the Bazel target covers Windows with an
+# explicit skipped test. gtest_discover_tests enumerates the compiled binary
+# rather than parsing the source, so the SO_NOSIGPIPE-only tests are registered
+# only where they are actually compiled, never as phantom tests on a platform
+# that compiled them out.
+if(NOT WIN32)
+  set(SOCKET_SIGPIPE_FILENAME socket_sigpipe_test)
+  add_executable(${SOCKET_SIGPIPE_FILENAME} ${SOCKET_SIGPIPE_FILENAME}.cc)
+  target_link_libraries(
+    ${SOCKET_SIGPIPE_FILENAME} opentelemetry_ext ${GMOCK_LIB}
+    ${GTEST_BOTH_LIBRARIES} ${CMAKE_THREAD_LIBS_INIT})
+  gtest_discover_tests(${SOCKET_SIGPIPE_FILENAME}
+                       TEST_PREFIX ext.http.socketsigpipe.)
+endif()
+
 set(URL_PARSER_FILENAME url_parser_test)
 add_executable(${URL_PARSER_FILENAME} ${URL_PARSER_FILENAME}.cc)
 target_link_libraries(${URL_PARSER_FILENAME} opentelemetry_ext ${GMOCK_LIB}

ext/test/http/socket_sigpipe_test.cc

@@ -0,0 +1,182 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include <gtest/gtest.h>
+
+#include "opentelemetry/ext/http/server/socket_tools.h"
+
+// SIGPIPE only exists on POSIX. Windows has no such signal and the suppression code is compiled
+// out there, so the POSIX tests live behind _WIN32 and Windows gets an explicit skipped test
+// rather than an empty binary.
+#ifndef _WIN32
+
+#  include <errno.h>
+#  include <pthread.h>
+#  include <signal.h>
+#  include <sys/socket.h>
+#  include <unistd.h>
+
+namespace
+{
+
+// The mask change and any pending SIGPIPE are cleaned up in TearDown, which runs even when a test
+// aborts on a fatal assertion. That matters here: Socket is a non-owning wrapper with no
+// destructor, and a leaked signal mask or a queued SIGPIPE would pollute later tests sharing this
+// binary.
+class SocketSigPipeTest : public ::testing::Test
+{
+protected:
+  void SetUp() override
+  {
+    sigemptyset(&blocked_);
+    sigaddset(&blocked_, SIGPIPE);
+    // Observe SIGPIPE through sigpending() instead of the default disposition, so a stray signal
+    // cannot kill the runner and the test does not depend on the process-wide handler.
+    ASSERT_EQ(pthread_sigmask(SIG_BLOCK, &blocked_, &previous_), 0);
+    mask_changed_ = true;
+    EXPECT_FALSE(SigPipeIsPending()) << "a SIGPIPE was already pending before the test";
+  }
+
+  void TearDown() override
+  {
+    DrainPendingSigPipe();
+    if (mask_changed_)
+    {
+      EXPECT_EQ(pthread_sigmask(SIG_SETMASK, &previous_, nullptr), 0);
+    }
+  }
+
+  static bool SigPipeIsPending()
+  {
+    sigset_t pending;
+    sigemptyset(&pending);
+    EXPECT_EQ(sigpending(&pending), 0);
+    return sigismember(&pending, SIGPIPE) == 1;
+  }
+
+  void DrainPendingSigPipe()
+  {
+    if (SigPipeIsPending())
+    {
+      int signo = 0;
+      EXPECT_EQ(sigwait(&blocked_, &signo), 0);
+      EXPECT_EQ(signo, SIGPIPE);
+    }
+  }
+
+  sigset_t blocked_{};
+  sigset_t previous_{};
+  bool mask_changed_{false};
+};
+
+// Writing to a socket whose peer has closed must report EPIPE without raising SIGPIPE, whose
+// default disposition would terminate the host process. A closed peer fails the write immediately,
+// so one byte is enough. A negative control on a raw socket first proves this thread really does
+// observe a SIGPIPE, so the suppressed result is meaningful rather than vacuous.
+TEST_F(SocketSigPipeTest, WriteToClosedPeerReportsEpipeWithoutRaisingSigPipe)
+{
+#  if !defined(MSG_NOSIGNAL) && !defined(SO_NOSIGPIPE)
+  GTEST_SKIP()
+      << "no SIGPIPE-suppression mechanism (MSG_NOSIGNAL or SO_NOSIGPIPE) on this platform";
+#  else
+  const char byte = 'x';
+
+  // Negative control: a raw send() with no suppression to a closed peer raises SIGPIPE.
+  {
+    int raw[2];
+    ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, raw), 0);
+    ::close(raw[1]);
+    errno           = 0;
+    const ssize_t r = ::send(raw[0], &byte, 1, 0);
+    EXPECT_EQ(r, -1);
+    EXPECT_EQ(errno, EPIPE);
+    EXPECT_TRUE(SigPipeIsPending()) << "the environment does not deliver SIGPIPE, so the "
+                                       "suppressed case below would pass vacuously";
+    DrainPendingSigPipe();
+    ::close(raw[0]);
+  }
+
+  // The wrapper suppresses it: EPIPE, and no SIGPIPE queued.
+  int fds[2];
+  ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, fds), 0);
+  SocketTools::Socket sender(fds[0]);
+  // Production sockets get SO_NOSIGPIPE from accept()/the (af,type,proto) ctor. Apply it here too,
+  // since on macOS and the BSDs, where MSG_NOSIGNAL is not defined, that option is the layer under
+  // test.
+  sender.suppressSigPipe();
+  ::close(fds[1]);
+
+  errno           = 0;
+  const int sent  = sender.send(&byte, 1);
+  const int saved = errno;
+  EXPECT_EQ(sent, -1);
+  EXPECT_EQ(saved, EPIPE);
+  EXPECT_FALSE(SigPipeIsPending());
+
+  sender.close();
+#  endif
+}
+
+#  ifdef SO_NOSIGPIPE
+// Where the platform offers SO_NOSIGPIPE (macOS, the BSDs), the (af,type,proto) constructor sets
+// it.
+TEST_F(SocketSigPipeTest, CreatedSocketCarriesSoNoSigPipe)
+{
+  SocketTools::Socket sock(AF_INET, SOCK_STREAM, 0);
+  ASSERT_FALSE(sock.invalid());
+
+  int value        = 0;
+  socklen_t length = sizeof(value);
+  ASSERT_EQ(::getsockopt(sock, SOL_SOCKET, SO_NOSIGPIPE, &value, &length), 0);
+  EXPECT_EQ(value, 1);
+
+  sock.close();
+}
+
+// accept() must set SO_NOSIGPIPE on the socket it returns. The listener is wrapped from a raw fd
+// (the Socket(Type) constructor does not call the helper) and the test drives the wrapper's own
+// accept(), so a pass proves accept()'s suppressSigPipe() call rather than option inheritance.
+TEST_F(SocketSigPipeTest, AcceptedSocketCarriesSoNoSigPipe)
+{
+  SocketTools::Socket listener(::socket(AF_INET, SOCK_STREAM, 0));
+  ASSERT_FALSE(listener.invalid());
+  listener.setReuseAddr();
+
+  SocketTools::SocketAddr bind_addr(SocketTools::SocketAddr::Loopback, 0);  // 127.0.0.1, ephemeral
+  ASSERT_TRUE(listener.bind(bind_addr));
+  ASSERT_TRUE(listener.listen(1));
+
+  SocketTools::SocketAddr bound;
+  ASSERT_TRUE(listener.getsockname(bound));
+
+  SocketTools::Socket client(AF_INET, SOCK_STREAM, 0);
+  ASSERT_FALSE(client.invalid());
+  SocketTools::SocketAddr server_addr(SocketTools::SocketAddr::Loopback, bound.port());
+  // Blocking connect on loopback completes into the listen backlog before accept() runs.
+  ASSERT_TRUE(client.connect(server_addr));
+
+  SocketTools::Socket accepted;
+  SocketTools::SocketAddr peer;
+  ASSERT_TRUE(listener.accept(accepted, peer));  // accept() calls suppressSigPipe() on the result
+
+  int value        = 0;
+  socklen_t length = sizeof(value);
+  ASSERT_EQ(::getsockopt(accepted, SOL_SOCKET, SO_NOSIGPIPE, &value, &length), 0);
+  EXPECT_EQ(value, 1);
+
+  accepted.close();
+  client.close();
+  listener.close();
+}
+#  endif  // SO_NOSIGPIPE
+
+}  // namespace
+
+#else  // _WIN32
+
+TEST(SocketSigPipeTest, NotApplicableOnWindows)
+{
+  GTEST_SKIP() << "Windows does not generate POSIX SIGPIPE";
+}
+
+#endif  // _WIN32