fix(logging): harden async logger shutdown

docs/TaskExecutor.md

@@ -75,19 +75,23 @@ end users through `LOGIT_GET_DROPPED_TASKS()`.
 application.
 
 1. `m_resizing` is set to `true` with release semantics.
-2. `wait()` drains the queue and ensures `m_active_tasks == 0`.
-3. The worker is stopped by setting `m_stop_flag`, notifying sleepers, and
+2. Producers that already entered `add_task()` are allowed to finish or the
+   resize is abandoned after the bounded resize deadline.
+3. `wait()` drains the queue and ensures `m_active_tasks == 0`.
+4. The worker is stopped by setting `m_stop_flag`, notifying sleepers, and
    joining the thread so it no longer touches `m_mpsc_queue`.
-4. In a single thread the ring is rebuilt with the new capacity. The resize
+5. In a single thread the ring is rebuilt with the new capacity. The resize
    keeps `m_dropped_tasks` intact but resets `m_active_tasks` to 0 because the
    queue is empty.
-5. The worker thread is restarted and the stop flag cleared.
-6. `m_resizing` flips back to `false` and `m_resize_cv.notify_all()` wakes
+6. The worker thread is restarted and the stop flag cleared.
+7. `m_resizing` flips back to `false` and `m_resize_cv.notify_all()` wakes
    producers that parked at the start of `add_task()`.
 
-While the resize is in progress, producers briefly wait on `m_resize_cv`. No
-accepted tasks are lost, and the consumer thread never observes partially
-initialised ring buffers.
+While the resize is in progress, new producers briefly wait on `m_resize_cv`.
+No accepted tasks are lost, and the consumer thread never observes partially
+initialised ring buffers. Calling `set_max_queue_size()` or
+`set_queue_policy()` after shutdown is a no-op; these calls must not restart or
+mutate the stopped singleton worker.
 
 ## 4. Ordering and completion guarantees
 
@@ -96,7 +100,9 @@ initialised ring buffers.
 * 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.
+  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.
 * `shutdown()` blocks until the worker thread terminates. It is safe to call
   multiple times.
 
@@ -198,15 +204,16 @@ const auto lost = LOGIT_GET_DROPPED_TASKS();
 * All public methods on non-Emscripten builds are thread-safe. Producers may
   call `add_task()` concurrently with `set_max_queue_size()` and
   `set_queue_policy()`.
-* The hot-resize barrier uses `m_resizing` and `m_resize_cv` so producers never
-  touch a ring buffer that is being rebuilt. This eliminates the data races that
-  TSAN previously reported on `try_pop()` vs. buffer assignment. The barrier
-  only drops once the worker thread fully stops and the queue drains; if a sink
-  blocks the worker or `QueuePolicy::Block` keeps `m_active_tasks` above the
-  limit for more than one second, `set_max_queue_size()` abandons the hot
-  resize, clears `m_resizing`, and leaves the existing ring untouched so
-  producers cannot wait indefinitely. Non-MPSC builds perform the resize as an
-  atomic update of `m_max_queue_size`, so they are not subject to this stall.
+* The hot-resize barrier uses `m_resizing`, `m_resize_cv`, and an active-producer
+  counter so producers never touch a ring buffer that is being rebuilt. This
+  eliminates the data races that TSAN previously reported on `try_pop()` vs.
+  buffer assignment. The barrier only proceeds once producers have paused, the
+  worker thread fully stops, and the queue drains; if a sink blocks the worker
+  or `QueuePolicy::Block` prevents producers from reaching the pause point for
+  more than one second, `set_max_queue_size()` abandons the hot resize, clears
+  `m_resizing`, and leaves the existing ring untouched so producers cannot wait
+  indefinitely. Non-MPSC builds perform the resize as an atomic update of
+  `m_max_queue_size`, so they are not subject to this stall.
 * Non-MPSC builds rely solely on mutexes and had no known data races.
 * The Emscripten path is single-threaded and should not be used concurrently.
 

include/logit_cpp/logit/Logger.hpp

@@ -59,14 +59,15 @@ namespace logit {
                 std::unique_ptr<ILogger> logger,
                 std::unique_ptr<ILogFormatter> formatter,
                 bool single_mode = false) {
-            if (m_shutdown) return;
+            if (m_shutdown.load(std::memory_order_acquire)) return;
             auto strategy = std::make_shared<LoggerStrategy>();
             strategy->logger = std::move(logger);
             strategy->formatter = std::move(formatter);
             strategy->single_mode = single_mode;
             strategy->enabled = true;
 
             LoggerWriteLock lock(m_loggers_mx);
+            if (m_shutdown.load(std::memory_order_acquire)) return;
             m_loggers.push_back(std::move(strategy));
         }
 
@@ -157,7 +158,7 @@ namespace logit {
         /// the formatted message to the logger.
         /// \param record Log record to be logged.
         void log(const LogRecord& record) {
-            if (m_shutdown) return;
+            if (m_shutdown.load(std::memory_order_acquire)) return;
 
             const bool targeted = record.logger_index >= 0;
 
@@ -178,6 +179,7 @@ namespace logit {
                 auto& strategy = snapshot[0];
 
                 std::lock_guard<std::mutex> exec_lock(strategy->exec_mx);
+                if (m_shutdown.load(std::memory_order_acquire)) return;
                 if (!strategy->enabled) return;
                 if (!record.raw_mode &&
                     static_cast<int>(record.log_level) < static_cast<int>(strategy->logger->get_log_level())) return;
@@ -189,6 +191,7 @@ namespace logit {
                 if (!strategy) continue;
 
                 std::lock_guard<std::mutex> exec_lock(strategy->exec_mx);
+                if (m_shutdown.load(std::memory_order_acquire)) return;
                 if (strategy->single_mode) continue;
                 if (!strategy->enabled) continue;
                 if (!record.raw_mode &&
@@ -400,7 +403,7 @@ namespace logit {
         /// Disables further logging, waits for asynchronous tasks to complete,
         /// and shuts down TaskExecutor.
         void shutdown() {
-            if (m_shutdown.exchange(true)) return;
+            if (m_shutdown.exchange(true, std::memory_order_acq_rel)) return;
 
             const auto snapshot = get_all_strategy_snapshots();
             for (const auto& strategy : snapshot) {

include/logit_cpp/logit/detail/TaskExecutor.hpp

@@ -211,21 +211,14 @@ namespace logit { namespace detail {
             lock.unlock();
             m_queue_condition.notify_one();
 #        else
-            // Hot-resize barrier: wait until the ring rebuild is finished.
-            if (m_resizing.load(std::memory_order_acquire)) {
-                std::unique_lock<std::mutex> lk(m_cv_mutex);
-                m_resize_cv.wait(lk, [this]{ return !m_resizing.load(std::memory_order_acquire); });
-            }
-
-            if (m_stop_flag.load(std::memory_order_acquire)) {
-                return;
-            }
+            enter_producer_();
 
             std::function<void()> local_task = std::move(task);
+            bool done = false;
 
-            for (;;) {
+            while (!done) {
                 if (m_stop_flag.load(std::memory_order_acquire)) {
-                    return;
+                    break;
                 }
 
                 const auto policy = m_overflow_policy.load(std::memory_order_relaxed);
@@ -243,20 +236,22 @@ namespace logit { namespace detail {
                 // Try to push into the ring buffer.
                 if (m_mpsc_queue.try_push(local_task)) {
                     m_cv.notify_one(); // wake the worker
-                    return;
+                    break;
                 }
 
                 // Apply the configured overflow policy when the ring is full.
                 switch (policy) {
                     case QueuePolicy::DropNewest:
                         m_dropped_tasks.fetch_add(1, std::memory_order_relaxed);
-                        return;
+                        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);
-                        return;
+                        done = true;
+                        break;
 
                     case QueuePolicy::Block: {
                         std::unique_lock<std::mutex> lk(m_cv_mutex);
@@ -265,6 +260,7 @@ namespace logit { namespace detail {
                     }
                 }
             }
+            leave_producer_();
 #        endif
         }
 
@@ -289,6 +285,7 @@ namespace logit { namespace detail {
 
         /// \brief Stop the worker thread and drain outstanding tasks.
         void shutdown() {
+            std::lock_guard<std::mutex> lifecycle_lock(m_lifecycle_mutex);
 #        ifndef LOGIT_USE_MPSC_RING
             std::unique_lock<std::mutex> lock(m_queue_mutex);
             m_stop_flag.store(true, std::memory_order_release);
@@ -314,17 +311,26 @@ namespace logit { namespace detail {
         /// \details On MPSC builds this performs the "hot" resize described in
         /// docs/TaskExecutor.md.
         void set_max_queue_size(std::size_t size) {
+            std::lock_guard<std::mutex> lifecycle_lock(m_lifecycle_mutex);
+            if (m_stop_flag.load(std::memory_order_acquire)) return;
 #       ifdef LOGIT_USE_MPSC_RING
             // Tell producers to pause before any wait()/stop conditions run.
             m_resizing.store(true, std::memory_order_release);
+            const auto deadline = std::chrono::steady_clock::now() +
+                                  std::chrono::seconds(1);
+            // Existing producers may be blocked by backpressure, so do not
+            // wait forever for the resize barrier.
+            if (!wait_until_producers_paused_(deadline)) {
+                m_resizing.store(false, std::memory_order_release);
+                m_resize_cv.notify_all();
+                return;
+            }
 
             // Drain the queue completely, but do not wait forever if the worker
             // is stalled (e.g., blocked sink or backpressure keeping
             // m_active_tasks > 0). If we fail to drain before the deadline,
             // abort the resize and re-open the barrier so producers can
             // continue.
-            const auto deadline = std::chrono::steady_clock::now() +
-                                  std::chrono::seconds(1);
             if (!wait_until_idle_(deadline)) {
                 m_resizing.store(false, std::memory_order_release);
                 m_resize_cv.notify_all();
@@ -334,7 +340,7 @@ namespace logit { namespace detail {
             // Stop the worker so it cannot touch m_mpsc_queue during the resize.
             std::unique_lock<std::mutex> lk(m_queue_mutex);
             m_stop_flag.store(true, std::memory_order_relaxed);
-                lk.unlock();
+            lk.unlock();
 
             m_cv.notify_all();
             m_queue_condition.notify_all();
@@ -343,15 +349,15 @@ namespace logit { namespace detail {
             }
 
             // Reinitialise the parameters and the ring on a single thread.
-                lk.lock();
-                m_max_queue_size = size;
-                const std::size_t cap =
-                        (m_max_queue_size == 0 ? m_default_ring_cap : m_max_queue_size);
-                m_mpsc_queue = MpscRingAny<std::function<void()>>(cap);
-                // Reset counters (except drops) because the queue is empty.
-                m_active_tasks.store(0, std::memory_order_relaxed);
-                // Keep m_dropped_tasks untouched — tests manage it via macros.
-                lk.unlock();
+            lk.lock();
+            m_max_queue_size = size;
+            const std::size_t cap =
+                    (m_max_queue_size == 0 ? m_default_ring_cap : m_max_queue_size);
+            m_mpsc_queue = MpscRingAny<std::function<void()>>(cap);
+            // Reset counters (except drops) because the queue is empty.
+            m_active_tasks.store(0, std::memory_order_relaxed);
+            // Keep m_dropped_tasks untouched; tests manage it via macros.
+            lk.unlock();
 
             // Clear the stop flag and restart the worker thread.
             m_stop_flag.store(false, std::memory_order_relaxed);
@@ -368,6 +374,8 @@ namespace logit { namespace detail {
 
         /// \brief Change the overflow policy for newly submitted tasks.
         void set_queue_policy(QueuePolicy policy) {
+            std::lock_guard<std::mutex> lifecycle_lock(m_lifecycle_mutex);
+            if (m_stop_flag.load(std::memory_order_acquire)) return;
             std::lock_guard<std::mutex> lock(m_queue_mutex);
             m_overflow_policy.store(policy, std::memory_order_relaxed);
         }
@@ -382,6 +390,7 @@ namespace logit { namespace detail {
         }
 
     private:
+        mutable std::mutex m_lifecycle_mutex;      ///< Serializes shutdown with lifecycle-changing operations.
     #ifndef LOGIT_USE_MPSC_RING
         std::deque<std::function<void()>> m_tasks_queue;
         mutable std::mutex m_queue_mutex;
@@ -401,6 +410,7 @@ namespace logit { namespace detail {
 
         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.
 
         std::thread m_worker_thread;
         std::atomic<bool> m_stop_flag;
@@ -445,9 +455,17 @@ namespace logit { namespace detail {
                 std::function<void()> task;
 
                 int budget = LOGIT_TASK_EXECUTOR_DRAIN_BUDGET;
-                while (budget-- && m_mpsc_queue.try_pop(task)) {
-                    drained_any = true;
+                while (budget--) {
+                    // Count the pop attempt as active so wait() cannot observe
+                    // an empty ring between try_pop() freeing a cell and the
+                    // dequeued task starting execution.
                     m_active_tasks.fetch_add(1, std::memory_order_relaxed);
+                    if (!m_mpsc_queue.try_pop(task)) {
+                        m_active_tasks.fetch_sub(1, std::memory_order_relaxed);
+                        break;
+                    }
+
+                    drained_any = true;
 
                     task();
 
@@ -488,6 +506,38 @@ namespace logit { namespace detail {
                         m_stop_flag.load(std::memory_order_acquire));
             });
         }
+
+        void enter_producer_() {
+            for (;;) {
+                if (m_resizing.load(std::memory_order_acquire)) {
+                    std::unique_lock<std::mutex> lk(m_cv_mutex);
+                    m_resize_cv.wait(lk, [this]() {
+                        return !m_resizing.load(std::memory_order_acquire);
+                    });
+                    continue;
+                }
+
+                m_active_producers.fetch_add(1, std::memory_order_acq_rel);
+                if (!m_resizing.load(std::memory_order_acquire)) {
+                    return;
+                }
+                leave_producer_();
+            }
+        }
+
+        void leave_producer_() {
+            if (m_active_producers.fetch_sub(1, std::memory_order_acq_rel) == 1 &&
+                m_resizing.load(std::memory_order_acquire)) {
+                m_resize_cv.notify_all();
+            }
+        }
+
+        bool wait_until_producers_paused_(std::chrono::steady_clock::time_point deadline) {
+            std::unique_lock<std::mutex> lk(m_cv_mutex);
+            return m_resize_cv.wait_until(lk, deadline, [this]() {
+                return m_active_producers.load(std::memory_order_acquire) == 0;
+            });
+        }
     #endif
 
         TaskExecutor()
@@ -499,6 +549,7 @@ namespace logit { namespace detail {
               m_active_tasks(0)
     #else
             : m_resizing(false),
+              m_active_producers(0),
               m_worker_thread(),
               m_stop_flag(false),
               m_max_queue_size(0),

include/logit_cpp/logit/loggers/ConsoleLogger.hpp

@@ -121,6 +121,7 @@ namespace logit {
             std::shared_ptr<detail::SingleThreadExecutor> old_executor;
             {
                 std::unique_lock<std::mutex> lock(m_mutex);
+                if (m_shutdown.load(std::memory_order_acquire)) return;
                 wait_for_pending_enqueues(lock);
                 m_config = config;
                 if (m_config.async && m_config.use_dedicated_executor) {
@@ -154,9 +155,11 @@ namespace logit {
         /// \param record The log record containing log information.
         /// \param message The formatted log message.
         void log(const LogRecord& record, const std::string& message) override {
-            m_last_log_ts = record.timestamp_ms;
+            if (m_shutdown.load(std::memory_order_acquire)) return;
 #ifdef __EMSCRIPTEN__
             std::unique_lock<std::mutex> lock(m_mutex);
+            if (m_shutdown.load(std::memory_order_acquire)) return;
+            m_last_log_ts = record.timestamp_ms;
             const int lvl = static_cast<int>(record.log_level);
             std::shared_ptr<detail::SingleThreadExecutor> executor = m_executor;
             if (!m_config.async) {
@@ -194,6 +197,8 @@ namespace logit {
             return;
 #else
             std::unique_lock<std::mutex> lock(m_mutex);
+            if (m_shutdown.load(std::memory_order_acquire)) return;
+            m_last_log_ts = record.timestamp_ms;
             std::shared_ptr<detail::SingleThreadExecutor> executor = m_executor;
             if (!m_config.async) {
 #               if defined(_WIN32)
@@ -306,6 +311,7 @@ namespace logit {
 
         /// \brief Stops the dedicated executor after draining pending messages.
         void shutdown() override {
+            if (m_shutdown.exchange(true, std::memory_order_acq_rel)) return;
             std::shared_ptr<detail::SingleThreadExecutor> executor;
             bool async = false;
             {
@@ -327,6 +333,7 @@ namespace logit {
         Config             m_config;    ///< Configuration for the console logger.
         std::atomic<int64_t> m_last_log_ts = ATOMIC_VAR_INIT(0);
         std::atomic<int>    m_log_level = ATOMIC_VAR_INIT(static_cast<int>(LogLevel::LOG_LVL_TRACE));
+        std::atomic<bool> m_shutdown = ATOMIC_VAR_INIT(false);
         std::shared_ptr<detail::SingleThreadExecutor> m_executor;
         std::condition_variable m_enqueue_cv;
         std::size_t m_pending_enqueues = 0;