Stop the flush tests leaking the handler

The fake session kept the handler so one test could complete a session
after the waiter parked, but a handler owns its session, so that closed a
reference cycle and every case leaked one handler and one session.

The script now receives the shared_ptr and the one test that needs the
handler holds it itself, so nothing the fakes own points back at them.

Found by LeakSanitizer in the Bazel asan job, which builds with
ENABLE_ASYNC_EXPORT and therefore runs these cases.

Author: thc1006

exporters/elasticsearch/include/opentelemetry/exporters/elasticsearch/es_log_record_exporter.h

@@ -153,7 +153,9 @@ class ElasticsearchLogRecordExporter final : public opentelemetry::sdk::logs::Lo
     std::atomic<std::size_t> finished_session_counter_{0};
     std::condition_variable force_flush_cv;
     std::mutex force_flush_cv_m;
-    std::recursive_mutex force_flush_m;
+    // Timed so that a second ForceFlush() caller is bounded by its own deadline rather than
+    // by however long the first one waits.
+    std::recursive_timed_mutex force_flush_m;
   };
   nostd::shared_ptr<SynchronizationData> synchronization_data_;
 #endif

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -505,8 +505,35 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
                                                 OPENTELEMETRY_MAYBE_UNUSED) noexcept
 {
 #ifdef ENABLE_ASYNC_EXPORT
+  // ASAN will report chrono: runtime error: signed integer overflow: A + B cannot be represented
+  //   in type 'long int' here. So we reset timeout to meet signed long int limit here. Zero is
+  //   what that returns for a timeout there is no point waiting against, which is also how a
+  //   caller asks for no deadline.
+  timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout(
+      timeout, std::chrono::microseconds::zero());
+
+  // One deadline for the whole call, taken before anything can block, so that time spent waiting
+  // for another ForceFlush() to finish counts against the caller's budget too.
+  const bool indefinite = timeout <= std::chrono::microseconds::zero();
+  const std::chrono::steady_clock::time_point deadline =
+      indefinite ? (std::chrono::steady_clock::time_point::max)()
+                 : std::chrono::steady_clock::now() +
+                       std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
+
   // Serialises concurrent ForceFlush() calls.
-  std::lock_guard<std::recursive_mutex> lock_guard{synchronization_data_->force_flush_m};
+  std::unique_lock<std::recursive_timed_mutex> flush_guard{synchronization_data_->force_flush_m,
+                                                           std::defer_lock};
+  if (indefinite)
+  {
+    flush_guard.lock();
+  }
+  else if (!flush_guard.try_lock_until(deadline))
+  {
+    // Ran out of time before this call could even start. Everything outstanding may still have
+    // finished while waiting, so answer from the counters rather than assuming failure.
+    return synchronization_data_->finished_session_counter_.load(std::memory_order_acquire) >=
+           synchronization_data_->session_counter_.load(std::memory_order_acquire);
+  }
 
   // The flush covers the sessions that were already running when it was asked for.
   const std::size_t running_counter =
@@ -516,29 +543,16 @@ bool ElasticsearchLogRecordExporter::ForceFlush(std::chrono::microseconds timeou
            running_counter;
   };
 
-  // ASAN will report chrono: runtime error: signed integer overflow: A + B cannot be represented
-  //   in type 'long int' here. So we reset timeout to meet signed long int limit here. Zero is
-  //   what that returns for a timeout there is no point waiting against, which is also how a
-  //   caller asks for no deadline.
-  timeout = opentelemetry::common::DurationUtil::AdjustWaitForTimeout(
-      timeout, std::chrono::microseconds::zero());
-
   std::unique_lock<std::mutex> lock(synchronization_data_->force_flush_cv_m);
 
-  if (timeout <= std::chrono::microseconds::zero())
+  if (indefinite)
   {
     // wait() only returns once the predicate holds, so the flush has completed.
     synchronization_data_->force_flush_cv.wait(lock, flushed);
     return true;
   }
 
-  // One deadline for the call, so a wakeup that is not a completion resumes against what is left
-  // rather than starting the wait again. wait_until() returns the predicate, so a flush that ran
-  // out of time cannot report success.
-  const std::chrono::steady_clock::time_point deadline =
-      std::chrono::steady_clock::now() +
-      std::chrono::duration_cast<std::chrono::steady_clock::duration>(timeout);
-
+  // wait_until() returns the predicate, so a flush that ran out of time cannot report success.
   return synchronization_data_->force_flush_cv.wait_until(lock, deadline, flushed);
 #else
   return true;

exporters/elasticsearch/test/es_log_record_exporter_test.cc

@@ -25,6 +25,7 @@
 #include <string>
 #include <thread>
 #include <utility>
+#include <vector>
 #include "nlohmann/json.hpp"
 
 namespace sdklogs       = opentelemetry::sdk::logs;
@@ -195,7 +196,7 @@ class FakeRequest : public http_client::Request
   void SetRetryPolicy(const http_client::RetryPolicy &) noexcept override {}
 };
 
-using EventScript = std::function<void(http_client::EventHandler &)>;
+using EventScript = std::function<void(const std::shared_ptr<http_client::EventHandler> &)>;
 
 class FakeSession : public http_client::Session
 {
@@ -207,12 +208,8 @@ class FakeSession : public http_client::Session
   }
   void SendRequest(std::shared_ptr<http_client::EventHandler> handler) noexcept override
   {
-    last_handler_ = handler;
-    script_(*handler);
+    script_(handler);
   }
-
-  // Kept so a test can deliver a completion after the flush has started waiting.
-  std::shared_ptr<http_client::EventHandler> last_handler_;
   bool IsSessionActive() noexcept override { return false; }
   bool CancelSession() noexcept override { return true; }
   bool FinishSession() noexcept override { return true; }
@@ -227,11 +224,8 @@ class FakeHttpClient : public http_client::HttpClient
   explicit FakeHttpClient(EventScript script) : script_(std::move(script)) {}
   std::shared_ptr<http_client::Session> CreateSession(nostd::string_view) noexcept override
   {
-    last_session_ = std::make_shared<FakeSession>(script_);
-    return last_session_;
+    return std::make_shared<FakeSession>(script_);
   }
-
-  std::shared_ptr<FakeSession> last_session_;
   bool CancelAllSessions() noexcept override { return true; }
   bool FinishAllSessions() noexcept override { return true; }
   void SetMaxSessionsPerConnection(std::size_t) noexcept override {}
@@ -289,7 +283,7 @@ void ExportOnce(logs_exporter::ElasticsearchLogRecordExporter &exporter)
 // so when the caller's deadline runs out rather than when the response timeout does.
 TEST(ElasticsearchForceFlushTests, ReportsFailureWhenTheFlushDoesNotComplete)
 {
-  auto fixture = MakeExporter([](http_client::EventHandler &) {});
+  auto fixture = MakeExporter([](const std::shared_ptr<http_client::EventHandler> &) {});
   ExportOnce(*fixture.exporter);
 
   const auto start   = std::chrono::steady_clock::now();
@@ -304,7 +298,7 @@ TEST(ElasticsearchForceFlushTests, ReportsFailureWhenTheFlushDoesNotComplete)
 // Nothing outstanding, so there is nothing to wait for.
 TEST(ElasticsearchForceFlushTests, ReturnsImmediatelyWithNothingInFlight)
 {
-  auto fixture = MakeExporter([](http_client::EventHandler &) {});
+  auto fixture = MakeExporter([](const std::shared_ptr<http_client::EventHandler> &) {});
 
   const auto start = std::chrono::steady_clock::now();
   EXPECT_TRUE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20}));
@@ -318,9 +312,9 @@ TEST(ElasticsearchForceFlushTests, ReturnsImmediatelyWithNothingInFlight)
 // completion is already published when the predicate is first evaluated.
 TEST(ElasticsearchForceFlushTests, SucceedsWhenTheSessionFinishedBeforeTheWait)
 {
-  auto fixture = MakeExporter([](http_client::EventHandler &handler) {
+  auto fixture = MakeExporter([](const std::shared_ptr<http_client::EventHandler> &handler) {
     FakeResponse response(200, kAcceptedBody);
-    handler.OnResponse(response);
+    handler->OnResponse(response);
   });
   ExportOnce(*fixture.exporter);
 
@@ -332,13 +326,14 @@ TEST(ElasticsearchForceFlushTests, SucceedsWhenTheSessionFinishedBeforeTheWait)
 TEST(ElasticsearchForceFlushTests, PartialCompletionIsNotSuccess)
 {
   bool respond = true;
-  auto fixture = MakeExporter([&respond](http_client::EventHandler &handler) {
-    if (respond)
-    {
-      FakeResponse response(200, kAcceptedBody);
-      handler.OnResponse(response);
-    }
-  });
+  auto fixture =
+      MakeExporter([&respond](const std::shared_ptr<http_client::EventHandler> &handler) {
+        if (respond)
+        {
+          FakeResponse response(200, kAcceptedBody);
+          handler->OnResponse(response);
+        }
+      });
   ExportOnce(*fixture.exporter);
   respond = false;
   ExportOnce(*fixture.exporter);
@@ -350,10 +345,14 @@ TEST(ElasticsearchForceFlushTests, PartialCompletionIsNotSuccess)
 // the half the inline scripts above cannot reach.
 TEST(ElasticsearchForceFlushTests, ACompletionAfterTheWaiterParksWakesIt)
 {
-  auto fixture = MakeExporter([](http_client::EventHandler &) {});
+  // Held by the test, not by the fakes: a handler owns its session, so a session that also owned
+  // its handler would be a reference cycle and leak.
+  std::shared_ptr<http_client::EventHandler> captured;
+  auto fixture =
+      MakeExporter([&captured](const std::shared_ptr<http_client::EventHandler> &handler) {
+        captured = handler;
+      });
   ExportOnce(*fixture.exporter);
-  ASSERT_NE(fixture.client->last_session_, nullptr);
-  const auto captured = fixture.client->last_session_->last_handler_;
   ASSERT_NE(captured, nullptr);
 
   std::thread responder([&captured] {
@@ -366,4 +365,27 @@ TEST(ElasticsearchForceFlushTests, ACompletionAfterTheWaiterParksWakesIt)
   responder.join();
   EXPECT_TRUE(flushed);
 }
+
+// A second caller gets its own deadline. Serialising the calls is fine, making the second one
+// wait out the first one's is not, since its timeout would mean nothing.
+TEST(ElasticsearchForceFlushTests, AConcurrentFlushKeepsItsOwnDeadline)
+{
+  std::vector<std::shared_ptr<http_client::EventHandler>> kept;
+  auto fixture = MakeExporter([&kept](const std::shared_ptr<http_client::EventHandler> &handler) {
+    kept.push_back(handler);
+  });
+  ExportOnce(*fixture.exporter);
+
+  std::thread slow([&fixture] { fixture.exporter->ForceFlush(std::chrono::seconds{3}); });
+  std::this_thread::sleep_for(std::chrono::milliseconds{100});
+
+  const auto start = std::chrono::steady_clock::now();
+  EXPECT_FALSE(fixture.exporter->ForceFlush(std::chrono::milliseconds{20}));
+  const auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
+                      std::chrono::steady_clock::now() - start)
+                      .count();
+  slow.join();
+
+  EXPECT_LT(ms, 1000) << "waited behind the first caller instead of its own deadline";
+}
 #endif  // ENABLE_ASYNC_EXPORT