[BUG] Stop the Elasticsearch async ForceFlush reporting success without waiting

[thc1006]

Fixes #4336.

ForceFlush waited for options_.response_timeout_ rather than the time the caller gave it, and the timeout branch left the loop without subtracting what it had just spent:

while (timeout_steady > std::chrono::steady_clock::duration::zero())
{
  if (finished_session_counter_.load(...) >= running_counter) { break; }
  auto start = std::chrono::steady_clock::now();
  if (std::cv_status::no_timeout != force_flush_cv.wait_for(
                                        lk_cv, std::chrono::seconds{options_.response_timeout_}))
  {
    break;
  }
  timeout_steady -= std::chrono::steady_clock::now() - start;
}
return timeout_steady > std::chrono::steady_clock::duration::zero();

The loop condition has already established that timeout_steady is positive, so the return on that path is true whatever happened. Every flush that ran out of time reported success. The only way to get false was to be notified repeatedly without completing until the subtraction drained the budget.

That contradicts what the method promises in es_log_record_exporter.h, return true when all data are exported, and false when timeout, and it works against the spec requirement that a processor MUST prioritize honoring the timeout over finishing all calls.

The change

One steady_clock deadline derived from the caller's timeout, and a wait on a completion predicate. wait_until returns the predicate, so the return value is now the answer to the question the caller asked rather than a leftover duration.

Serialising concurrent calls was a second way to miss the deadline: force_flush_m was taken unconditionally at the top, so a second caller waited out the first one's wait however short its own timeout was. Measured at 2921 ms for a ForceFlush(20ms) behind a 3 second one.

That lock is gone rather than made timed. It protected nothing: each call snapshots the session counter it waits for and waits on its own predicate, and the wait publishes no state, so two callers were already safe side by side. It was declared in the header and used in exactly one place. Dropping it changes the layout of SynchronizationData, which is declared in the installed header, but that struct is a private member and the whole block is behind ENABLE_ASYNC_EXPORT, so it carries no ABI promise. There is no ABI diff job in CI to catch or complain about it either way. I first made it a timed acquisition instead, which ThreadSanitizer could not see through, since libstdc++ routes try_lock_until to pthread_mutex_clocklock and libtsan does not intercept it. Deleting the lock was the better answer to the same problem and that revision is not in this diff.

The completion counter moves under the mutex the waiter holds. It was an atomic incremented outside it, which does not lose the state but does lose the notification when it lands between the waiter's predicate check and its park.

AdjustWaitForTimeout already maps microseconds::max() and anything that would overflow now() + timeout to zero, so zero is the sentinel for a caller that asked for no deadline, and that branch waits until the flush actually completes.

Two notes on the edges of that:

• With no deadline this now blocks until the sessions finish, where before it returned true after one response_timeout_. For the bundled curl client the wait is bounded anyway, because Export() sets CURLOPT_TIMEOUT_MS from response_timeout_ and every session therefore terminates and dispatches. An HTTP client injected through the factory that accepts a request and never calls back would block, where it used to be told the flush had succeeded. Blocking on an unbounded wait is what BatchLogRecordProcessor::ForceFlush does, so this matches the SDK, but say so if you would rather it stayed bounded and returned false.
• The flush still covers the sessions that were running when it was entered, which is what the counter snapshot already meant. I asked in [BUG] Elasticsearch async ForceFlush reports success on timeout and ignores the caller deadline #4336 whether it should instead keep waiting while new exports arrive, and left the existing behaviour until you have a view.

Tests

Eight cases, with a fake HTTP client injected through the public constructor:

• a session that never calls back makes the flush report failure, at the caller's deadline rather than at response_timeout_,
• nothing in flight returns at once,
• a session that finished before the flush was asked for returns at once, which is the predicate being true on first evaluation,
• two exports where only one finishes is not a success,
• a completion delivered from another thread after the waiter has parked wakes it,
• a second caller arriving while the first is parked returns on its own deadline,
• a flush with no deadline returns once everything has finished,
• a failed export still finishes its session, so the flush completes even though the export did not.

Three of those fail if the change is reverted: the two that measure a flush which cannot complete, and the concurrent one, which on the old code queues behind the first caller. The other five pass either way, and I would rather name that than let eight cases look like eight guards. They exercise paths that return before any wait, or that the old polling loop happened to get right.

The counter moving under force_flush_cv_m has no case of its own. It closes a window between the waiter's predicate check and its park, and the fake completes from a thread that sleeps first, so the window is never hit.

Verification

[ PASSED ] 22 tests. built with WITH_ELASTICSEARCH=ON -DWITH_ASYNC_EXPORT_PREVIEW=ON, and the same binary in the synchronous configuration, where the cases that describe a wait skip in SetUp. Both build with no warnings under OTELCPP_MAINTAINER_MODE=ON.

Restoring the previous ForceFlush and rebuilding the same tests fails the ones that should discriminate:

[  FAILED  ] ElasticsearchForceFlushTests.ReportsFailureWhenTheFlushDoesNotComplete (2000 ms)
[  FAILED  ] ElasticsearchForceFlushTests.PartialCompletionIsNotSuccess (2000 ms)
[  FAILED  ] ElasticsearchForceFlushTests.AConcurrentFlushKeepsItsOwnDeadline (4000 ms)
[  PASSED  ] 5 tests.
[  FAILED  ] 3 tests, listed below:

The 2000 ms is the whole response_timeout_ being spent against a 20 ms caller deadline, which is the second half of the defect.

./ci/do_ci.sh format exits 0 with no diff. All 22 pass under AddressSanitizer with leak detection on and under ThreadSanitizer with no warnings, which the two cases that deliver a response and a terminal event from separate threads are the reason to check rather than assume.

include-what-you-use and clang-tidy were measured against main rather than in isolation, and over the test target so that the test file is compiled rather than only the library. Across all three cmake option presets the workflow builds, include-what-you-use reports the same blocks and no include changes on either tree; on all-options-abiv2-preview clang-tidy reports the same eighteen warnings on either, with nothing on this branch that is not also on main. The async-only includes sit behind the ENABLE_ASYNC_EXPORT guard, because the presets that build without it ask for them to go while the ones that build it ask for them.

The three the first revision left open

The first revision fixed the deadline and named three other ways this function
reports success without having waited. They are the same defect from the caller's
side, and two of them make the deadline fix meaningless on its own, so they are
here rather than in follow-ups.

A session could be counted twice, or not at all. OnResponse and every
terminal event called the result callback directly with no guard, while
ReadError, WriteError and Destroyed fell through a default label and
called nothing. One session could therefore finish twice, which lets the total
overshoot and stay overshot for the life of the exporter, or never finish, which
leaves an undeadlined flush waiting forever. Every path now goes through one
CompleteOnce, a compare-exchange that reports at most once and keeps the first
verdict. The switch lists every state with no default, so a state added
upstream fails to compile rather than going uncounted, and the destructor reports
a failure for a handler torn down without an outcome.

A completion satisfied any waiter. Both counters were monotonic totals with
no session identity. A flush entering with two sessions outstanding waits for two
completions; a third session started afterwards and completed, one of the
original two completed, the count reached two, and the flush reported success
with the other original still running. Sessions now carry an id and the running
ones live in an ordered set. The snapshot is the next id to be issued and ids are
issued in order, so the smallest one still running decides.

Why a set of what is running rather than a completed-sequence frontier, since
both are correct. Tracking what is outstanding and waiting for it to drain is
what the neighbours do: this repository's own OtlpHttpClient keeps
running_sessions_ and waits for it to empty, SimpleSpanProcessor in
opentelemetry-java keeps a Set<CompletableResultCode> pendingExports and
returns ofAll of it, and opentelemetry-go's batch processor uses a
WaitGroup. A frontier answers a stronger question, whether a contiguous
prefix is complete, which is what write-ahead logs and replication need; the
extra strength is what costs the unbounded buffer, because one stalled session
holds back the prefix even though every completion behind it is individually
known. Measured over the same traces, one stalled session with a million
completions after it leaves 1000000 entries in the frontier's buffer and 2 in
this one. BatchLogRecordProcessor here does use a sequence, and correctly so:
its work is drained in order by one worker, where a monotone acknowledgement is
exactly the right model.

Two of those neighbours also over-wait: running_sessions_.empty() and a
WaitGroup both include sessions started after the call. The watermark is what
keeps this one to the sessions the caller asked about.

The predicate was checked against an independent oracle over 20000 random
schedules, 54 million evaluations, with sessions starting, completing out of
order, and flushes taking snapshots in between:

predicate	duplicate completions	reported flushed too early	reported not flushed when it was
the counter this replaces	no	9000364	0
this one	no	0	0
the counter this replaces	yes	22488236	0
this one	yes	0	0

The counter's column is what makes the zero meaningful: the same check finds
the defect it replaces. The right hand column is the other half, that this does
not over-wait either, and the duplicate rows are the exactly-once fix and the
tracker covering each other.

A batch already inside Export() was not waited for. The session was
registered after the request had been created and the whole batch serialised into
its body, so a flush asked during that window snapshotted past a batch whose
records had already been handed over. The Logs SDK draws the line at records
received prior to the call, so they belong to it. Registration moves to the top
of Export(), with a scope guard that releases the id if the call gives up
before a handler takes it over. Nothing between the two returns today and
Export() is noexcept, but CreateSession() can return null without being
checked, and the check that eventually adds that early return would otherwise
leave a waiter blocked on a session that can never finish.

What it still does not fix

true means the sessions the call snapshotted have settled, not that their
batches reached Elasticsearch. A failed export settles its session and reports
through the internal log, so a flush can return true for a batch that was
rejected. That is
#3075 rather
than something this changes; the header's @return now says so instead of
promising that all data are exported.

Shutdown(timeout) still ignores its timeout, calls CancelAllSessions() and
FinishAllSessions() in order, and returns true unconditionally. Separate, and
not something to read this pull request as having fixed.

The cases build in every configuration and skip in SetUp where the wait does not exist. An earlier revision compiled them out instead, which was wrong in a way worth naming: gtest_add_tests registers from the source, so all eight stayed registered with CTest in the synchronous jobs, and a gtest filter matching nothing exits zero, so each reported a pass without running. Skipping in SetUp rather than at the top of each body also keeps GTEST_SKIP, which returns, from leaving the rest of a body unreachable, which MSVC reports as C4702.

Each of the three defects is pinned by reverting it: moving the registration back below the request build turns AnExportAlreadyUnderWayIsSomethingToWaitFor red, restoring the counter comparison turns the two substitution cases red, and removing the compare-exchange from CompleteOnce() turns eight of the nine completion cases red. That last number is the reason those cases count the callback through a log handler rather than through ForceFlush(): sessions are identified rather than counted, so a repeated completion erases an id that has already gone, and every flush-based assertion passed with the guard removed.

The cases also carry a thirty second CTest timeout. AnIndefiniteFlushReturnsOnceEverythingIsFinished exercises the branch that waits with no deadline, so if its precondition ever stops holding the case does not fail, it stops, and CTest's own default bound is twenty five minutes. The slowest case takes 1.5 seconds today, and that 1.5 seconds is a deliberate wait rather than compute, so it does not stretch with runner load.

Landing next to the other Elasticsearch changes

exporters/elasticsearch/test/es_log_record_exporter_test.cc is also touched by #4297 and #4331, and all three add a fake HTTP client to it, so any two of them conflict there. #4331 adds the same set_tests_properties(... TIMEOUT 30) line to exporters/elasticsearch/CMakeLists.txt that this one does. Whichever lands first, I rebase the rest onto it and drop the duplicate.

[codecov]

Codecov Report

❌ Patch coverage is 98.83721% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 81.43%. Comparing base (43b656d) to head (eb9d9e5).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
...orters/elasticsearch/src/es_log_record_exporter.cc	98.84%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4337      +/-   ##
==========================================
+ Coverage   81.00%   81.43%   +0.43%     
==========================================
  Files         448      450       +2     
  Lines       19121    19221     +100     
==========================================
+ Hits        15488    15650     +162     
+ Misses       3633     3571      -62     

Files with missing lines	Coverage Δ
...y/exporters/elasticsearch/es_log_record_exporter.h	100.00% <ø> (ø)
...orters/elasticsearch/src/es_log_record_exporter.cc	92.17% <98.84%> (+79.96%)	⬆️

... and 8 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.