Make RemoteConnectionImpl thread-safe (#42380)

Summary: Pull Request resolved: https://github.com/facebook/react-native/pull/42380 Changelog: [Internal] Makes it explicitly legal to call `IRemoteConnection`'s methods from any thread when used as part of the C++ implementation of `InspectorPackagerConnection`. Implementation details: * This relies on `InspectorPackagerConnectionDelegate::scheduleCallback` being thread-safe and handling any necessary synchronisation (which is already required for the existing `reconnect()` use case). * We add *very basic* tracking of *sessions* within `InspectorPackagerConnection` to make sure events don't leak from one `RemoteConnection` instance to the next. * In the future we'll want to build on this to properly allow multiple concurrent sessions to a single page. That's not the primary goal here though. Reviewed By: rubennorte Differential Revision: D52807388 fbshipit-source-id: 6900386a1f047c99f15dc91597f308c82adf5281
2025-11-01 09:14:26 +00:00 · 2024-01-19 09:26:16 -08:00
parent fd0ca4dd62
commit 62117b304e
6 changed files with 323 additions and 60 deletions
@@ -41,6 +41,8 @@ struct InspectorPageDescription {
 using InspectorPage = InspectorPageDescription;

 /// IRemoteConnection allows the VM to send debugger messages to the client.
+/// IRemoteConnection's methods are safe to call from any thread *if*
+/// InspectorPackagerConnection.cpp is in use.
 class JSINSPECTOR_EXPORT IRemoteConnection : public IDestructible {
 public:
  virtual ~IRemoteConnection() = 0;
@@ -15,6 +15,8 @@
 #include <cerrno>
 #include <chrono>

+using namespace std::literals;
+
 namespace facebook::react::jsinspector_modern {

 static constexpr const std::chrono::duration RECONNECT_DELAY =
@@ -51,7 +53,8 @@ void InspectorPackagerConnection::Impl::handleProxyMessage(
    folly::const_dynamic_view message) {
  std::string event = message.descend("event").string_or(INVALID);
  if (event == "getPages") {
-    sendEvent("getPages", pages());
+    sendToPackager(
+        folly::dynamic::object("event", "getPages")("payload", pages()));
  } else if (event == "wrappedEvent") {
    handleWrappedEvent(message.descend("payload"));
  } else if (event == "connect") {
@@ -65,26 +68,26 @@ void InspectorPackagerConnection::Impl::handleProxyMessage(

 void InspectorPackagerConnection::Impl::sendEventToAllConnections(
    std::string event) {
-  for (auto& connection : inspectorConnections_) {
-    connection.second->sendMessage(event);
+  for (auto& connection : inspectorSessions_) {
+    connection.second.localConnection->sendMessage(event);
  }
 }

 void InspectorPackagerConnection::Impl::closeAllConnections() {
-  for (auto& connection : inspectorConnections_) {
-    connection.second->disconnect();
+  for (auto& connection : inspectorSessions_) {
+    connection.second.localConnection->disconnect();
  }
-  inspectorConnections_.clear();
+  inspectorSessions_.clear();
 }

 void InspectorPackagerConnection::Impl::handleConnect(
    folly::const_dynamic_view payload) {
  std::string pageId = payload.descend("pageId").string_or(INVALID);
-  auto existingConnectionIt = inspectorConnections_.find(pageId);
-  if (existingConnectionIt != inspectorConnections_.end()) {
+  auto existingConnectionIt = inspectorSessions_.find(pageId);
+  if (existingConnectionIt != inspectorSessions_.end()) {
    auto existingConnection = std::move(existingConnectionIt->second);
-    inspectorConnections_.erase(existingConnectionIt);
-    existingConnection->disconnect();
+    inspectorSessions_.erase(existingConnectionIt);
+    existingConnection.localConnection->disconnect();
    LOG(WARNING) << "Already connected: " << pageId;
    return;
  }
@@ -95,13 +98,18 @@ void InspectorPackagerConnection::Impl::handleConnect(
    LOG(ERROR) << "Invalid page id: " << pageId;
    return;
  }
+  auto sessionId = nextSessionId_++;
  auto remoteConnection =
-      std::make_unique<InspectorPackagerConnection::RemoteConnectionImpl>(
-          weak_from_this(), pageId);
+      std::make_unique<InspectorPackagerConnection::Impl::RemoteConnection>(
+          weak_from_this(), pageId, sessionId);
  auto& inspector = getInspectorInstance();
  auto inspectorConnection =
      inspector.connect(pageIdInt, std::move(remoteConnection));
-  inspectorConnections_.emplace(pageId, std::move(inspectorConnection));
+  inspectorSessions_.emplace(
+      pageId,
+      Session{
+          .localConnection = std::move(inspectorConnection),
+          .sessionId = sessionId});
 }

 void InspectorPackagerConnection::Impl::handleDisconnect(
@@ -115,11 +123,11 @@ void InspectorPackagerConnection::Impl::handleDisconnect(

 std::unique_ptr<ILocalConnection>
 InspectorPackagerConnection::Impl::removeConnectionForPage(std::string pageId) {
-  auto it = inspectorConnections_.find(pageId);
-  if (it != inspectorConnections_.end()) {
+  auto it = inspectorSessions_.find(pageId);
+  if (it != inspectorSessions_.end()) {
    auto connection = std::move(it->second);
-    inspectorConnections_.erase(it);
-    return connection;
+    inspectorSessions_.erase(it);
+    return std::move(connection.localConnection);
  }
  return nullptr;
 }
@@ -128,13 +136,13 @@ void InspectorPackagerConnection::Impl::handleWrappedEvent(
    folly::const_dynamic_view payload) {
  std::string pageId = payload.descend("pageId").string_or(INVALID);
  std::string wrappedEvent = payload.descend("wrappedEvent").string_or(INVALID);
-  auto connectionIt = inspectorConnections_.find(pageId);
-  if (connectionIt == inspectorConnections_.end()) {
+  auto connectionIt = inspectorSessions_.find(pageId);
+  if (connectionIt == inspectorSessions_.end()) {
    LOG(WARNING) << "Not connected to page: " << pageId
                 << " , failed trying to handle event: " << wrappedEvent;
    return;
  }
-  connectionIt->second->sendMessage(wrappedEvent);
+  connectionIt->second.localConnection->sendMessage(wrappedEvent);
 }

 folly::dynamic InspectorPackagerConnection::Impl::pages() {
@@ -149,22 +157,6 @@ folly::dynamic InspectorPackagerConnection::Impl::pages() {
  return array;
 }

-void InspectorPackagerConnection::Impl::sendWrappedEvent(
-    std::string pageId,
-    std::string message) {
-  sendEvent(
-      "wrappedEvent",
-      folly::dynamic::object("pageId", pageId)("wrappedEvent", message));
-}
-
-void InspectorPackagerConnection::Impl::sendEvent(
-    std::string event,
-    folly::dynamic payload) {
-  folly::dynamic message =
-      folly::dynamic::object("event", event)("payload", payload);
-  sendToPackager(message);
-}
-
 void InspectorPackagerConnection::Impl::didFailWithError(
    std::optional<int> posixCode,
    std::string error) {
@@ -258,6 +250,28 @@ void InspectorPackagerConnection::Impl::sendToPackager(folly::dynamic message) {
  webSocket_->send(folly::toJson(message));
 }

+void InspectorPackagerConnection::Impl::scheduleSendToPackager(
+    folly::dynamic message,
+    SessionId sourceSessionId,
+    std::string sourcePageId) {
+  delegate_->scheduleCallback(
+      [weakSelf = weak_from_this(),
+       message = std::move(message),
+       sourceSessionId,
+       sourcePageId]() mutable {
+        auto strongSelf = weakSelf.lock();
+        if (!strongSelf) {
+          return;
+        }
+        auto sessionIt = strongSelf->inspectorSessions_.find(sourcePageId);
+        if (sessionIt != strongSelf->inspectorSessions_.end() &&
+            sessionIt->second.sessionId == sourceSessionId) {
+          strongSelf->sendToPackager(std::move(message));
+        }
+      },
+      0ms);
+}
+
 void InspectorPackagerConnection::Impl::abort(
    std::optional<int> posixCode,
    const std::string& message,
@@ -276,28 +290,38 @@ void InspectorPackagerConnection::Impl::disposeWebSocket() {
  webSocket_.reset();
 }

-// InspectorPackagerConnection::RemoteConnectionImpl method definitions
+// InspectorPackagerConnection::Impl::RemoteConnection method definitions

-InspectorPackagerConnection::RemoteConnectionImpl::RemoteConnectionImpl(
+InspectorPackagerConnection::Impl::RemoteConnection::RemoteConnection(
    std::weak_ptr<InspectorPackagerConnection::Impl> owningPackagerConnection,
-    std::string pageId)
+    std::string pageId,
+    SessionId sessionId)
    : owningPackagerConnection_(owningPackagerConnection),
-      pageId_(std::move(pageId)) {}
+      pageId_(std::move(pageId)),
+      sessionId_(sessionId) {}

-void InspectorPackagerConnection::RemoteConnectionImpl::onMessage(
+void InspectorPackagerConnection::Impl::RemoteConnection::onMessage(
    std::string message) {
  auto owningPackagerConnectionStrong = owningPackagerConnection_.lock();
  if (!owningPackagerConnectionStrong) {
    return;
  }
-  owningPackagerConnectionStrong->sendWrappedEvent(pageId_, message);
+  owningPackagerConnectionStrong->scheduleSendToPackager(
+      folly::dynamic::object("event", "wrappedEvent")(
+          "payload",
+          folly::dynamic::object("pageId", pageId_)("wrappedEvent", message)),
+      sessionId_,
+      pageId_);
 }

-void InspectorPackagerConnection::RemoteConnectionImpl::onDisconnect() {
+void InspectorPackagerConnection::Impl::RemoteConnection::onDisconnect() {
  auto owningPackagerConnectionStrong = owningPackagerConnection_.lock();
  if (owningPackagerConnectionStrong) {
-    owningPackagerConnectionStrong->sendEvent(
-        "disconnect", makePageIdPayload(pageId_));
+    owningPackagerConnectionStrong->scheduleSendToPackager(
+        folly::dynamic::object("event", "disconnect")(
+            "payload", makePageIdPayload(pageId_)),
+        sessionId_,
+        pageId_);
  }
 }

@@ -48,7 +48,6 @@ class InspectorPackagerConnection {

 private:
  class Impl;
-  class RemoteConnectionImpl;

  const std::shared_ptr<Impl> impl_;
 };
@@ -72,10 +71,11 @@ class InspectorPackagerConnectionDelegate {
  /**
   * Schedules a function to run after a delay. If the function is called
   * asynchronously, the implementer of InspectorPackagerConnectionDelegate
-   * is responsible for thread safety (e.g. scheduling the callback on the same
-   * thread that called scheduleCallback, or otherwise ensuring
-   * synchronization). The callback MAY be dropped and never called, e.g. if the
-   * application is terminating.
+   * is responsible for thread safety (e.g. scheduling the callback on a thread
+   * that has unique access to the InspectorPackagerConnection instance, or
+   * otherwise ensuring synchronization). The callback MAY be dropped and never
+   * called if no further callbacks are being accepted, e.g. if the application
+   * is terminating.
   */
  virtual void scheduleCallback(
      std::function<void(void)> callback,
@@ -24,6 +24,8 @@ class InspectorPackagerConnection::Impl
      // Used to generate `weak_ptr`s we can pass around.
      public std::enable_shared_from_this<InspectorPackagerConnection::Impl> {
 public:
+  using SessionId = uint32_t;
+
  /**
   * Implements InspectorPackagerConnection's constructor.
   */
@@ -39,12 +41,24 @@ class InspectorPackagerConnection::Impl
  void sendEventToAllConnections(std::string event);
  std::unique_ptr<ILocalConnection> removeConnectionForPage(std::string pageId);

-  // Exposed for RemoteConnectionImpl's use
-  void sendEvent(std::string event, folly::dynamic payload);
-  // Exposed for RemoteConnectionImpl's use
-  void sendWrappedEvent(std::string pageId, std::string message);
+  /**
+   * Send a message to the packager as soon as possible. This method is safe
+   * to call from any thread. The connection may be closed before the message
+   * is sent, in which case the message will be dropped. The message is also
+   * dropped if the session is no longer valid.
+   */
+  void scheduleSendToPackager(
+      folly::dynamic message,
+      SessionId sourceSessionId,
+      std::string sourcePageId);

 private:
+  struct Session {
+    std::unique_ptr<ILocalConnection> localConnection;
+    SessionId sessionId;
+  };
+  class RemoteConnection;
+
  Impl(
      std::string url,
      std::string app,
@@ -80,22 +94,24 @@ class InspectorPackagerConnection::Impl
  const std::string app_;
  const std::unique_ptr<InspectorPackagerConnectionDelegate> delegate_;

-  std::unordered_map<std::string, std::unique_ptr<ILocalConnection>>
-      inspectorConnections_;
+  std::unordered_map<std::string, Session> inspectorSessions_;
  std::unique_ptr<IWebSocket> webSocket_;
  bool closed_{false};
  bool suppressConnectionErrors_{false};

  // Whether a reconnection is currently pending.
  bool reconnectPending_{false};
+
+  SessionId nextSessionId_{1};
 };

-class InspectorPackagerConnection::RemoteConnectionImpl
+class InspectorPackagerConnection::Impl::RemoteConnection
    : public IRemoteConnection {
 public:
-  RemoteConnectionImpl(
+  RemoteConnection(
      std::weak_ptr<InspectorPackagerConnection::Impl> owningPackagerConnection,
-      std::string pageId);
+      std::string pageId,
+      SessionId sessionId);

  // IRemoteConnection methods
  void onMessage(std::string message) override;
@@ -105,6 +121,7 @@ class InspectorPackagerConnection::RemoteConnectionImpl
  const std::weak_ptr<InspectorPackagerConnection::Impl>
      owningPackagerConnection_;
  const std::string pageId_;
+  const SessionId sessionId_;
 };

 } // namespace facebook::react::jsinspector_modern
@@ -59,12 +59,16 @@ class MockLocalConnection : public ILocalConnection {
    return *remoteConnection_;
  }

+  std::unique_ptr<IRemoteConnection> dangerouslyReleaseRemoteConnection() {
+    return std::move(remoteConnection_);
+  }
+
  // ILocalConnection methods
  MOCK_METHOD(void, sendMessage, (std::string message), (override));
  MOCK_METHOD(void, disconnect, (), (override));

 private:
-  const std::unique_ptr<IRemoteConnection> remoteConnection_;
+  std::unique_ptr<IRemoteConnection> remoteConnection_;
 };

 class MockInspectorPackagerConnectionDelegate
@@ -960,4 +960,220 @@ TEST_F(InspectorPackagerConnectionTest, TestDestroyConnectionOnPageRemoved) {
  getInspectorInstance().removePage(pageId);
  EXPECT_FALSE(localConnections_[0]);
 }
+
+TEST_F(
+    InspectorPackagerConnectionTestAsync,
+    TestAttemptSendToRemoteAfterDestroyed) {
+  // Configure gmock to expect calls in a specific order.
+  InSequence mockCallsMustBeInSequence;
+
+  packagerConnection_->connect();
+  auto pageId = getInspectorInstance().addPage(
+      "mock-title",
+      "mock-vm",
+      localConnections_
+          .lazily_make_unique<std::unique_ptr<IRemoteConnection>>());
+
+  // Connect to the page.
+  webSockets_[0]->getDelegate().didReceiveMessage(sformat(
+      R"({{
+          "event": "connect",
+          "payload": {{
+            "pageId": {0}
+          }}
+        }})",
+      toJson(std::to_string(pageId))));
+  ASSERT_TRUE(localConnections_[0]);
+
+  // Send an event from the mocked backend (local) to the frontend (remote)
+  // but don't flush the callback queue yet.
+  localConnections_[0]->getRemoteConnection().onMessage(R"({
+                                                            "method": "FakeDomain.eventTriggered",
+                                                            "params": ["arg1", "arg2"]
+                                                          })");
+
+  EXPECT_CALL(*localConnections_[0], disconnect()).RetiresOnSaturation();
+  getInspectorInstance().removePage(pageId);
+
+  packagerConnection_.reset();
+
+  // Flush the callback queue. This doesn't crash.
+  EXPECT_EQ(asyncExecutor_.run(), 1);
+}
+
+TEST_F(
+    InspectorPackagerConnectionTestAsync,
+    TestAttemptSendToStaleRemoteConnection) {
+  // Configure gmock to expect calls in a specific order.
+  InSequence mockCallsMustBeInSequence;
+
+  packagerConnection_->connect();
+  auto pageId = getInspectorInstance().addPage(
+      "mock-title",
+      "mock-vm",
+      localConnections_
+          .lazily_make_unique<std::unique_ptr<IRemoteConnection>>());
+
+  // Connect to the page.
+  webSockets_[0]->getDelegate().didReceiveMessage(sformat(
+      R"({{
+          "event": "connect",
+          "payload": {{
+            "pageId": {0}
+          }}
+        }})",
+      toJson(std::to_string(pageId))));
+  ASSERT_TRUE(localConnections_[0]);
+
+  // Send an event from the mocked backend (local) to the frontend (remote)
+  // but don't flush the callback queue yet.
+  localConnections_[0]->getRemoteConnection().onMessage(R"({
+                                                            "method": "FakeDomain.eventToBeDropped",
+                                                            "params": ["arg1", "arg2"]
+                                                          })");
+
+  // Disconnect from the page.
+  EXPECT_CALL(*localConnections_[0], disconnect()).RetiresOnSaturation();
+  webSockets_[0]->getDelegate().didReceiveMessage(sformat(
+      R"({{
+          "event": "disconnect",
+          "payload": {{
+            "pageId": {0}
+          }}
+        }})",
+      toJson(std::to_string(pageId))));
+  EXPECT_FALSE(localConnections_[0]);
+
+  // Connect to the same page again.
+  webSockets_[0]->getDelegate().didReceiveMessage(sformat(
+      R"({{
+          "event": "connect",
+          "payload": {{
+            "pageId": {0}
+          }}
+        }})",
+      toJson(std::to_string(pageId))));
+
+  EXPECT_TRUE(localConnections_[1]);
+
+  // Send an event from the mocked backend (local) to the frontend (remote) over
+  // the new connection, then flush the callback queue.
+  // Only this event should be sent over the socket.
+  EXPECT_CALL(
+      *webSockets_[0],
+      send(JsonParsed(AllOf(
+          AtJsonPtr("/event", Eq("wrappedEvent")),
+          AtJsonPtr("/payload/pageId", Eq(std::to_string(pageId))),
+          AtJsonPtr(
+              "/payload/wrappedEvent",
+              JsonEq(
+                  R"({
+                    "method": "FakeDomain.eventToBeDelivered",
+                    "params": ["arg1", "arg2"]
+                  })"))))))
+      .RetiresOnSaturation();
+  localConnections_[1]->getRemoteConnection().onMessage(R"({
+                                                            "method": "FakeDomain.eventToBeDelivered",
+                                                            "params": ["arg1", "arg2"]
+                                                          })");
+  EXPECT_EQ(asyncExecutor_.run(), 2);
+
+  // Clean up.
+  EXPECT_CALL(*localConnections_[1], disconnect()).RetiresOnSaturation();
+  getInspectorInstance().removePage(pageId);
+}
+
+TEST_F(
+    InspectorPackagerConnectionTestAsync,
+    TestAttemptSendToStaleRemoteConnectionWhenRetained) {
+  // Configure gmock to expect calls in a specific order.
+  InSequence mockCallsMustBeInSequence;
+
+  packagerConnection_->connect();
+  auto pageId = getInspectorInstance().addPage(
+      "mock-title",
+      "mock-vm",
+      localConnections_
+          .lazily_make_unique<std::unique_ptr<IRemoteConnection>>());
+
+  // Connect to the page.
+  webSockets_[0]->getDelegate().didReceiveMessage(sformat(
+      R"({{
+          "event": "connect",
+          "payload": {{
+            "pageId": {0}
+          }}
+        }})",
+      toJson(std::to_string(pageId))));
+  ASSERT_TRUE(localConnections_[0]);
+
+  // Send an event from the mocked backend (local) to the frontend (remote)
+  // but don't flush the callback queue yet.
+  localConnections_[0]->getRemoteConnection().onMessage(R"({
+                                                            "method": "FakeDomain.eventToBeDropped",
+                                                            "params": ["arg1", "arg2"]
+                                                          })");
+
+  // Forcibly retain the remote connection beyond localConnections_[0]'s
+  // lifetime.
+  auto retainedRemoteConnection0 =
+      localConnections_[0]->dangerouslyReleaseRemoteConnection();
+
+  // Disconnect from the page.
+  EXPECT_CALL(*localConnections_[0], disconnect()).RetiresOnSaturation();
+  webSockets_[0]->getDelegate().didReceiveMessage(sformat(
+      R"({{
+          "event": "disconnect",
+          "payload": {{
+            "pageId": {0}
+          }}
+        }})",
+      toJson(std::to_string(pageId))));
+  EXPECT_FALSE(localConnections_[0]);
+
+  // Connect to the same page again.
+  webSockets_[0]->getDelegate().didReceiveMessage(sformat(
+      R"({{
+          "event": "connect",
+          "payload": {{
+            "pageId": {0}
+          }}
+        }})",
+      toJson(std::to_string(pageId))));
+
+  EXPECT_TRUE(localConnections_[1]);
+
+  // Remember localConnections_[0]'s remote connection? We can still use it
+  // without crashing, but it will not deliver any messages.
+  retainedRemoteConnection0->onMessage(R"({
+                                           "method": "FakeDomain.anotherEventToBeDropped",
+                                           "params": ["arg1", "arg2"]
+                                         })");
+
+  // Send an event from the mocked backend (local) to the frontend (remote) over
+  // the new connection, then flush the callback queue.
+  // Only this event should be sent over the socket.
+  EXPECT_CALL(
+      *webSockets_[0],
+      send(JsonParsed(AllOf(
+          AtJsonPtr("/event", Eq("wrappedEvent")),
+          AtJsonPtr("/payload/pageId", Eq(std::to_string(pageId))),
+          AtJsonPtr(
+              "/payload/wrappedEvent",
+              JsonEq(
+                  R"({
+                    "method": "FakeDomain.eventToBeDelivered",
+                    "params": ["arg1", "arg2"]
+                  })"))))))
+      .RetiresOnSaturation();
+  localConnections_[1]->getRemoteConnection().onMessage(R"({
+                                                            "method": "FakeDomain.eventToBeDelivered",
+                                                            "params": ["arg1", "arg2"]
+                                                          })");
+  EXPECT_EQ(asyncExecutor_.run(), 3);
+
+  // Clean up.
+  EXPECT_CALL(*localConnections_[1], disconnect()).RetiresOnSaturation();
+  getInspectorInstance().removePage(pageId);
+}
 } // namespace facebook::react::jsinspector_modern