[mojo-edk] Broadcast surprise port disruptions

mojo/edk/system/ports/ports_unittest.cc

Index: mojo/edk/system/ports/ports_unittest.cc
diff --git a/mojo/edk/system/ports/ports_unittest.cc b/mojo/edk/system/ports/ports_unittest.cc
index 1098e8e2942784b46d2fa44da1822e3fdd8d7bff..94f453e477b089abb2777186b2409857914d31de 100644
--- a/mojo/edk/system/ports/ports_unittest.cc
+++ b/mojo/edk/system/ports/ports_unittest.cc
@@ -88,8 +88,10 @@ struct TaskComparator {
   }
 };
 
+const size_t kMaxNodes = 3;
+
 std::priority_queue<Task*, std::vector<Task*>, TaskComparator> task_queue;
-Node* node_map[2];
+Node* node_map[kMaxNodes];
 
 Node* GetNode(const NodeName& name) {
   return node_map[name.v1];
@@ -105,7 +107,8 @@ void PumpTasks() {
     task_queue.pop();
 
     Node* node = GetNode(task->node_name);
-    node->AcceptMessage(std::move(task->message));
+    if (node)
+      node->AcceptMessage(std::move(task->message));
 
     delete task;
   }
@@ -194,6 +197,21 @@ class TestNodeDelegate : public NodeDelegate {
     task_queue.push(new Task(node_name, std::move(message)));
   }
 
+  void BroadcastMessage(ScopedMessage message) override {
+    for (size_t i = 0; i < kMaxNodes; ++i) {
+      Node* node = node_map[i];
+      if (node) {
+        // NOTE: We only need to support broadcast of events, which have no
+        // payload or ports bytes.
+        ScopedMessage new_message(
+            new TestMessage(message->num_header_bytes(), 0, 0));
+        memcpy(new_message->mutable_header_bytes(), message->header_bytes(),
+               message->num_header_bytes());
+        node->AcceptMessage(std::move(new_message));
+      }
+    }
+  }
+
   void PortStatusChanged(const PortRef& port) override {
     DVLOG(1) << "PortStatusChanged for " << port.name() << "@" << node_name_;
     if (!read_messages_)
@@ -239,15 +257,11 @@ class TestNodeDelegate : public NodeDelegate {
 
 class PortsTest : public testing::Test {
  public:
-  PortsTest() {
-    SetNode(NodeName(0, 1), nullptr);
-    SetNode(NodeName(1, 1), nullptr);
-  }
-
-  ~PortsTest() override {
+  void SetUp() override {
     DiscardPendingTasks();
     SetNode(NodeName(0, 1), nullptr);
     SetNode(NodeName(1, 1), nullptr);
+    SetNode(NodeName(2, 1), nullptr);
   }
 };
 
@@ -461,6 +475,80 @@ TEST_F(PortsTest, LostConnectionToNode2) {
   EXPECT_TRUE(node1.CanShutdownCleanly(false));
 }
 
+TEST_F(PortsTest, LostConnectionToNodeWithSecondaryProxy) {
+  // Tests that a proxy gets cleaned up when its indirect peer lives on a lost
+  // node.
+
+  NodeName node0_name(0, 1);
+  TestNodeDelegate node0_delegate(node0_name);
+  Node node0(node0_name, &node0_delegate);
+  node_map[0] = &node0;
+
+  NodeName node1_name(1, 1);
+  TestNodeDelegate node1_delegate(node1_name);
+  Node node1(node1_name, &node1_delegate);
+  node_map[1] = &node1;
+
+  NodeName node2_name(2, 1);
+  TestNodeDelegate node2_delegate(node2_name);
+  Node node2(node2_name, &node2_delegate);
+  node_map[2] = &node2;
+
+  node1_delegate.set_save_messages(true);
+
+  // Create A-B spanning nodes 0 and 1 and C-D spanning 1 and 2.
+  PortRef A, B, C, D;
+  EXPECT_EQ(OK, node0.CreateUninitializedPort(&A));
+  EXPECT_EQ(OK, node1.CreateUninitializedPort(&B));
+  EXPECT_EQ(OK, node0.InitializePort(A, node1_name, B.name()));
+  EXPECT_EQ(OK, node1.InitializePort(B, node0_name, A.name()));
+  EXPECT_EQ(OK, node1.CreateUninitializedPort(&C));
+  EXPECT_EQ(OK, node2.CreateUninitializedPort(&D));
+  EXPECT_EQ(OK, node1.InitializePort(C, node2_name, D.name()));
+  EXPECT_EQ(OK, node2.InitializePort(D, node1_name, C.name()));
+
+  // Create E-F and send F over A to node 1.
+  PortRef E, F;
+  EXPECT_EQ(OK, node0.CreatePortPair(&E, &F));
+  EXPECT_EQ(OK, SendStringMessageWithPort(&node0, A, ".", F));
+
+  PumpTasks();
+
+  ScopedMessage message;
+  ASSERT_TRUE(node1_delegate.GetSavedMessage(&message));
+  ASSERT_EQ(1u, message->num_ports());
+
+  EXPECT_EQ(OK, node1.GetPort(message->ports()[0], &F));
+
+  // Send F over C to node 2 and then simulate node 2 loss from node 1. Node 1
+  // will trivially become aware of the loss, and this test verifies that the
+  // port A on node 0 will eventually also become aware of it.
+
+  EXPECT_EQ(OK, SendStringMessageWithPort(&node1, C, ".", F));
+
+  node_map[2] = nullptr;
+  EXPECT_EQ(OK, node1.LostConnectionToNode(node2_name));
+
+  PumpTasks();
+
+  // Port F should be gone.
+  EXPECT_EQ(ERROR_PORT_UNKNOWN, node1.GetPort(F.name(), &F));
+
+  // Port E should have detected peer closure despite the fact that there is
+  // no longer a continuous route from F to E over which the event could travel.
+  PortStatus status;
+  EXPECT_EQ(OK, node0.GetStatus(E, &status));
+  EXPECT_TRUE(status.peer_closed);
+
+  EXPECT_EQ(OK, node0.ClosePort(A));
+  EXPECT_EQ(OK, node1.ClosePort(B));
+  EXPECT_EQ(OK, node1.ClosePort(C));
+  EXPECT_EQ(OK, node0.ClosePort(E));
+
+  EXPECT_TRUE(node0.CanShutdownCleanly(false));
+  EXPECT_TRUE(node1.CanShutdownCleanly(false));
+}
+
 TEST_F(PortsTest, GetMessage1) {
   NodeName node0_name(0, 1);
   TestNodeDelegate node0_delegate(node0_name);