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

include/snowplow/configuration/emitter_configuration.cpp

@@ -35,6 +35,7 @@ void EmitterConfiguration::shared_init() {
   m_batch_size = SNOWPLOW_EMITTER_DEFAULT_BATCH_SIZE;
   m_byte_limit_get = SNOWPLOW_EMITTER_DEFAULT_BYTE_LIMIT_GET;
   m_byte_limit_post = SNOWPLOW_EMITTER_DEFAULT_BYTE_LIMIT_POST;
+  m_flush_timeout_ms = 30000;
 }
 
 void EmitterConfiguration::set_event_store(shared_ptr<EventStore> event_store) {

include/snowplow/configuration/emitter_configuration.hpp

@@ -88,14 +88,23 @@ class EmitterConfiguration {
 
   /**
    * @brief Set a custom retry rule for when the HTTP status code is received in emit response from Collector.
-   * 
+   *
    * This overrides default behavior for HTTP status codes greater than 300.
-   * 
+   *
    * @param http_status_code HTTP status code
    * @param retry Whether events should be retried or not
    */
   void set_custom_retry_for_status_code(int http_status_code, bool retry);
 
+  /**
+   * @brief Set the maximum time flush() will wait for the event queue to drain before stopping.
+   *
+   * Pass 0 for no timeout (waits indefinitely). Undelivered events remain in SQLite for the next session.
+   *
+   * @param flush_timeout_ms Maximum wait time in milliseconds (default: 30000).
+   */
+  void set_flush_timeout_ms(int flush_timeout_ms) { m_flush_timeout_ms = flush_timeout_ms; }
+
   /**
    * @brief Get the event store.
    * 
@@ -149,17 +158,25 @@ class EmitterConfiguration {
 
   /**
    * @brief Get the custom retry rule settings for HTTP status codes.
-   * 
+   *
    * @return map<int, bool> Map of status code –> retry or not boolean.
    */
   map<int, bool> get_custom_retry_for_status_codes() const { return m_custom_retry_for_status_codes; }
 
+  /**
+   * @brief Get the flush timeout in milliseconds.
+   *
+   * @return int Maximum time flush() waits for the queue to drain (0 = no timeout).
+   */
+  int get_flush_timeout_ms() const { return m_flush_timeout_ms; }
+
 private:
   void shared_init();
 
   int m_batch_size;
   int m_byte_limit_post;
   int m_byte_limit_get;
+  int m_flush_timeout_ms;
   shared_ptr<EventStore> m_event_store;
   EmitterCallback m_callback;
   EmitStatus m_callback_emit_status;

include/snowplow/emitter/emitter.cpp

@@ -25,7 +25,6 @@ using std::to_string;
 using std::transform;
 using std::equal;
 using std::future;
-using std::this_thread::sleep_for;
 
 const int post_wrapper_bytes = 88; // "schema":"iglu:com.snowplowanalytics.snowplow/payload_data/jsonschema/1-0-4","data":[]
 const int post_stm_bytes = 22;     // "stm":"1443452851000"
@@ -62,6 +61,7 @@ Emitter::Emitter(NetworkConfiguration &network_config, const EmitterConfiguratio
   m_callback = emitter_config.get_request_callback();
   m_callback_emit_status = emitter_config.get_request_callback_emit_status();
   m_custom_retry_for_status_codes = emitter_config.get_custom_retry_for_status_codes();
+  m_flush_timeout_ms = emitter_config.get_flush_timeout_ms();
 }
 
 Emitter::Emitter(shared_ptr<EventStore> event_store, const string &uri, Method method, Protocol protocol, int batch_size,
@@ -86,6 +86,7 @@ Emitter::Emitter(shared_ptr<EventStore> event_store, const string &uri, Method m
   }
 
   this->m_running = false;
+  this->m_flush_timeout_ms = 30000;
   this->m_method = method;
   this->m_batch_size = batch_size;
   this->m_byte_limit_post = byte_limit_post;
@@ -109,6 +110,8 @@ void Emitter::start() {
   if (this->m_running) {
     return; // refuse to start more than once
   }
+  this->m_stop_requested = false;
+  this->m_flush_done = false;
   this->m_running = true;
   this->m_daemon_thread = thread(&Emitter::run, this);
 }
@@ -117,10 +120,18 @@ void Emitter::stop() {
   unique_lock<mutex> locker(this->m_run_check);
   if (this->m_running == true) {
     this->m_running = false;
+    this->m_stop_requested = true;
     locker.unlock();
 
+    // Close the lost-wakeup race: by acquiring and releasing m_db_select before
+    // notify_all, we guarantee the daemon is either already in wait_for (will
+    // receive the notify) or will see m_stop_requested=true at its next pre-check.
+    { unique_lock<mutex> db_locker(this->m_db_select); }
     this->m_check_db.notify_all();
     this->m_daemon_thread.join();
+
+    // Unblock flush() if it is waiting on m_check_fin (e.g. stop() called externally)
+    this->m_check_fin.notify_all();
   }
 }
 
@@ -136,10 +147,19 @@ void Emitter::flush() {
   }
   locker_1.unlock();
 
+  // Reset flush_done before waking the daemon so that the predicate below
+  // reflects only queue state observed after this flush() call.
+  m_flush_done = false;
   this->m_check_db.notify_all();
 
   unique_lock<mutex> locker_2(this->m_flush_fin);
-  this->m_check_fin.wait(locker_2);
+  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(); });
+  }
   locker_2.unlock();
 
   this->stop();
@@ -188,17 +208,25 @@ void Emitter::run() {
         m_retry_delay.wont_retry_emit();
       }
 
-      // sleep for the retry delay if there is one
+      // sleep for the retry delay if there is one — interruptible by stop()
       auto retry_delay = m_retry_delay.get();
       if (retry_delay.count() > 0) {
-        sleep_for(retry_delay);
+        unique_lock<mutex> retry_locker(m_db_select);
+        if (!m_stop_requested.load()) {
+          m_check_db.wait_for(retry_locker, retry_delay);
+        }
       }
     } else {
+      // Queue is empty: signal flush() waiters
+      m_flush_done = true;
       m_check_fin.notify_all();
 
-      // if there are no events to send, sleep for a while
+      // Idle sleep — pre-check m_stop_requested so stop() calling notify_all between
+      // here and wait_for is guaranteed visible via the m_db_select lock-handshake in stop()
       unique_lock<mutex> locker(m_db_select);
-      m_check_db.wait_for(locker, std::chrono::seconds(5));
+      if (!m_stop_requested.load()) {
+        m_check_db.wait_for(locker, std::chrono::seconds(5));
+      }
     }
   } while (is_running());
 }

include/snowplow/emitter/emitter.hpp

@@ -16,6 +16,7 @@ See the Apache License Version 2.0 for the specific language governing permissio
 
 #include <string>
 #include <sstream>
+#include <atomic>
 #include <condition_variable>
 #include <future>
 #include <thread>
@@ -193,6 +194,9 @@ class Emitter {
   mutex m_db_select;
   mutex m_run_check;
   bool m_running;
+  std::atomic<bool> m_stop_requested{false};
+  std::atomic<bool> m_flush_done{false};
+  int m_flush_timeout_ms;
   EmitterCallback m_callback;
   EmitStatus m_callback_emit_status;
   map<int, bool> m_custom_retry_for_status_codes;

test/configuration/emitter_configuration_test.cpp

@@ -45,4 +45,12 @@ TEST_CASE("emitter configuration") {
     REQUIRE(raised_error);
     REQUIRE(emitter_config.get_custom_retry_for_status_codes().size() == 0);
   }
+
+  SECTION("flush timeout getter and setter") {
+    auto storage = std::make_shared<SqliteStorage>("test-emitter.db");
+    EmitterConfiguration emitter_config(storage);
+    REQUIRE(emitter_config.get_flush_timeout_ms() == 30000);
+    emitter_config.set_flush_timeout_ms(5000);
+    REQUIRE(emitter_config.get_flush_timeout_ms() == 5000);
+  }
 }

test/emitter/emitter_test.cpp

@@ -374,4 +374,69 @@ TEST_CASE("emitter") {
 
     emitter.stop();
   }
+
+  SECTION("stop() on idle emitter returns promptly") {
+    Emitter emitter(storage, "com.acme.collector", Method::POST, Protocol::HTTP, 500, 500, 500, unique_ptr<HttpClient>(new TestHttpClient()));
+    emitter.start();
+    auto t_start = std::chrono::high_resolution_clock::now();
+    emitter.stop();
+    auto t_end = std::chrono::high_resolution_clock::now();
+    double elapsed_ms = std::chrono::duration<double, std::milli>(t_end - t_start).count();
+    REQUIRE(elapsed_ms < 2000);
+  }
+
+  SECTION("flush() with unreachable collector returns after flush timeout") {
+    TestHttpClient::set_http_response_code(503); // always 503 (will retry indefinitely without timeout)
+
+    auto test_storage = std::make_shared<SqliteStorage>("test-emitter-flushtimeout.db");
+    NetworkConfiguration network_config("com.acme.unreachable.collector", POST);
+    network_config.set_http_client(unique_ptr<HttpClient>(new TestHttpClient()));
+    EmitterConfiguration emitter_config(test_storage);
+    emitter_config.set_flush_timeout_ms(500);
+
+    Emitter emitter(network_config, emitter_config);
+    emitter.start();
+
+    Payload payload;
+    payload.add("e", "pv");
+    emitter.add(payload);
+
+    auto t_start = std::chrono::high_resolution_clock::now();
+    emitter.flush();
+    auto t_end = std::chrono::high_resolution_clock::now();
+    double elapsed_ms = std::chrono::duration<double, std::milli>(t_end - t_start).count();
+    REQUIRE(elapsed_ms < 3000);
+
+    // Events remain in SQLite — not dropped on timeout
+    list<EventRow> remaining;
+    test_storage->get_all_event_rows(&remaining);
+    REQUIRE(remaining.size() > 0);
+
+    TestHttpClient::reset();
+    remove("test-emitter-flushtimeout.db");
+  }
+
+  SECTION("stop() during retry sleep returns promptly") {
+    auto test_storage = std::make_shared<SqliteStorage>("test-emitter-retrystop.db");
+    TestHttpClient::set_temporary_response_code(503, 20); // 20 failures to keep daemon retrying
+
+    Emitter emitter(test_storage, "com.acme.collector", Method::POST, Protocol::HTTP, 500, 500, 500, unique_ptr<HttpClient>(new TestHttpClient()));
+    emitter.start();
+
+    Payload payload;
+    payload.add("e", "pv");
+    emitter.add(payload);
+
+    // Give the daemon time to attempt the first send and enter the retry sleep
+    sleep_for(milliseconds(100));
+
+    auto t_start = std::chrono::high_resolution_clock::now();
+    emitter.stop();
+    auto t_end = std::chrono::high_resolution_clock::now();
+    double elapsed_ms = std::chrono::duration<double, std::milli>(t_end - t_start).count();
+    REQUIRE(elapsed_ms < 2000);
+
+    TestHttpClient::reset();
+    remove("test-emitter-retrystop.db");
+  }
 }