[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.

Retirement happens under the mutex the waiter holds. It used to be an atomic counter 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. The counter itself is gone, replaced further down by ids in an ordered set, and this paragraph is about the mutex rather than about what it protects.

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_. The no-deadline form is deliberately unbounded. An ordinarily working HTTP client drives each export to a terminal outcome through its own request deadline, but ForceFlush adds no bound of its own, and [BUG] Complete a curl operation that is never scheduled #4395 describes curl operations that receive a future and are never scheduled, so the request timeout never gets to run on them. 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

The cases use a fake HTTP client injected through the public constructor. The wait itself is described by:

• 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 others pass either way, and I would rather name that than let a list of cases look like a list of guards. They exercise paths that return before any wait, or that the old polling loop happened to get right.

Two more were added after a review pointed at what the list above does not hold, and both were confirmed by mutating the predicate rather than by reading it. Each of these left the suite green:

if (timeout <= 0) { return flushed(); }                  // the no deadline branch
return running.empty();                                  // the whole watermark
return running.empty() || *running.begin() > watermark;  // the boundary

The first survived because the case named for the indefinite wait never parks: the fake answers from inside SendRequest(), so the predicate already holds when ForceFlush() is called. AnIndefiniteFlushParksUntilTheOutcomeArrives waits with nothing answering, checks it has not returned, and then answers. The other two survived because nothing held the opposite boundary from the substitution cases: those say a newer completion cannot finish an older flush, and ANewerSessionDoesNotHoldAnOlderFlushOpen says a newer export still running cannot keep that flush open. All three mutations are caught now.

Four cases used a 50 or 100 millisecond sleep to put an export, a completion or a second caller after the flush's snapshot. That leaves the order to the scheduler, and in the other order the newer work is inside the snapshot, where the counting model these cases exist to rule out reports a pass too. So ForceFlush() raises a counter under the mutex it already holds, immediately after reading its watermark, and the cases wait on that. Nothing in the exporter reads it.

The exactly-once cases read the completion count while the handler they retained is still alive, so its destructor had not run and a second report from there would have landed after the check. Making the destructor bypass CompleteOnce() left the suite green; the handler is released before a second read now, and three cases catch it.

What the log ordering does not buy

Retiring an export before its terminal diagnostic keeps a flushing log handler from waiting on the session that is reporting itself, and AFlushFromInsideTheLogHandlerDoesNotWaitForItsOwnSession holds that. It is not general re-entrancy safety and the code now says so.

A handler that flushes from any callback the HTTP client dispatches can wait on work only that client thread advances. That one is not introduced here and not fixed here, and #4435 has the detail.

The other half of what I wrote there was wrong, so it is worth correcting rather than quietly dropping. I said the Created debug line was already on main. It is, in the synchronous ResponseHandler, and it is not in the asynchronous one, which is the handler this change is about and the only one whose ForceFlush() waits for anything. On 60c3d11e that handler's OnEvent is a default: break; with no progress logging at all. So the progress diagnostics were new on that path, and they are gone again below.

One more thing a reviewer asked about and I have deliberately left alone. The scope guard's active destructor is not reachable from any current return: the shutdown branch disarms it and retires explicitly, and there is no other return before the handler takes over. It is there for a return added later, which is what its comment says. Adding the missing null checks for an injected HttpClient, CreateSession() and CreateRequest() would give it something reachable to do, and those dereferences really are unguarded today, but that is a different defect from the one this changes and it wants its own case rather than a ride on this one.

One thing the deterministic wait broke on the way in, worth naming because it is the same trap the cases already avoid. The helper they wait on went behind ENABLE_ASYNC_EXPORT, while the cases that call it are compiled in both configurations on purpose: a case removed from the binary stays registered with CTest and reports a pass without running, so they skip in SetUp instead. Six call sites then failed to compile in a synchronous build. The helper has a stub there now, inline so an unused static function does not trip the maintainer mode ratchet.

Verification

[ PASSED ] 28 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 28 pass under AddressSanitizer with leak detection on and under ThreadSanitizer with no warnings, the latter checked for 161 __tsan_ symbols in the binary so that no warnings means it was looking. Synchronous with OTELCPP_MAINTAINER_MODE=ON builds clean and skips what it should, asynchronous with it builds with no warnings, and ./ci/do_ci.sh format exits 0 with no diff, 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 they all 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.

What a second reading of this branch changed

The states a session passes through on the way to an outcome report nothing, and in particular log
nothing. The session is registered before the request is handed to the client, and none of those
events retires it, so a log handler that flushed from one waited for the export whose call stack it
was standing in. Every other report in this file already retires before it logs, which is what made
the omission visible: the progress cases were the one place that broke the file's own rule. The
switch is still exhaustive, so a new state is still a compile error.

A flush with no deadline of its own now gets one. The bound is response_timeout_, which is what
this path waited for before the accounting was fixed and is also the value the request itself is
given, so a session outstanding when the watermark is taken has at most that long before the client
owes it a terminal event. The bound is unchanged and the answer is what changes: running out of it
no longer reports success. That also keeps this independent of the curl client work, where a
transfer can currently be left with no callback at all.

The deadline is taken at entry rather than after the mutex, so time spent waiting for the lock
comes out of the caller's budget instead of being handed back as a fresh one, and ForceFlush has
one wait path rather than two.

GiveUpGuard is gone. It could not reach its active destructor: the shutdown return retired by
hand and disarmed it, the successful path disarmed it after SendRequest, and there was no other
return between them. The handler takes the completion by move rather than by copy, and the
invariant the guard was watching is stated where the two exits are.

ForceFlush and Shutdown are required to be safe to call concurrently by the stable Logs SDK
specification and had no case for it. Two now: one where the shutdown settles the outstanding
session and the flush is woken by that rather than by its own bound, and one where the shutdown
settles nothing and the flush ends anyway.

WaitForWatermarks had the same thirty second bound as the whole CTest case, so a watermark that
never arrived killed the process before the case reached its own assertion. Five seconds for the
helper, sixty for CTest. The CMake comment named a case that never parks and now names the one that
does. watermarks_taken is labelled as test synchronization rather than left looking like exporter
state.

Each of these is held by putting the defect back: a progress line restored fails
EverySessionStateIsClassifiedAndReportsAtMostOnce, the bound removed fails
ANoDeadlineFlushGivesUpAtTheResponseTimeout, and the unbounded wait restored hangs
AParkedFlushEndsEvenIfTheShutdownSettlesNothing at exit 124. All 31 cases pass in the async
build, under AddressSanitizer and under ThreadSanitizer, with no sanitizer report and the same
count in each.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 83.19%. Comparing base (60c3d11) to head (069dfde).
⚠️ Report is 4 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4337      +/-   ##
==========================================
+ Coverage   82.61%   83.19%   +0.58%     
==========================================
  Files         511      511              
  Lines       20132    20153      +21     
==========================================
+ Hits        16631    16765     +134     
+ Misses       3501     3388     -113     

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

... and 2 files with indirect coverage changes

🚀 New features to boost your workflow:

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