[BUG] Retire a refused export before describing the refusal

The export is registered before the shutdown check, and the guard that
retires it could only run after the refusal had already been described. The
log handler is replaceable, so a handler that calls ForceFlush() from that
error waited for the Export() that was calling it. The same path also
described one refusal twice, once itself and once through the guard.

Retiring is now separate from reporting. The refusal retires, disarms the
guard, and says so once. The completion still answers true, because changing
what it answers would change what the response handler reads.

Measured with a handler that calls ForceFlush(20ms) from the shutdown error.
Before: the flush returns false after waiting out its full 20 ms, 3 of 3.
After: it returns true in 0 ms, 3 of 3, and one refusal reads as one line.
The file holds at 25 passing, 3 of 3.

Three statements alongside it claimed more than they hold. An array cannot
notice that an enum grew, so the comment now names the switch without a
default under -Wswitch as what catches a new state. The curl completion
lambda tests the response first and the abort in an else, so the comment
that called it two independent ifs is out of date. And the fixture put a
fresh default log handler back rather than the one it found.

Signed-off-by: thc1006 <84045975+thc1006@users.noreply.github.com>

Author: thc1006

exporters/elasticsearch/src/es_log_record_exporter.cc

@@ -458,16 +458,24 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
     synchronization_data_->running_sessions.insert(session_id);
   }
 
-  using Completion    = std::function<bool(opentelemetry::sdk::common::ExportResult)>;
-  Completion complete = [span_count, session_id,
-                         synchronization_data](opentelemetry::sdk::common::ExportResult result) {
+  // Retiring the export is separate from describing what happened to it, because the refusal
+  // below has to retire without having an outcome to report through the completion.
+  auto retire = [session_id, synchronization_data]() noexcept {
+    bool retired = false;
     {
       // Published under the mutex ForceFlush() waits on. A waiter that has evaluated its
       // predicate but not yet parked would otherwise not see this until the next wakeup.
       std::lock_guard<std::mutex> lock(synchronization_data->force_flush_cv_m);
-      synchronization_data->running_sessions.erase(session_id);
+      retired = synchronization_data->running_sessions.erase(session_id) == 1;
     }
     synchronization_data->force_flush_cv.notify_all();
+    return retired;
+  };
+
+  using Completion    = std::function<bool(opentelemetry::sdk::common::ExportResult)>;
+  Completion complete = [span_count,
+                         retire](opentelemetry::sdk::common::ExportResult result) noexcept {
+    retire();
 
     // Logged after the session is retired. The log handler is replaceable, and one that calls
     // ForceFlush() would otherwise wait for the very session this call has not let go of yet.
@@ -512,6 +520,12 @@ sdk::common::ExportResult ElasticsearchLogRecordExporter::Export(
   // Return failure if this exporter has been shutdown
   if (isShutdown())
   {
+#ifdef ENABLE_ASYNC_EXPORT
+    // Retired before anything replaceable runs, and reported here rather than through the guard,
+    // so a log handler that flushes does not wait for this Export() and one refusal reads as one.
+    guard.report = nullptr;
+    retire();
+#endif
     OTEL_INTERNAL_LOG_ERROR("[ES Log Exporter] Exporting "
                             << records.size() << " log(s) failed, exporter is shutdown");
     return sdk::common::ExportResult::kFailure;

exporters/elasticsearch/test/es_log_record_exporter_test.cc

@@ -626,6 +626,7 @@ class ElasticsearchAsyncCompletionTests : public ::testing::Test
 #else
     // One skip point, because GTEST_SKIP returns and a second one after it would leave the rest of
     // this body unreachable, which MSVC reports as C4702 under maintainer mode.
+    previous_handler_ = internal_log::GlobalLogHandler::GetLogHandler();
     handler_ = nostd::shared_ptr<internal_log::LogHandler>(new CompletionCountingLogHandler());
     internal_log::GlobalLogHandler::SetLogHandler(handler_);
     previous_level_ = internal_log::GlobalLogHandler::GetLogLevel();
@@ -638,8 +639,7 @@ class ElasticsearchAsyncCompletionTests : public ::testing::Test
     if (handler_)
     {
       internal_log::GlobalLogHandler::SetLogLevel(previous_level_);
-      internal_log::GlobalLogHandler::SetLogHandler(
-          nostd::shared_ptr<internal_log::LogHandler>(new internal_log::DefaultLogHandler()));
+      internal_log::GlobalLogHandler::SetLogHandler(previous_handler_);
     }
   }
 
@@ -651,6 +651,7 @@ class ElasticsearchAsyncCompletionTests : public ::testing::Test
   int Completions() const { return Counter().completions(); }
 
   nostd::shared_ptr<internal_log::LogHandler> handler_;
+  nostd::shared_ptr<internal_log::LogHandler> previous_handler_;
   internal_log::LogLevel previous_level_ = internal_log::LogLevel::Warning;
 };
 }  // namespace
@@ -664,8 +665,9 @@ TEST_F(ElasticsearchAsyncCompletionTests, EverySessionStateIsClassifiedAndReport
     bool terminal;
   };
 
-  // Every member of the enum, so a state added upstream shows up here as a missing row rather than
-  // as a session that quietly never finishes. The switch has no default label for the same reason.
+  // Every member of the enum as it stands. An array cannot notice that the enum grew, so what
+  // catches a new state is the switch in the exporter having no default, under the -Wswitch that
+  // maintainer mode turns into an error. This table is the second half of that.
   const Case cases[] = {
       {http_client::SessionState::CreateFailed, true},
       {http_client::SessionState::Created, false},
@@ -846,9 +848,13 @@ namespace
 class FlushingLogHandler : public internal_log::LogHandler
 {
 public:
-  void Watch(logs_exporter::ElasticsearchLogRecordExporter *exporter) noexcept
+  // The needle picks which diagnostic re-enters the exporter, because the two paths that log
+  // one describe it differently.
+  void Watch(logs_exporter::ElasticsearchLogRecordExporter *exporter,
+             const char *needle = "Logs were not written") noexcept
   {
     exporter_ = exporter;
+    needle_   = needle;
   }
 
   void Handle(internal_log::LogLevel /* level */,
@@ -861,10 +867,11 @@ class FlushingLogHandler : public internal_log::LogHandler
     {
       return;
     }
-    if (std::string(msg).find("Logs were not written") == std::string::npos)
+    if (std::string(msg).find(needle_) == std::string::npos)
     {
       return;
     }
+    lines_.fetch_add(1, std::memory_order_relaxed);
     if (reentered_.exchange(true, std::memory_order_relaxed))
     {
       return;
@@ -874,11 +881,14 @@ class FlushingLogHandler : public internal_log::LogHandler
 
   bool reentered() const noexcept { return reentered_.load(std::memory_order_relaxed); }
   bool flushed() const noexcept { return flushed_.load(std::memory_order_relaxed); }
+  int lines() const noexcept { return lines_.load(std::memory_order_relaxed); }
 
 private:
   logs_exporter::ElasticsearchLogRecordExporter *exporter_{nullptr};
+  const char *needle_{"Logs were not written"};
   std::atomic<bool> reentered_{false};
   std::atomic<bool> flushed_{false};
+  std::atomic<int> lines_{0};
 };
 }  // namespace
 
@@ -903,6 +913,27 @@ TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromInsideTheLogHandlerDoesNotWa
   raw->Watch(nullptr);
 }
 
+// The same rule on the path that refuses the batch. The export is registered before the shutdown
+// check, so reporting the refusal before retiring it makes a flushing handler wait for the
+// Export() that is calling it, and the refusal is described twice.
+TEST_F(ElasticsearchAsyncCompletionTests, AFlushFromTheShutdownErrorDoesNotWaitForItsOwnExport)
+{
+  auto fixture = MakeExporter([](const std::shared_ptr<http_client::EventHandler> &) {});
+  ASSERT_TRUE(fixture.exporter->Shutdown());
+
+  auto watcher = nostd::shared_ptr<internal_log::LogHandler>(new FlushingLogHandler());
+  auto *raw    = static_cast<FlushingLogHandler *>(watcher.get());
+  raw->Watch(fixture.exporter.get(), "exporter is shutdown");
+  internal_log::GlobalLogHandler::SetLogHandler(watcher);
+
+  ExportOnce(*fixture.exporter);
+
+  ASSERT_TRUE(raw->reentered()) << "the shutdown refusal never reached the log handler";
+  EXPECT_TRUE(raw->flushed()) << "the flush waited for the export that was refusing itself";
+  EXPECT_EQ(1, raw->lines()) << "one refusal was described " << raw->lines() << " times";
+  raw->Watch(nullptr);
+}
+
 // Two responses for one request write the same body and race for the same outcome. The body is a
 // local so there is nothing shared to tear, and the exchange decides which one reports.
 TEST_F(ElasticsearchAsyncCompletionTests, TwoConcurrentResponsesReportOnce)
@@ -958,8 +989,9 @@ TEST_F(ElasticsearchAsyncCompletionTests, TwoTerminalEventsCountAsOneSession)
       << "the second session is still in flight";
 }
 
-// The other in-tree double fire: the completion lambda uses two independent ifs, so an aborted
-// operation that also has a response reports Cancelled and then delivers the response.
+// The curl client's completion lambda tests the response first and the abort in an else, so this
+// ordering no longer comes from it. It still comes from anywhere: EventHandler promises callers
+// nothing about how many terminal events arrive, and #4360 is open on exactly that.
 TEST_F(ElasticsearchAsyncCompletionTests, CancelledThenAResponseCountsOnce)
 {
   std::vector<std::shared_ptr<http_client::EventHandler>> kept;