fix(emitter): resolve idle-sleep and flush-timeout shutdown hangs

[snowplow-claude-review]

The C++ tracker's Emitter had three diagnosed shutdown-hang bugs, all fixed in this PR.

Bug 1 — idle-sleep lost-wakeup (stop() blocks up to 5 s)

The background thread's idle sleep used a non-predicated wait_for. Because stop() called notify_all() without holding m_db_select, the notification could fire between the pre-check and the wait_for, causing the thread to sleep out the full 5 s interval before checking the stop flag.

Fixed by introducing std::atomic<bool> m_stop_requested{false}. stop() sets m_stop_requested = true and then acquires/releases m_db_select before notify_all(), closing the race: by the time stop() releases the lock the daemon is either already in wait_for (gets the notification) or will see m_stop_requested = true at its next pre-check. start() resets the flag to false so the emitter can be restarted.

Bug 2 — flush() hangs indefinitely on unreachable collector

flush() waited on m_check_fin with no timeout. If the collector was unreachable (bad DNS, network down), flush() — and therefore stop() — would hang forever.

Fixed by replacing the infinite wait with a wait_for bounded by a configurable timeout:

if (m_flush_timeout_ms > 0) {
    this->m_check_fin.wait_for(locker_2,
        std::chrono::milliseconds(m_flush_timeout_ms),
        [this]{ return m_flush_done.load() || m_stop_requested.load(); });
} else {
    this->m_check_fin.wait(locker_2,
        [this]{ return m_flush_done.load() || m_stop_requested.load(); });
}
this->stop();

The timeout is exposed via EmitterConfiguration::set_flush_timeout_ms() (default 30,000 ms; 0 = no timeout), following the existing setter pattern (set_batch_size, set_byte_limit_post, etc.). If the timeout expires before the queue drains, flush() calls stop() and returns; undelivered events remain in the SQLite event store for the next session.

A dedicated std::atomic<bool> m_flush_done flag is set by the daemon when the queue empties (adjacent to the existing m_check_fin.notify_all()) and reset by flush() before waking the daemon, so the predicate only reflects state observed during this flush() call. stop() also notifies m_check_fin after joining the daemon thread so a concurrent external stop() correctly unblocks a waiting flush().

Bug 3 — retry sleep not interruptible

The per-batch retry delay used sleep_for, which stop() could not interrupt, causing stop to block for the full retry interval.

Fixed by replacing sleep_for with a pre-check on m_stop_requested plus a bare wait_for on m_check_db, so a concurrent stop() wakes the retry sleep immediately.

New tests

Four new tests added:

• stop() on idle emitter returns promptly — asserts elapsed < 2,000 ms (measured: ~0.08 ms).
• flush() with unreachable collector returns after flush timeout — 500 ms config, 503-always client, asserts returns < 3,000 ms and events remain in SQLite.
• stop() during retry sleep returns promptly — asserts < 2,000 ms (measured: ~0.09 ms).
• EmitterConfiguration flush timeout getter/setter — round-trip assertion.

Timing bounds are ≥ 2,000 ms to be reliable on loaded CI runners; actual latency is sub-millisecond.

What to review

• emitter.cpp: pre-check + bare wait_for in run() (idle sleep and retry path); stop() lock-handshake closing the notification race; flush() timeout logic, m_flush_done predicate, and stop() call after wait.
• emitter_configuration.hpp/.cpp: set_flush_timeout_ms / get_flush_timeout_ms setter/getter and shared_init() default.
• emitter.hpp: #include <atomic>, m_stop_requested{false}, m_flush_done{false}, m_flush_timeout_ms declarations.
• New tests for correctness of timing assertions and SQLite-retention assertion in the flush-timeout case.

---

Draft PR opened for review — please verify and run CI before merging.

[Snowplow CLA bot (snowplowcla)]

Thanks for your pull request. Is this your first contribution to a Snowplow open source project? Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

📝 Please visit https://docs.snowplowanalytics.com/docs/contributing/contributor-license-agreement/ to learn more and sign.

Once you've signed, please reply here (e.g. I signed it!) and we'll verify. Thanks.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[Matus Tomlein (matus-tomlein)]

Correctness gap: flush()'s m_check_fin wait has no predicate

In flush() the wait on m_check_fin is unpredicated, and the notifier (m_check_fin.notify_all() in run()) does not hold m_flush_fin:

unique_lock<mutex> locker_2(this->m_flush_fin);
if (m_flush_timeout_ms > 0) {
  this->m_check_fin.wait_for(locker_2, std::chrono::milliseconds(m_flush_timeout_ms));
} else {
  this->m_check_fin.wait(locker_2);
}

This leaves two windows open:

1. Spurious wakeup → flush returns early. A condition variable may wake without a matching notify. With no predicate, flush() can return before the queue has actually drained, then call stop() — silently regressing the flush contract (flush is expected to block until events are sent). This is a genuine correctness bug, not just a latency issue.

2. Lost wakeup → flush waits out the full timeout. If the daemon empties the queue and calls notify_all() in the window between flush()'s own m_check_db.notify_all() and entering wait_for, the notification is missed. The m_flush_timeout_ms > 0 bound hides this as "just slow," but with m_flush_timeout_ms == 0 (documented as "wait indefinitely") it reintroduces the original Issue What's up with this?  #2 indefinite hang, since the untimed wait() is also unpredicated and can never observe stop.

Suggested fix: gate both waits on a predicate tied to real drained/stop state — e.g. an atomic flag the daemon sets when the queue is empty (or when stopping), checked inside the wait:

this->m_check_fin.wait_for(locker_2, std::chrono::milliseconds(m_flush_timeout_ms),
    [this] { return m_flush_done.load() || m_stop_requested.load(); });

and set that flag under/adjacent to the same notify in run(). This closes the spurious-wakeup regression and makes even the m_flush_timeout_ms == 0 path interruptible by stop().

[Matus Tomlein (matus-tomlein)]

Re-review of latest push (2dc3579)

Both gaps from my earlier comment are addressed. Verified locally: builds clean, all 16 non-integration test cases pass (605 assertions), and the emitter case is stable across 8 consecutive runs.

Gap #1 (unpredicated m_check_fin wait) — resolved. Both wait branches in flush() now carry the predicate [this]{ return m_flush_done.load() || m_stop_requested.load(); }. flush() resets m_flush_done = false before notifying the daemon, and run() sets m_flush_done = true when the queue reads empty (after the batch has been sent/deleted), so the spurious-wakeup early return is closed and the flush contract holds — flush only returns once the queue has genuinely drained (or on timeout/stop).

Gap #2 (flush_timeout_ms == 0 re-hang) — resolved. The untimed wait() now shares the same predicate, so m_stop_requested breaks it, and stop() additionally calls m_check_fin.notify_all() after join to wake any external waiter. The 0 path is now stop-interruptible rather than an unconditional indefinite block.

Concurrency check. m_flush_done / m_stop_requested are std::atomic, set before their notify_all() and re-checked under the CV lock on each wakeup, so there's no lost-wakeup even though the daemon writes them without holding m_flush_fin. The m_db_select lock-handshake in stop() still covers both the idle and retry sleeps.

Minor (non-blocking), unchanged from before:

• New scenarios are SECTIONs inside the existing emitter TEST_CASE, so they don't show in --list-tests and can't be run in isolation.
• Timing assertions (< 2000 / < 3000 ms) remain wall-clock based; generous margins vs. sub-ms actual, but worth keeping an eye on given this repo's CI timing-flakiness history.
• stop() is still not interruptible during an in-flight blocking do_send() HTTP call — pre-existing and out of scope, but shutdown latency ultimately depends on the platform HTTP client's connect/read timeouts.

LGTM to merge once CI is green.