fix(executor): make wait a completion barrier

CHANGELOG.md

@@ -11,6 +11,11 @@ All notable changes to this project will be documented in this file.
 - Added a non-public binary log record format and reader research prototype;
   no production binary backend is enabled yet.
 
+### Fixed
+
+- Fixed a race in the MPSC `TaskExecutor::wait()` completion barrier that could
+  allow a wait to return before the final accepted task completed.
+
 ## [v1.0.2] - 2026-09-19
 
 ### Added

docs/TaskExecutor.md

@@ -29,9 +29,14 @@ integrations rely on.
 * Synchronisation primitives:
   * `m_cv` + `m_cv_mutex` coordinate sleepers for both the worker and producers
     that wait for capacity during `QueuePolicy::Block`.
-  * `m_queue_condition` wakes `wait()` callers once the queue drains.
+  * `m_queue_condition` coordinates worker/lifecycle drain notifications.
+  * A completion ticket is reserved before each MPSC submission is published;
+    `wait()` advances a completion frontier until every submission visible at
+    the completion check has completed or been rejected. This closes the
+    worker-side race between an empty-ring check and the next `try_pop()`
+    attempt.
   * `m_active_tasks` tracks in-flight work so that `Block` limits concurrent
-    execution and `wait()` can determine quiescence.
+    execution and lifecycle resize checks can observe quiescence.
   * `m_stop_flag` terminates the worker and stops accepting new tasks.
 * Enables very low producer overhead while maintaining FIFO ordering on the
   consumer side.
@@ -103,10 +108,14 @@ mutate the stopped singleton worker.
   accepted by the consumer.
 * When the ring build is enabled, `DropNewest` and `DropOldest` both drop the
   incoming task; accepted tasks keep their order.
-* `wait()` returns once the queue is empty and `m_active_tasks == 0`, or when a
-  shutdown is requested. In MPSC builds the worker marks a pop attempt active
-  before removing a task, so `wait()` cannot return in the narrow window between
-  a dequeued cell becoming free and the task body starting.
+* `wait()` advances a completion frontier until every submission visible at
+  the completion check has either completed or been rejected, or until a
+  shutdown is requested. Submissions created by already-running tasks before
+  the barrier closes are included, so continuous concurrent submissions may
+  delay completion. In MPSC builds this is enforced by completion tickets
+  rather than by an independent empty-ring/active-task observation, so
+  `wait()` cannot return in the narrow window between a dequeued cell becoming
+  free and the task body starting.
 * `shutdown()` blocks until the worker thread terminates. It is safe to call
   multiple times.
 

include/logit_cpp/logit/detail/TaskExecutor.hpp

@@ -213,12 +213,19 @@ namespace logit { namespace detail {
             m_queue_condition.notify_one();
 #        else
             enter_producer_();
+
+            // Reserve a completion ticket before attempting publication.  A
+            // waiter that observes this ticket must wait for either the task
+            // to run or the submission to be rejected, so it cannot race the
+            // worker between an empty-ring observation and try_pop().
+            m_submitted_tasks.fetch_add(1, std::memory_order_release);
 
             std::function<void()> local_task = std::move(task);
             bool done = false;
 
             while (!done) {
                 if (m_stop_flag.load(std::memory_order_acquire)) {
+                    complete_task_();
                     break;
                 }
 
@@ -248,13 +255,15 @@ namespace logit { namespace detail {
                 switch (policy) {
                     case QueuePolicy::DropNewest:
                         m_dropped_tasks.fetch_add(1, std::memory_order_relaxed);
+                        complete_task_();
                         done = true;
                         break;
 
                     case QueuePolicy::DropOldest:
                         // Safe MPSC behaviour: drop the incoming task.
                         // Preserves ordering and avoids producer/consumer deadlocks.
                         m_dropped_tasks.fetch_add(1, std::memory_order_relaxed);
+                        complete_task_();
                         done = true;
                         break;
 
@@ -279,12 +288,19 @@ namespace logit { namespace detail {
                         m_stop_flag.load(std::memory_order_acquire));
             });
 #        else
-            std::unique_lock<std::mutex> lock(m_queue_mutex);
-            m_queue_condition.wait(lock, [this]() {
-                return ((queue_empty_() &&
-                        m_active_tasks.load(std::memory_order_relaxed) == 0) ||
-                        m_stop_flag.load(std::memory_order_acquire));
-            });
+            std::unique_lock<std::mutex> completion_lock(m_completion_wait_mutex);
+            for (;;) {
+                const auto target = m_submitted_tasks.load(std::memory_order_acquire);
+                m_completion_cv.wait(completion_lock, [this, target]() {
+                    return m_completed_tasks >= target ||
+                           m_stop_flag.load(std::memory_order_acquire);
+                });
+                if (m_stop_flag.load(std::memory_order_acquire) ||
+                    m_completed_tasks >=
+                        m_submitted_tasks.load(std::memory_order_acquire)) {
+                    break;
+                }
+            }
 #        endif
         }
 
@@ -306,6 +322,7 @@ namespace logit { namespace detail {
             }
             m_cv.notify_all();
             m_queue_condition.notify_all();
+            m_completion_cv.notify_all();
             if (m_worker_thread.joinable()) {
                 m_worker_thread.join();
             }
@@ -413,6 +430,11 @@ namespace logit { namespace detail {
         std::condition_variable m_cv;              ///< Wakes the worker or producers.
         std::mutex m_cv_mutex;                     ///< Protects producer/worker sleeps.
 
+        std::mutex m_completion_wait_mutex;       ///< Serializes completion waiters.
+        std::condition_variable m_completion_cv;   ///< Notifies completion waiters.
+        std::atomic<std::size_t> m_submitted_tasks; ///< Reserved submission tickets.
+        std::size_t m_completed_tasks;             ///< Completed submission tickets.
+
         std::atomic<bool> m_resizing;              ///< true while a hot resize is in flight.
         std::condition_variable m_resize_cv;       ///< Producers wait here during a resize.
         std::atomic<std::size_t> m_active_producers; ///< Producers currently touching the ring.
@@ -473,6 +495,8 @@ namespace logit { namespace detail {
                     drained_any = true;
 
                     task();
+
+                    complete_task_();
 
                     m_active_tasks.fetch_sub(1, std::memory_order_relaxed);
                     m_cv.notify_one(); // freed an in-flight slot
@@ -504,12 +528,33 @@ namespace logit { namespace detail {
         }
 
         bool wait_until_idle_(std::chrono::steady_clock::time_point deadline) {
-            std::unique_lock<std::mutex> lock(m_queue_mutex);
-            return m_queue_condition.wait_until(lock, deadline, [this]() {
-                return ((queue_empty_() &&
-                         m_active_tasks.load(std::memory_order_relaxed) == 0) ||
-                        m_stop_flag.load(std::memory_order_acquire));
-            });
+            std::unique_lock<std::mutex> lock(m_completion_wait_mutex);
+            for (;;) {
+                const auto target = m_submitted_tasks.load(std::memory_order_acquire);
+                if (m_completed_tasks < target &&
+                    !m_completion_cv.wait_until(lock, deadline, [this, target]() {
+                        return m_completed_tasks >= target ||
+                               m_stop_flag.load(std::memory_order_acquire);
+                    })) {
+                    return false;
+                }
+                if (m_stop_flag.load(std::memory_order_acquire) ||
+                    m_completed_tasks >=
+                        m_submitted_tasks.load(std::memory_order_acquire)) {
+                    return true;
+                }
+                if (std::chrono::steady_clock::now() >= deadline) {
+                    return false;
+                }
+            }
+        }
+
+        void complete_task_() {
+            {
+                std::lock_guard<std::mutex> lock(m_completion_wait_mutex);
+                ++m_completed_tasks;
+            }
+            m_completion_cv.notify_all();
         }
 
         void enter_producer_() {
@@ -561,6 +606,8 @@ namespace logit { namespace detail {
               m_overflow_policy(QueuePolicy::Block),
               m_dropped_tasks(0),
               m_active_tasks(0),
+              m_submitted_tasks(0),
+              m_completed_tasks(0),
               m_mpsc_queue(m_default_ring_cap)
     #endif
         {

tests/CMakeLists.txt

@@ -83,6 +83,7 @@ else()
         scope_timer_test.cpp
         single_thread_executor_test.cpp
         task_executor_resize_race_test.cpp
+        task_executor_wait_barrier_test.cpp
         unique_file_logger_file_api_test.cpp
         unique_file_logger_set_queue_config_test.cpp
         windows_debug_logger_set_queue_config_test.cpp
@@ -137,6 +138,9 @@ else()
         if(test_name STREQUAL "logger_legacy_targeted_path_test")
             target_compile_definitions(${test_name} PRIVATE LOGIT_BENCH_LEGACY_REGISTRY=1)
         endif()
+        if(test_name STREQUAL "task_executor_wait_barrier_test")
+            set_tests_properties(${test_name} PROPERTIES TIMEOUT 30)
+        endif()
         if(LOGIT_WITH_OTLP AND test_name MATCHES "^otlp_http_logger_(integration|callback|gzip|zstd)_test$")
             target_include_directories(${test_name} PRIVATE
                 "${CMAKE_CURRENT_SOURCE_DIR}/../external/kurlyk/external/Simple-Web-Server")

tests/task_executor_wait_barrier_test.cpp

@@ -0,0 +1,151 @@
+#include <logit.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <cstddef>
+#include <iostream>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+namespace {
+
+bool test_wait_blocks_for_running_task(logit::detail::TaskExecutor& executor) {
+    std::mutex gate_mutex;
+    std::condition_variable gate_cv;
+    bool task_started = false;
+    bool release_task = false;
+
+    executor.add_task([&]() {
+        std::unique_lock<std::mutex> lock(gate_mutex);
+        task_started = true;
+        gate_cv.notify_all();
+        gate_cv.wait(lock, [&]() { return release_task; });
+    });
+
+    {
+        std::unique_lock<std::mutex> lock(gate_mutex);
+        if (!gate_cv.wait_for(lock, std::chrono::seconds(2), [&]() {
+                return task_started;
+            })) {
+            return false;
+        }
+    }
+
+    std::mutex completion_mutex;
+    std::condition_variable completion_cv;
+    bool waiter_started = false;
+    bool waiter_done = false;
+
+    std::thread waiter([&]() {
+        {
+            std::lock_guard<std::mutex> lock(completion_mutex);
+            waiter_started = true;
+        }
+        completion_cv.notify_all();
+
+        executor.wait();
+
+        {
+            std::lock_guard<std::mutex> lock(completion_mutex);
+            waiter_done = true;
+        }
+        completion_cv.notify_all();
+    });
+
+    bool blocked = false;
+    {
+        std::unique_lock<std::mutex> lock(completion_mutex);
+        if (completion_cv.wait_for(lock, std::chrono::seconds(2), [&]() {
+                return waiter_started;
+            })) {
+            blocked = !completion_cv.wait_for(lock, std::chrono::milliseconds(100), [&]() {
+                return waiter_done;
+            });
+        }
+    }
+
+    {
+        std::lock_guard<std::mutex> lock(gate_mutex);
+        release_task = true;
+    }
+    gate_cv.notify_all();
+    waiter.join();
+
+    return blocked && waiter_done;
+}
+
+bool test_wait_drains_mpsc_submissions(logit::detail::TaskExecutor& executor) {
+    constexpr std::size_t kRounds = 200;
+    constexpr std::size_t kProducers = 4;
+    constexpr std::size_t kTasksPerProducer = 64;
+
+    for (std::size_t round = 0; round < kRounds; ++round) {
+        std::atomic<bool> start(false);
+        std::atomic<std::size_t> completed(0);
+        std::vector<std::thread> producers;
+        producers.reserve(kProducers);
+
+        for (std::size_t producer = 0; producer < kProducers; ++producer) {
+            producers.emplace_back([&]() {
+                while (!start.load(std::memory_order_acquire)) {
+                    std::this_thread::yield();
+                }
+                for (std::size_t task = 0; task < kTasksPerProducer; ++task) {
+                    executor.add_task([&completed]() {
+                        completed.fetch_add(1, std::memory_order_relaxed);
+                    });
+                }
+            });
+        }
+
+        start.store(true, std::memory_order_release);
+        for (auto& producer : producers) {
+            producer.join();
+        }
+
+        executor.wait();
+        if (completed.load(std::memory_order_relaxed) !=
+            kProducers * kTasksPerProducer) {
+            return false;
+        }
+    }
+
+    return true;
+}
+
+bool test_wait_drains_nested_submission(logit::detail::TaskExecutor& executor) {
+    std::atomic<std::size_t> completed(0);
+    executor.add_task([&]() {
+        completed.fetch_add(1, std::memory_order_relaxed);
+        executor.add_task([&completed]() {
+            completed.fetch_add(1, std::memory_order_relaxed);
+        });
+    });
+    executor.wait();
+    return completed.load(std::memory_order_relaxed) == 2;
+}
+
+} // namespace
+
+int main() {
+    auto& executor = logit::detail::TaskExecutor::get_instance();
+    executor.wait();
+    executor.set_queue_policy(logit::detail::QueuePolicy::Block);
+    executor.set_max_queue_size(0);
+    executor.reset_dropped_tasks();
+
+    const bool running_task = test_wait_blocks_for_running_task(executor);
+    executor.wait();
+    const bool mpsc_submissions = test_wait_drains_mpsc_submissions(executor);
+    const bool nested_submission = test_wait_drains_nested_submission(executor);
+
+    executor.wait();
+    executor.reset_dropped_tasks();
+
+    const bool ok = running_task && mpsc_submissions && nested_submission;
+    std::cout << (ok ? "PASS" : "FAIL")
+              << ": task_executor_wait_barrier" << std::endl;
+    return ok ? 0 : 1;
+}