Mojo C++ bindings: error notification behavior related to sync calls.

This CL changes the behavior to:
- connection error handler doesn't reenter ongoing sync calls.
- connection error handler is delayed until all queued asynchronous
  messages are processed.

BUG=577699

Review URL: https://codereview.chromium.org/1781573004

Cr-Commit-Position: refs/heads/master@{#380259}

4 files changed, 152 insertions, 10 deletions

diff --git a/mojo/public/cpp/bindings/lib/connector.cc b/mojo/public/cpp/bindings/lib/connector.cc
index 812b4c11..c7378f0 100644
--- a/mojo/public/cpp/bindings/lib/connector.cc
+++ b/mojo/public/cpp/bindings/lib/connector.cc
@@ -368,16 +368,18 @@ void Connector::HandleError(bool force_pipe_reset, bool force_async_handler) {
   if (error_ || !message_pipe_.is_valid())
     return;
 
-  if (!force_pipe_reset && force_async_handler)
-    force_pipe_reset = true;
-
-  if (paused_) {
-    // If the user has paused receiving messages, we shouldn't call the error
-    // handler right away. We need to wait until the user starts receiving
-    // messages again.
+  if (during_sync_handle_watcher_callback() || paused_) {
+    // Enforce calling the error handler asynchronously if:
+    // - currently we are in a sync handle watcher callback. We don't want the
+    // error handler to reenter an ongoing sync call.
+    // - the user has paused receiving messages. We need to wait until the user
+    // starts receiving messages again.
     force_async_handler = true;
   }
 
+  if (!force_pipe_reset && force_async_handler)
+    force_pipe_reset = true;
+
   if (force_pipe_reset) {
     CancelWait();
     MayAutoLock locker(lock_.get());
diff --git a/mojo/public/cpp/bindings/lib/router.cc b/mojo/public/cpp/bindings/lib/router.cc
index ddfc0cfa8..d929e80 100644
--- a/mojo/public/cpp/bindings/lib/router.cc
+++ b/mojo/public/cpp/bindings/lib/router.cc
@@ -93,11 +93,13 @@ Router::Router(ScopedMessagePipeHandle message_pipe,
       next_request_id_(0),
       testing_mode_(false),
       pending_task_for_messages_(false),
+      encountered_error_(false),
       weak_factory_(this) {
   filters_.SetSink(&thunk_);
   if (expects_sync_requests)
     connector_.RegisterSyncHandleWatch();
   connector_.set_incoming_receiver(filters_.GetHead());
+  connector_.set_connection_error_handler([this]() { OnConnectionError(); });
 }
 
 Router::~Router() {}
@@ -205,6 +207,12 @@ void Router::HandleQueuedMessages() {
   }
 
   pending_task_for_messages_ = false;
+
+  // We may have already seen a connection error from the connector, but
+  // haven't notified the user because we want to process all the queued
+  // messages first. We should do it now.
+  if (connector_.encountered_error() && !encountered_error_)
+    OnConnectionError();
 }
 
 bool Router::HandleMessageInternal(Message* message) {
@@ -250,6 +258,21 @@ bool Router::HandleMessageInternal(Message* message) {
   }
 }
 
+void Router::OnConnectionError() {
+  DCHECK(!encountered_error_);
+
+  if (!pending_messages_.empty()) {
+    // After all the pending messages are processed, we will check whether an
+    // error has been encountered and run the user's connection error handler
+    // if necessary.
+    DCHECK(pending_task_for_messages_);
+    return;
+  }
+
+  encountered_error_ = true;
+  error_handler_.Run();
+}
+
 // ----------------------------------------------------------------------------
 
 }  // namespace internal
diff --git a/mojo/public/cpp/bindings/lib/router.h b/mojo/public/cpp/bindings/lib/router.h
index 2fec5d2..6374f18 100644
--- a/mojo/public/cpp/bindings/lib/router.h
+++ b/mojo/public/cpp/bindings/lib/router.h
@@ -37,14 +37,14 @@ class Router : public MessageReceiverWithResponder {
   // Sets the error handler to receive notifications when an error is
   // encountered while reading from the pipe or waiting to read from the pipe.
   void set_connection_error_handler(const Closure& error_handler) {
-    connector_.set_connection_error_handler(error_handler);
+    error_handler_ = error_handler;
   }
 
   // Returns true if an error was encountered while reading from the pipe or
   // waiting to read from the pipe.
   bool encountered_error() const {
     DCHECK(thread_checker_.CalledOnValidThread());
-    return connector_.encountered_error();
+    return encountered_error_;
   }
 
   // Is the router bound to a MessagePipe handle?
@@ -143,9 +143,10 @@ class Router : public MessageReceiverWithResponder {
 
   bool HandleIncomingMessage(Message* message);
   void HandleQueuedMessages();
-
   bool HandleMessageInternal(Message* message);
 
+  void OnConnectionError();
+
   HandleIncomingMessageThunk thunk_;
   FilterChain filters_;
   Connector connector_;
@@ -158,6 +159,8 @@ class Router : public MessageReceiverWithResponder {
   // Whether a task has been posted to trigger processing of
   // |pending_messages_|.
   bool pending_task_for_messages_;
+  bool encountered_error_;
+  Closure error_handler_;
   base::ThreadChecker thread_checker_;
   base::WeakPtrFactory<Router> weak_factory_;
 };
diff --git a/mojo/public/cpp/bindings/tests/sync_method_unittest.cc b/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
index 207a6e2..f1a4767 100644
--- a/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
+++ b/mojo/public/cpp/bindings/tests/sync_method_unittest.cc
@@ -345,6 +345,120 @@ TEST_F(SyncMethodTest, AsyncRequestQueuedDuringSyncCall) {
   EXPECT_TRUE(async_echo_response_dispatched);
 }
 
+TEST_F(SyncMethodTest, QueuedMessagesProcessedBeforeErrorNotification) {
+  // Test that while an interface pointer is waiting for the response to a sync
+  // call, async responses are queued. If the message pipe is disconnected
+  // before the queued messages are processed, the connection error
+  // notification is delayed until all the queued messages are processed.
+
+  TestSyncPtr ptr;
+  TestSyncImpl impl(GetProxy(&ptr));
+
+  int32_t async_echo_request_value = -1;
+  TestSync::AsyncEchoCallback async_echo_request_callback;
+  base::RunLoop run_loop1;
+  impl.set_async_echo_handler(
+      [&async_echo_request_value, &async_echo_request_callback, &run_loop1](
+          int32_t value, const TestSync::AsyncEchoCallback& callback) {
+        async_echo_request_value = value;
+        async_echo_request_callback = callback;
+        run_loop1.Quit();
+      });
+
+  bool async_echo_response_dispatched = false;
+  base::RunLoop run_loop2;
+  ptr->AsyncEcho(123,
+                 [&async_echo_response_dispatched, &run_loop2](int32_t result) {
+                   async_echo_response_dispatched = true;
+                   EXPECT_EQ(123, result);
+                   run_loop2.Quit();
+                 });
+  // Run until the AsyncEcho request reaches the service side.
+  run_loop1.Run();
+
+  impl.set_echo_handler(
+      [&impl, &async_echo_request_value, &async_echo_request_callback](
+          int32_t value, const TestSync::EchoCallback& callback) {
+        // Send back the async response first.
+        EXPECT_FALSE(async_echo_request_callback.is_null());
+        async_echo_request_callback.Run(async_echo_request_value);
+
+        impl.binding()->Close();
+      });
+
+  bool connection_error_dispatched = false;
+  base::RunLoop run_loop3;
+  ptr.set_connection_error_handler(
+      [&connection_error_dispatched, &run_loop3]() {
+        connection_error_dispatched = true;
+        run_loop3.Quit();
+      });
+
+  int32_t result_value = -1;
+  ASSERT_FALSE(ptr->Echo(456, &result_value));
+  EXPECT_EQ(-1, result_value);
+  ASSERT_FALSE(connection_error_dispatched);
+  EXPECT_FALSE(ptr.encountered_error());
+
+  // Although the AsyncEcho response arrives before the Echo response, it should
+  // be queued and not yet dispatched.
+  EXPECT_FALSE(async_echo_response_dispatched);
+
+  // Run until the AsyncEcho response is dispatched.
+  run_loop2.Run();
+
+  EXPECT_TRUE(async_echo_response_dispatched);
+  ASSERT_FALSE(connection_error_dispatched);
+  EXPECT_FALSE(ptr.encountered_error());
+
+  // Run until the error notification is dispatched.
+  run_loop3.Run();
+
+  ASSERT_TRUE(connection_error_dispatched);
+  EXPECT_TRUE(ptr.encountered_error());
+}
+
+TEST_F(SyncMethodTest, InvalidMessageDuringSyncCall) {
+  // Test that while an interface pointer is waiting for the response to a sync
+  // call, an invalid incoming message will disconnect the message pipe, cause
+  // the sync call to return false, and run the connection error handler
+  // asynchronously.
+
+  MessagePipe pipe;
+
+  TestSyncPtr ptr;
+  ptr.Bind(TestSyncPtrInfo(std::move(pipe.handle0), 0u));
+
+  MessagePipeHandle raw_binding_handle = pipe.handle1.get();
+  TestSyncImpl impl(MakeRequest<TestSync>(std::move(pipe.handle1)));
+
+  impl.set_echo_handler([&raw_binding_handle](
+      int32_t value, const TestSync::EchoCallback& callback) {
+    // Write a 1-byte message, which is considered invalid.
+    char invalid_message = 0;
+    MojoResult result =
+        WriteMessageRaw(raw_binding_handle, &invalid_message, 1u, nullptr, 0u,
+                        MOJO_WRITE_MESSAGE_FLAG_NONE);
+    ASSERT_EQ(MOJO_RESULT_OK, result);
+    callback.Run(value);
+  });
+
+  bool connection_error_dispatched = false;
+  base::RunLoop run_loop;
+  ptr.set_connection_error_handler([&connection_error_dispatched, &run_loop]() {
+    connection_error_dispatched = true;
+    run_loop.Quit();
+  });
+
+  int32_t result_value = -1;
+  ASSERT_FALSE(ptr->Echo(456, &result_value));
+  EXPECT_EQ(-1, result_value);
+  ASSERT_FALSE(connection_error_dispatched);
+
+  run_loop.Run();
+  ASSERT_TRUE(connection_error_dispatched);
+}
+
 }  // namespace
 }  // namespace test
 }  // namespace mojo