ipcz: Implement basic node connection

Split out from the ipcz uber-CL: https://crrev.com/c/3570271

Introduces NodeConnector and some associated ipcz messages which will be
used by ConnectNode() to bootstrap new NodeLinks between nodes.

A NodeConnector is a short-lived object which takes initial ownership
of a DriverTransport to transmit and listen for appropriate handshake
messages. Once handshake is complete, the transport is passed on to
a newly created NodeLink and given to whichever Node initiated a
connection on the transport.

This does not actually implement the ConnectNode() API yet. This
CL is primarily about landing the most important NodeConnector
implementations (broker-to-non-broker, and non-broker-to-broker)
and getting test coverage for them.

Bug: 1299283
Change-Id: I389ac8796c36446661fe1d96f63254fc96a95cad
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/3621786
Commit-Queue: Ken Rockot <rockot@google.com>
Reviewed-by: Daniel Cheng <dcheng@chromium.org>
Cr-Commit-Position: refs/heads/main@{#1001227}

Author: krockot

third_party/ipcz/src/BUILD.gn

@@ -169,7 +169,9 @@ ipcz_source_set("impl") {
     "ipcz/link_type.h",
     "ipcz/message_internal.h",
     "ipcz/node.h",
+    "ipcz/node_connector.h",
     "ipcz/node_link.h",
+    "ipcz/node_messages.h",
     "ipcz/parcel.h",
     "ipcz/parcel_queue.h",
     "ipcz/portal.h",
@@ -198,9 +200,9 @@ ipcz_source_set("impl") {
     "ipcz/message_macros/message_params_declaration_macros.h",
     "ipcz/message_macros/undef_message_macros.h",
     "ipcz/node.cc",
+    "ipcz/node_connector.cc",
     "ipcz/node_link.cc",
     "ipcz/node_messages.cc",
-    "ipcz/node_messages.h",
     "ipcz/node_messages_generator.h",
     "ipcz/node_name.cc",
     "ipcz/node_name.h",
@@ -253,6 +255,7 @@ ipcz_source_set("ipcz_tests_sources") {
     "ipcz/driver_object_test.cc",
     "ipcz/driver_transport_test.cc",
     "ipcz/message_internal_test.cc",
+    "ipcz/node_connector_test.cc",
     "ipcz/node_link_test.cc",
     "ipcz/parcel_queue_test.cc",
     "ipcz/sequenced_queue_test.cc",
@@ -261,6 +264,8 @@ ipcz_source_set("ipcz_tests_sources") {
     "test/mock_driver.h",
     "test/test_base.cc",
     "test/test_base.h",
+    "test/test_transport_listener.cc",
+    "test/test_transport_listener.h",
     "trap_test.cc",
     "util/ref_counted_test.cc",
     "util/stack_trace_test.cc",

third_party/ipcz/src/ipcz/driver_transport_test.cc

@@ -15,6 +15,7 @@
 #include "ipcz/driver_object.h"
 #include "ipcz/node.h"
 #include "test/mock_driver.h"
+#include "test/test_transport_listener.h"
 #include "testing/gtest/include/gtest/gtest.h"
 #include "third_party/abseil-cpp/absl/base/macros.h"
 #include "util/ref_counted.h"
@@ -30,11 +31,6 @@ DriverTransport::Message MakeMessage(std::string_view s) {
       absl::MakeSpan(reinterpret_cast<const uint8_t*>(s.data()), s.size()));
 }
 
-std::string_view MessageAsString(const DriverTransport::Message& message) {
-  return std::string_view(reinterpret_cast<const char*>(message.data.data()),
-                          message.data.size());
-}
-
 class DriverTransportTest : public testing::Test {
  public:
   DriverTransportTest() = default;
@@ -56,34 +52,6 @@ class DriverTransportTest : public testing::Test {
                                        IPCZ_INVALID_DRIVER_HANDLE)};
 };
 
-class TestListener : public DriverTransport::Listener {
- public:
-  using MessageHandler =
-      std::function<IpczResult(const DriverTransport::Message&)>;
-  using ErrorHandler = std::function<void()>;
-
-  explicit TestListener(MessageHandler message_handler,
-                        ErrorHandler error_handler = nullptr)
-      : message_handler_(std::move(message_handler)),
-        error_handler_(std::move(error_handler)) {}
-  ~TestListener() override = default;
-
-  IpczResult OnTransportMessage(
-      const DriverTransport::Message& message) override {
-    return message_handler_(message);
-  }
-
-  void OnTransportError() override {
-    if (error_handler_) {
-      error_handler_();
-    }
-  }
-
- private:
-  MessageHandler message_handler_;
-  ErrorHandler error_handler_;
-};
-
 TEST_F(DriverTransportTest, Activation) {
   constexpr IpczDriverHandle kTransport0 = 5;
   constexpr IpczDriverHandle kTransport1 = 42;
@@ -101,24 +69,23 @@ TEST_F(DriverTransportTest, Activation) {
       })
       .RetiresOnSaturation();
 
-  // Verify that activation of a DriverTransport feeds the driver an activity
-  // handler and valid ipcz handle to use when notifying ipcz of incoming
-  // communications.
-  b->Activate();
-  EXPECT_NE(IPCZ_INVALID_HANDLE, ipcz_transport);
-  EXPECT_TRUE(activity_handler);
-
   // And verify that the activity handler actually invokes the transport's
   // Listener.
 
   const std::string kTestMessage = "hihihihi";
   bool received = false;
-  TestListener listener([&](const DriverTransport::Message& message) {
-    EXPECT_EQ(kTestMessage, MessageAsString(message));
+  test::TestTransportListener listener(b);
+  listener.OnStringMessage([&](std::string_view message) {
+    EXPECT_EQ(kTestMessage, message);
     received = true;
     return IPCZ_RESULT_OK;
   });
-  b->set_listener(&listener);
+
+  // Verify that activation of a DriverTransport feeds the driver an activity
+  // handler and valid ipcz handle to use when notifying ipcz of incoming
+  // communications.
+  EXPECT_NE(IPCZ_INVALID_HANDLE, ipcz_transport);
+  EXPECT_TRUE(activity_handler);
 
   EXPECT_FALSE(received);
   EXPECT_EQ(
@@ -135,7 +102,6 @@ TEST_F(DriverTransportTest, Activation) {
 
   EXPECT_CALL(driver(), Close(kTransport1, _, _));
   EXPECT_CALL(driver(), Close(kTransport0, _, _));
-  b->Deactivate();
 
   // The driver must also release its handle to ipcz' DriverTransport, which it
   // does by an invocation of the activity handler like this. Without this, we'd
@@ -162,17 +128,9 @@ TEST_F(DriverTransportTest, Error) {
       })
       .RetiresOnSaturation();
 
-  b->Activate();
-
   bool observed_error = false;
-  TestListener listener(
-      [&](const DriverTransport::Message& message) {
-        ABSL_ASSERT(false);
-        return IPCZ_RESULT_INVALID_ARGUMENT;
-      },
-      [&] { observed_error = true; });
-
-  b->set_listener(&listener);
+  test::TestTransportListener listener(b);
+  listener.OnError([&] { observed_error = true; });
 
   // Verify that a driver invoking the activity handler with
   // IPCZ_TRANSPORT_ACTIVITY_ERROR results in an error notification on the
@@ -191,6 +149,7 @@ TEST_F(DriverTransportTest, Error) {
             activity_handler(ipcz_transport, nullptr, 0, nullptr, 0,
                              IPCZ_TRANSPORT_ACTIVITY_DEACTIVATED, nullptr));
 
+  EXPECT_CALL(driver(), DeactivateTransport(kTransport1, _, _));
   EXPECT_CALL(driver(), Close(kTransport1, _, _));
   EXPECT_CALL(driver(), Close(kTransport0, _, _));
 }

third_party/ipcz/src/ipcz/node.cc

@@ -4,9 +4,17 @@
 
 #include "ipcz/node.h"
 
+#include <vector>
+
 #include "ipcz/ipcz.h"
+#include "ipcz/node_connector.h"
+#include "ipcz/node_link.h"
+#include "ipcz/portal.h"
+#include "ipcz/router.h"
+#include "third_party/abseil-cpp/absl/base/macros.h"
 #include "third_party/abseil-cpp/absl/synchronization/mutex.h"
 #include "util/log.h"
+#include "util/ref_counted.h"
 
 namespace ipcz {
 
@@ -24,6 +32,16 @@ Node::Node(Type type, const IpczDriver& driver, IpczDriverHandle driver_node)
 Node::~Node() = default;
 
 IpczResult Node::Close() {
+  absl::flat_hash_map<NodeName, Ref<NodeLink>> node_links;
+  {
+    absl::MutexLock lock(&mutex_);
+    node_links_.swap(node_links);
+    broker_link_.reset();
+  }
+
+  for (const auto& entry : node_links) {
+    entry.second->Deactivate();
+  }
   return IPCZ_RESULT_OK;
 }
 
@@ -32,6 +50,29 @@ NodeName Node::GetAssignedName() {
   return assigned_name_;
 }
 
+Ref<NodeLink> Node::GetBrokerLink() {
+  absl::MutexLock lock(&mutex_);
+  return broker_link_;
+}
+
+void Node::SetBrokerLink(Ref<NodeLink> link) {
+  absl::MutexLock lock(&mutex_);
+  ABSL_ASSERT(!broker_link_);
+  broker_link_ = std::move(link);
+}
+
+void Node::SetAssignedName(const NodeName& name) {
+  absl::MutexLock lock(&mutex_);
+  ABSL_ASSERT(!assigned_name_.is_valid());
+  assigned_name_ = name;
+}
+
+bool Node::AddLink(const NodeName& remote_node_name, Ref<NodeLink> link) {
+  absl::MutexLock lock(&mutex_);
+  auto [it, inserted] = node_links_.insert({remote_node_name, std::move(link)});
+  return inserted;
+}
+
 NodeName Node::GenerateRandomName() const {
   NodeName name;
   IpczResult result =

third_party/ipcz/src/ipcz/node.h

@@ -8,10 +8,14 @@
 #include "ipcz/api_object.h"
 #include "ipcz/ipcz.h"
 #include "ipcz/node_name.h"
+#include "third_party/abseil-cpp/absl/container/flat_hash_map.h"
 #include "third_party/abseil-cpp/absl/synchronization/mutex.h"
+#include "third_party/abseil-cpp/absl/types/span.h"
 
 namespace ipcz {
 
+class NodeLink;
+
 // A Node controls creation and interconnection of a collection of routers which
 // can establish links to and from other routers in other nodes. Every node is
 // assigned a globally unique name by a trusted broker node, and nodes may be
@@ -43,6 +47,30 @@ class Node : public APIObjectImpl<Node, APIObject::kNode> {
   // Retrieves the name assigned to this node, if any.
   NodeName GetAssignedName();
 
+  // Gets a reference to the node's broker link, if it has one.
+  Ref<NodeLink> GetBrokerLink();
+
+  // Sets this node's broker link, which is used e.g. to make introduction
+  // requests.
+  //
+  // This is called by a NodeConnector implementation after accepting a valid
+  // handshake message from a broker node, and `link` will be used as this
+  // node's permanent broker.
+  //
+  // Note that like any other NodeLink used by this Node, the same `link` must
+  // also be registered via AddLink() to associate it with its remote Node's
+  // name. This is also done by NodeConnector.
+  void SetBrokerLink(Ref<NodeLink> link);
+
+  // Sets this node's assigned name as given by a broker. NodeConnector is
+  // responsible for calling on non-broker Nodes this after receiving the
+  // expected handshake from a broker. Must not be called on broker nodes, as
+  // they assign their own name at construction time.
+  void SetAssignedName(const NodeName& name);
+
+  // Registers a new NodeLink for the given `remote_node_name`.
+  bool AddLink(const NodeName& remote_node_name, Ref<NodeLink> link);
+
   // Generates a new random NodeName using this node's driver as a source of
   // randomness.
   NodeName GenerateRandomName() const;
@@ -60,6 +88,18 @@ class Node : public APIObjectImpl<Node, APIObject::kNode> {
   // self-assigned if this is a broker node. Once assigned, this name remains
   // constant through the lifetime of the node.
   NodeName assigned_name_ ABSL_GUARDED_BY(mutex_);
+
+  // A link to the first broker this node connected to. If this link is broken,
+  // the node will lose all its other links too.
+  Ref<NodeLink> broker_link_ ABSL_GUARDED_BY(mutex_);
+
+  // Lookup table of broker-assigned node names and links to those nodes. All of
+  // these links and their associated names are received by the `broker_link_`
+  // if this is a non-broker node. If this is a broker node, these links are
+  // either assigned by this node itself, or received from other brokers in the
+  // system.
+  absl::flat_hash_map<NodeName, Ref<NodeLink>> node_links_
+      ABSL_GUARDED_BY(mutex_);
 };
 
 }  // namespace ipcz