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

[thc1006]

Fixes #4336. Fixes #4338.

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.

That argument holds only while the ids keep increasing, so they are uint64_t rather than size_t. A 32 bit counter reaches its end in days at a rate this exporter is built to sustain, and the watermark taken after that point is small enough for a session that is still running to satisfy it, which is the failure that matters rather than the harmless one.

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 every export the call snapshotted has reported a terminal outcome, not that their batches reached Elasticsearch. A failed export 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 says so instead of promising that all data are exported.

It is also not a statement about the transport. Every completion path publishes the outcome first: OnResponse and the terminal OnEvent states each ahead of their logging, and the handler destructor ahead of FinishSession(). That order is deliberate, since the log handler is replaceable and one that calls ForceFlush() would otherwise wait on the session its own call is still holding. The @return says that too now. It used to claim the session had ended, which the code never promised.

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.

Its narrower cousin is worth naming too, since this changes the code around it. is_shutdown_ is atomic, but the check in Export() and the cancellation in Shutdown() are not one step: an export that has already passed the check and is still building its session can hand a request to the client after Shutdown() has returned true. The window is the same on main, and registering the session id earlier does not widen it, since nothing moved between the check and CreateSession(). Closing it needs an admission gate the two share rather than a flag each reads on its own, which is a larger change than this one and not what the flush accounting here is about.

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.87640% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 82.65%. Comparing base (768d04d) to head (648de6e).
⚠️ Report is 1 commits behind head on main.

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

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4337      +/-   ##
==========================================
+ Coverage   82.11%   82.65%   +0.55%     
==========================================
  Files         494      494              
  Lines       19670    19706      +36     
==========================================
+ Hits        16151    16287     +136     
+ Misses       3519     3419     -100     

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

... and 1 file with indirect coverage changes

🚀 New features to boost your workflow:

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