[mojo] Fix remote process disconnection raciness

Currently if a remote process is detected as disconnected (i.e. your
internal Channel transport to that process signals an error), unread
messages which have already arrived from that process may be unreadable
from the ports that received them.

Bug: 1005510
Change-Id: Id743ce4560a0309a2058d0b825a2bb4a6065b342
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1832244
Commit-Queue: Ken Rockot <rockot@google.com>
Reviewed-by: Darin Fisher <darin@chromium.org>
Cr-Commit-Position: refs/heads/master@{#701907}

mojo/core/multiprocess_message_pipe_unittest.cc

@@ -1311,6 +1311,38 @@ DEFINE_TEST_CLIENT_TEST_WITH_PIPE(MessagePipeStatusChangeInTransitClient,
     CloseHandle(handles[i]);
 }
 
+TEST_P(MultiprocessMessagePipeTestWithPeerSupport,
+       ReceiveMessagesSentJustBeforeProcessDeath) {
+  // Regression test for https://crbug.com/1005510. The client will write a
+  // message to the pipe it gives us and then it will die immediately. We should
+  // always be able to read the message received on that pipe.
+  RunTestClient("SpotaneouslyDyingProcess", [&](MojoHandle child) {
+    MojoHandle receiver;
+    EXPECT_EQ("receiver", ReadMessageWithHandles(child, &receiver, 1));
+    EXPECT_EQ("ok", ReadMessage(receiver));
+  });
+}
+
+DEFINE_TEST_CLIENT_TEST_WITH_PIPE(SpotaneouslyDyingProcess,
+                                  MultiprocessMessagePipeTest,
+                                  parent) {
+  MojoHandle sender;
+  MojoHandle receiver;
+  CreateMessagePipe(&sender, &receiver);
+
+  WriteMessageWithHandles(parent, "receiver", &receiver, 1);
+
+  // Wait for the pipe to actually appear as remote. Before this happens, it's
+  // possible for message transmission to be deferred to the IO thread, and
+  // sudden termination might preempt that work.
+  WaitForSignals(sender, MOJO_HANDLE_SIGNAL_PEER_REMOTE);
+
+  WriteMessage(sender, "ok");
+
+  // Here process termination is imminent. If the bug reappears this test will
+  // fail flakily.
+}
+
 TEST_F(MultiprocessMessagePipeTest, MessagePipeStatusChangeInTransit) {
   MojoHandle local_handles[4];
   MojoHandle sent_handles[4];

mojo/core/ports/node.cc

@@ -87,6 +87,8 @@ bool CanAcceptMoreMessages(const Port* port) {
   if (port->state == Port::kClosed)
     return false;
   if (port->peer_closed || port->remove_proxy_on_last_message) {
+    if (port->peer_lost_unexpectedly)
+      return port->message_queue.HasNextMessage();
     if (port->last_sequence_num_to_receive == next_sequence_num - 1)
       return false;
   }
@@ -1538,9 +1540,7 @@ void Node::DestroyAllPortsWithPeer(const NodeName& node_name,
           // messages.
 
           port->peer_closed = true;
-          port->last_sequence_num_to_receive =
-              port->message_queue.next_sequence_num() - 1;
-
+          port->peer_lost_unexpectedly = true;
           if (port->state == Port::kReceiving)
             ports_to_notify.push_back(local_port_ref);
         }

mojo/core/ports/port.cc

@@ -18,7 +18,8 @@ Port::Port(uint64_t next_sequence_num_to_send,
       sequence_num_to_acknowledge(0),
       message_queue(next_sequence_num_to_receive),
       remove_proxy_on_last_message(false),
-      peer_closed(false) {}
+      peer_closed(false),
+      peer_lost_unexpectedly(false) {}
 
 Port::~Port() {}
 

mojo/core/ports/port.h

@@ -166,6 +166,13 @@ class Port : public base::RefCountedThreadSafe<Port> {
   // non-zero cyclic routing distance) receiving Port has been closed.
   bool peer_closed;
 
+  // Indicates that this Port lost its peer unexpectedly (e.g. via process death
+  // rather than receiving an ObserveClosure event). In this case
+  // |peer_closed| will be true but |last_sequence_num_to_receive| cannot be
+  // known. Such ports will continue to make message available until their
+  // message queue is empty.
+  bool peer_lost_unexpectedly;
+
   Port(uint64_t next_sequence_num_to_send,
        uint64_t next_sequence_num_to_receive);