From e973531691723a64bdbd77f24c3f2af28e379983 Mon Sep 17 00:00:00 2001
From: Aster Seker <git.electroyar@gmail.com>
Date: Tue, 19 May 2026 14:51:39 +0300
Subject: [PATCH] fix(executor): align single-thread lifecycle semantics

Make SingleThreadExecutor::wait() drain already accepted tasks even after shutdown starts, and make queue policy updates no-op after shutdown in native and Emscripten implementations. Document the dedicated executor lifecycle contract and add a regression test for wait during concurrent shutdown.
---
 docs/TaskExecutor.md                          |  4 +-
 .../logit/detail/SingleThreadExecutor.hpp     |  7 ++-
 tests/single_thread_executor_test.cpp         | 60 +++++++++++++++++++
 3 files changed, 67 insertions(+), 4 deletions(-)

diff --git a/docs/TaskExecutor.md b/docs/TaskExecutor.md
index 6d0e046..3dc6afa 100644
--- a/docs/TaskExecutor.md
+++ b/docs/TaskExecutor.md
@@ -119,7 +119,9 @@ Logger backends with `Config::use_dedicated_executor=true` own a
 worker thread per configured logger, while single-threaded Emscripten builds use
 a cooperative per-instance queue. `Logger::shutdown()` calls each backend's
 `ILogger::shutdown()` hook before stopping the global executor so these
-logger-owned workers drain and stop cleanly.
+logger-owned workers drain and stop cleanly. For `SingleThreadExecutor`,
+`wait()` waits for already accepted tasks to drain, while `shutdown()` rejects
+new tasks and joins the worker on native builds.
 
 ## 6. Emscripten (no pthreads)
 
diff --git a/include/logit_cpp/logit/detail/SingleThreadExecutor.hpp b/include/logit_cpp/logit/detail/SingleThreadExecutor.hpp
index d8ca1d1..18da07e 100644
--- a/include/logit_cpp/logit/detail/SingleThreadExecutor.hpp
+++ b/include/logit_cpp/logit/detail/SingleThreadExecutor.hpp
@@ -89,9 +89,8 @@ class SingleThreadExecutor {
     void wait() {
         std::unique_lock<std::mutex> lock(m_mutex);
         m_cv.wait(lock, [this]() {
-            return (m_queue.empty() &&
-                    m_active_tasks.load(std::memory_order_relaxed) == 0) ||
-                   m_stop.load(std::memory_order_acquire);
+            return m_queue.empty() &&
+                   m_active_tasks.load(std::memory_order_relaxed) == 0;
         });
     }
 
@@ -127,6 +126,7 @@ class SingleThreadExecutor {
     /// \brief Change the queue overflow policy.
     void set_queue_policy(QueuePolicy policy) {
         std::lock_guard<std::mutex> lock(m_mutex);
+        if (m_stop.load(std::memory_order_acquire)) return;
         m_overflow_policy = policy;
     }
 
@@ -274,6 +274,7 @@ class SingleThreadExecutor {
 
     void set_queue_policy(QueuePolicy policy) {
         std::lock_guard<std::mutex> lock(m_state->mutex);
+        if (m_state->shutdown_requested) return;
         m_state->overflow_policy = policy;
     }
 
diff --git a/tests/single_thread_executor_test.cpp b/tests/single_thread_executor_test.cpp
index 0785cd4..2e2db21 100644
--- a/tests/single_thread_executor_test.cpp
+++ b/tests/single_thread_executor_test.cpp
@@ -308,6 +308,65 @@ static bool test_post_shutdown_rejection() {
     return counter.load() == 1;
 }
 
+static bool test_wait_drains_after_shutdown_start() {
+    SingleThreadExecutor ex;
+    std::atomic<int> counter{0};
+    std::atomic<bool> wait_done{false};
+
+    std::mutex gate_mutex;
+    std::condition_variable gate_cv;
+    bool first_task_started = false;
+    bool gate_open = false;
+
+    ex.add_task([&]() {
+        std::unique_lock<std::mutex> lk(gate_mutex);
+        first_task_started = true;
+        gate_cv.notify_all();
+        gate_cv.wait(lk, [&]() { return gate_open; });
+        counter.fetch_add(1, std::memory_order_relaxed);
+    });
+    ex.add_task([&]() {
+        counter.fetch_add(1, std::memory_order_relaxed);
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(gate_mutex);
+        if (!gate_cv.wait_for(lk, std::chrono::seconds(1), [&]() {
+                return first_task_started;
+            })) {
+            gate_open = true;
+            gate_cv.notify_all();
+            return false;
+        }
+    }
+
+    std::thread waiter([&]() {
+        ex.wait();
+        wait_done.store(true, std::memory_order_release);
+    });
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    std::thread stopper([&]() {
+        ex.shutdown();
+    });
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    const bool returned_while_blocked = wait_done.load(std::memory_order_acquire);
+
+    {
+        std::lock_guard<std::mutex> lk(gate_mutex);
+        gate_open = true;
+    }
+    gate_cv.notify_all();
+
+    stopper.join();
+    waiter.join();
+
+    return !returned_while_blocked &&
+           wait_done.load(std::memory_order_acquire) &&
+           counter.load(std::memory_order_relaxed) == 2;
+}
+
 int main() {
     int passed = 0;
     int failed = 0;
@@ -328,6 +387,7 @@ int main() {
     run("exception_in_task", test_exception_in_task());
     run("concurrent_producers", test_concurrent_producers());
     run("post_shutdown_rejection", test_post_shutdown_rejection());
+    run("wait_drains_after_shutdown_start", test_wait_drains_after_shutdown_start());
 
     std::cout << "\n" << passed << " passed, " << failed << " failed" << std::endl;
     return failed > 0 ? 1 : 0;