Verify socket open when handling err messages

Summary:
When a socket is configured to read from `MSG_ERRQUEUE`, the corresponding callback handler for `errMessageCallback_->errMessage(*cmsg)` may close the socket when processing a message.

If this happens, `fd_` will be set to -1 (reinitialized) and subsequent calls to `netops::recvmsg(fd_, &msg, MSG_ERRQUEUE)` will fail. On the read failure, an error will be propagated via `failErrMessageRead(__func__, ex)`. This quickly becomes confusing -- you end up with an error message `::recvmsg exited with code 9` (EBADF) with little context.

This diff adds logic to check if the socket is still open before each call to `netops::recvmsg(fd_, &msg, MSG_ERRQUEUE)`. If not, we exit out.

It's arguable that the application is at fault in this scenario instead of the library. However, I've found this problem and given the complexity, it seems like a scenario worth guarding against in the library.

Reviewed By: yfeldblum

Differential Revision: D14354115

fbshipit-source-id: 49e1efbbf6cecbe3bf6db7ee6bdfcac4c8aaa8c2

folly/io/async/AsyncSocket.cpp

@@ -1844,7 +1844,8 @@ size_t AsyncSocket::handleErrMessages() noexcept {
 
   int ret;
   size_t num = 0;
-  while (true) {
+  // the socket may be closed by errMessage callback, so check on each iteration
+  while (fd_ != NetworkSocket()) {
     ret = netops::recvmsg(fd_, &msg, MSG_ERRQUEUE);
     VLOG(5) << "AsyncSocket::handleErrMessages(): recvmsg returned " << ret;
 
@@ -1852,7 +1853,7 @@ size_t AsyncSocket::handleErrMessages() noexcept {
       if (errno != EAGAIN) {
         auto errnoCopy = errno;
         LOG(ERROR) << "::recvmsg exited with code " << ret
-                   << ", errno: " << errnoCopy;
+                   << ", errno: " << errnoCopy << ", fd: " << fd_;
         AsyncSocketException ex(
             AsyncSocketException::INTERNAL_ERROR,
             withAddr("recvmsg() failed"),
@@ -1876,6 +1877,7 @@ size_t AsyncSocket::handleErrMessages() noexcept {
       }
     }
   }
+  return num;
 #else
   return 0;
 #endif // FOLLY_HAVE_MSG_ERRQUEUE

folly/io/async/test/AsyncSocketTest2.cpp

@@ -3032,6 +3032,94 @@ enum SOF_TIMESTAMPING {
   SOF_TIMESTAMPING_OPT_TSONLY = (1 << 11),
 };
 
+struct AsyncSocketErrMessageCallbackTestParams {
+  folly::Optional<int> resetCallbackAfter;
+  folly::Optional<int> closeSocketAfter;
+  int gotTimestampExpected{0};
+  int gotByteSeqExpected{0};
+};
+
+class AsyncSocketErrMessageCallbackTest
+    : public ::testing::TestWithParam<AsyncSocketErrMessageCallbackTestParams> {
+ public:
+  static std::vector<AsyncSocketErrMessageCallbackTestParams>
+  getTestingValues() {
+    std::vector<AsyncSocketErrMessageCallbackTestParams> vals;
+    // each socket err message triggers two socket callbacks:
+    //   (1) timestamp callback
+    //   (2) byteseq callback
+
+    // reset callback cases
+    // resetting the callback should prevent any further callbacks
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.resetCallbackAfter = 1;
+      params.gotTimestampExpected = 1;
+      params.gotByteSeqExpected = 0;
+      vals.push_back(params);
+    }
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.resetCallbackAfter = 2;
+      params.gotTimestampExpected = 1;
+      params.gotByteSeqExpected = 1;
+      vals.push_back(params);
+    }
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.resetCallbackAfter = 3;
+      params.gotTimestampExpected = 2;
+      params.gotByteSeqExpected = 1;
+      vals.push_back(params);
+    }
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.resetCallbackAfter = 4;
+      params.gotTimestampExpected = 2;
+      params.gotByteSeqExpected = 2;
+      vals.push_back(params);
+    }
+
+    // close socket cases
+    // closing the socket will prevent callbacks after the current err message
+    // callbacks (both timestamp and byteseq) are completed
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.closeSocketAfter = 1;
+      params.gotTimestampExpected = 1;
+      params.gotByteSeqExpected = 1;
+      vals.push_back(params);
+    }
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.closeSocketAfter = 2;
+      params.gotTimestampExpected = 1;
+      params.gotByteSeqExpected = 1;
+      vals.push_back(params);
+    }
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.closeSocketAfter = 3;
+      params.gotTimestampExpected = 2;
+      params.gotByteSeqExpected = 2;
+      vals.push_back(params);
+    }
+    {
+      AsyncSocketErrMessageCallbackTestParams params;
+      params.closeSocketAfter = 4;
+      params.gotTimestampExpected = 2;
+      params.gotByteSeqExpected = 2;
+      vals.push_back(params);
+    }
+    return vals;
+  }
+};
+
+INSTANTIATE_TEST_CASE_P(
+    ErrMessageTests,
+    AsyncSocketErrMessageCallbackTest,
+    ::testing::ValuesIn(AsyncSocketErrMessageCallbackTest::getTestingValues()));
+
 class TestErrMessageCallback : public folly::AsyncSocket::ErrMessageCallback {
  public:
   TestErrMessageCallback()
@@ -3041,11 +3129,13 @@ class TestErrMessageCallback : public folly::AsyncSocket::ErrMessageCallback {
     if (cmsg.cmsg_level == SOL_SOCKET && cmsg.cmsg_type == SCM_TIMESTAMPING) {
       gotTimestamp_++;
       checkResetCallback();
+      checkCloseSocket();
     } else if (
         (cmsg.cmsg_level == SOL_IP && cmsg.cmsg_type == IP_RECVERR) ||
         (cmsg.cmsg_level == SOL_IPV6 && cmsg.cmsg_type == IPV6_RECVERR)) {
       gotByteSeq_++;
       checkResetCallback();
+      checkCloseSocket();
     }
   }
 
@@ -3055,20 +3145,28 @@ class TestErrMessageCallback : public folly::AsyncSocket::ErrMessageCallback {
   }
 
   void checkResetCallback() noexcept {
-    if (socket_ != nullptr && resetAfter_ != -1 &&
-        gotTimestamp_ + gotByteSeq_ == resetAfter_) {
+    if (socket_ != nullptr && resetCallbackAfter_ != -1 &&
+        gotTimestamp_ + gotByteSeq_ == resetCallbackAfter_) {
       socket_->setErrMessageCB(nullptr);
     }
   }
 
+  void checkCloseSocket() noexcept {
+    if (socket_ != nullptr && closeSocketAfter_ != -1 &&
+        gotTimestamp_ + gotByteSeq_ == closeSocketAfter_) {
+      socket_->close();
+    }
+  }
+
   folly::AsyncSocket* socket_{nullptr};
   folly::AsyncSocketException exception_;
   int gotTimestamp_{0};
   int gotByteSeq_{0};
-  int resetAfter_{-1};
+  int resetCallbackAfter_{-1};
+  int closeSocketAfter_{-1};
 };
 
-TEST(AsyncSocketTest, ErrMessageCallback) {
+TEST_P(AsyncSocketErrMessageCallbackTest, ErrMessageCallback) {
   TestServer server;
 
   // connect()
@@ -3097,8 +3195,15 @@ TEST(AsyncSocketTest, ErrMessageCallback) {
       socket->getErrMessageCallback(),
       static_cast<folly::AsyncSocket::ErrMessageCallback*>(&errMsgCB));
 
+  // set the number of error messages before socket is closed or callback reset
+  const auto testParams = GetParam();
   errMsgCB.socket_ = socket.get();
-  errMsgCB.resetAfter_ = 3;
+  if (testParams.resetCallbackAfter.hasValue()) {
+    errMsgCB.resetCallbackAfter_ = testParams.resetCallbackAfter.value();
+  }
+  if (testParams.closeSocketAfter.hasValue()) {
+    errMsgCB.closeSocketAfter_ = testParams.closeSocketAfter.value();
+  }
 
   // Enable timestamp notifications
   ASSERT_NE(socket->getNetworkSocket(), NetworkSocket());
@@ -3139,10 +3244,8 @@ TEST(AsyncSocketTest, ErrMessageCallback) {
   // Check for the timestamp notifications.
   ASSERT_EQ(
       errMsgCB.exception_.getType(), folly::AsyncSocketException::UNKNOWN);
-  ASSERT_GT(errMsgCB.gotByteSeq_, 0);
-  ASSERT_GT(errMsgCB.gotTimestamp_, 0);
-  ASSERT_EQ(
-      errMsgCB.gotByteSeq_ + errMsgCB.gotTimestamp_, errMsgCB.resetAfter_);
+  ASSERT_EQ(errMsgCB.gotByteSeq_, testParams.gotByteSeqExpected);
+  ASSERT_EQ(errMsgCB.gotTimestamp_, testParams.gotTimestampExpected);
 }
 #endif // FOLLY_HAVE_MSG_ERRQUEUE