mobile: Wait until the Engine is ready before calling terminate (#34287)

When the Engine is not fully initialized, calling terminate() will cause an assertion failure.

[2024-05-21 21:28:16.718][18][critical][assert] [external/envoy/source/common/stats/thread_local_store.cc:49] assert failure: scopes_.empty().

This PR adds a workaround to wait until the Engine before proceeding to call terminate().

Risk Level: low
Testing: unit tests
Docs Changes: n/a
Release Notes: n/a
Platform Specific Features: mobile

Signed-off-by: Fredy Wijaya <fredyw@google.com>

mobile/library/common/internal_engine.cc

@@ -11,6 +11,9 @@
 #include "library/common/stats/utility.h"
 
 namespace Envoy {
+namespace {
+constexpr absl::Duration ENGINE_RUNNING_TIMEOUT = absl::Seconds(3);
+} // namespace
 
 static std::atomic<envoy_stream_t> current_stream_handle_{0};
 
@@ -173,6 +176,7 @@ envoy_status_t InternalEngine::main(std::shared_ptr<Envoy::OptionsImplBase> opti
                                                         server_->serverFactoryContext().scope(),
                                                         server_->api().randomGenerator());
           dispatcher_->drain(server_->dispatcher());
+          engine_running_.Notify();
           callbacks_->on_engine_running_();
         });
   } // mutex_
@@ -212,6 +216,10 @@ envoy_status_t InternalEngine::terminate() {
     return ENVOY_FAILURE;
   }
 
+  // Wait until the Engine is ready before calling terminate to avoid assertion failures.
+  // TODO(fredyw): Fix this without having to wait.
+  ASSERT(engine_running_.WaitForNotificationWithTimeout(ENGINE_RUNNING_TIMEOUT));
+
   // We need to be sure that MainCommon is finished being constructed so we can dispatch shutdown.
   {
     Thread::LockGuard lock(mutex_);

mobile/library/common/internal_engine.h

@@ -7,6 +7,7 @@
 #include "source/common/common/posix/thread_impl.h"
 #include "source/common/common/thread.h"
 
+#include "absl/synchronization/notification.h"
 #include "absl/types/optional.h"
 #include "extension_registry.h"
 #include "library/common/engine_common.h"
@@ -163,6 +164,7 @@ class InternalEngine : public Logger::Loggable<Logger::Id::main> {
   // instructions scheduled on the main_thread_ need to have a longer lifetime.
   Thread::PosixThreadPtr main_thread_{nullptr}; // Empty placeholder to be populated later.
   bool terminated_{false};
+  absl::Notification engine_running_;
 };
 
 } // namespace Envoy

mobile/test/cc/engine_test.cc

@@ -141,4 +141,56 @@ TEST(EngineTest, SetEventTracker) {
   EXPECT_TRUE(on_track.WaitForNotificationWithTimeout(absl::Seconds(3)));
 }
 
+TEST(EngineTest, DontWaitForOnEngineRunning) {
+  Platform::EngineBuilder engine_builder;
+  engine_builder.setLogLevel(Logger::Logger::debug).enforceTrustChainVerification(false);
+  EngineWithTestServer engine_with_test_server(engine_builder, TestServerType::HTTP2_WITH_TLS);
+
+  std::string actual_status_code;
+  bool actual_end_stream = false;
+  absl::Notification stream_complete;
+  EnvoyStreamCallbacks stream_callbacks;
+  stream_callbacks.on_headers_ = [&](const Http::ResponseHeaderMap& headers, bool end_stream,
+                                     envoy_stream_intel) {
+    actual_status_code = headers.getStatusValue();
+    actual_end_stream = end_stream;
+  };
+  stream_callbacks.on_data_ = [&](const Buffer::Instance&, uint64_t /* length */, bool end_stream,
+                                  envoy_stream_intel) { actual_end_stream = end_stream; };
+  stream_callbacks.on_complete_ = [&](envoy_stream_intel, envoy_final_stream_intel) {
+    stream_complete.Notify();
+  };
+  stream_callbacks.on_error_ = [&](EnvoyError, envoy_stream_intel, envoy_final_stream_intel) {
+    stream_complete.Notify();
+  };
+  stream_callbacks.on_cancel_ = [&](envoy_stream_intel, envoy_final_stream_intel) {
+    stream_complete.Notify();
+  };
+  auto stream_prototype = engine_with_test_server.engine()->streamClient()->newStreamPrototype();
+  Platform::StreamSharedPtr stream = stream_prototype->start(std::move(stream_callbacks));
+
+  auto headers = Http::Utility::createRequestHeaderMapPtr();
+  headers->addCopy(Http::LowerCaseString(":method"), "GET");
+  headers->addCopy(Http::LowerCaseString(":scheme"), "https");
+  headers->addCopy(Http::LowerCaseString(":authority"),
+                   engine_with_test_server.testServer().getAddress());
+  headers->addCopy(Http::LowerCaseString(":path"), "/");
+  stream->sendHeaders(std::move(headers), true);
+  stream_complete.WaitForNotification();
+
+  EXPECT_EQ(actual_status_code, "200");
+  EXPECT_TRUE(actual_end_stream);
+}
+
+TEST(EngineTest, TerminateWithoutWaitingForOnEngineRunning) {
+  absl::Notification engine_running;
+  auto engine_callbacks = std::make_unique<EngineCallbacks>();
+  engine_callbacks->on_engine_running_ = [&] { engine_running.Notify(); };
+
+  Platform::EngineBuilder engine_builder;
+  auto engine = engine_builder.setLogLevel(Logger::Logger::debug).build();
+
+  engine->terminate();
+}
+
 } // namespace Envoy