feat: enhance race condition tests to verify producer resilience after consumer crash

Reviewed-on: #8

src/app/main.cxx

@@ -1,4 +1,4 @@
-^// main.cxx
+// main.cxx
 // SPDX-License-Identifier: GPL-3.0-or-later
 // Author: Unai Blazquez <unaibg2000@gmail.com>
 

tests/CMakeLists.txt

@@ -73,3 +73,18 @@ target_link_libraries(test_main_window
 )
 
 add_test(NAME test_main_window COMMAND test_main_window)
+
+add_executable(test_race_conditions
+    test_race_conditions.cxx
+)
+
+target_link_libraries(test_race_conditions
+    PRIVATE
+        core
+        gtest
+        gtest_main
+        Qt5::Core
+        Qt5::Test
+)
+
+add_test(NAME test_race_conditions COMMAND test_race_conditions)

tests/test_consumer.cxx

@@ -1,7 +1,11 @@
 #include <gtest/gtest.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <unistd.h>
 
 #include <QCoreApplication>
 #include <QSignalSpy>
+#include <cstring>
 
 #include "Consumer.hpp"
 #include "UnixIpcBridge.hpp"
@@ -118,3 +122,94 @@ TEST(ConsumerThreadTest, StopsCleanlyWhenNeverStarted)
   // stop() on a consumer that was never started must not crash
   consumer.stop();
 }
+
+// ---------------------------------------------------------------------------
+// Requirement 2: Consumer receiving corrupted data (non-numeric)
+// ---------------------------------------------------------------------------
+
+/// Helper: raw-connect to a UNIX socket and send arbitrary bytes.
+static void send_raw_bytes(const std::string& path, const void* data,
+                           size_t len)
+{
+  int fd = socket(AF_UNIX, SOCK_STREAM, 0);
+  ASSERT_GE(fd, 0);
+
+  struct sockaddr_un addr = {};
+  addr.sun_family = AF_UNIX;
+  std::strncpy(addr.sun_path, path.c_str(), sizeof(addr.sun_path) - 1);
+
+  ASSERT_EQ(
+      connect(fd, reinterpret_cast<sockaddr*>(&addr), sizeof(addr)), 0);
+
+  if (len > 0)
+  {
+    ::send(fd, data, len, 0);
+  }
+  close(fd);
+}
+
+TEST(ConsumerThreadTest, DropsCorruptedShortMessage)
+{
+  const std::string sock = "/tmp/test_ct_corrupt_short.sock";
+
+  ConsumerThread consumer(sock);
+  QSignalSpy spy(&consumer, &ConsumerThread::valueReceived);
+  consumer.start();
+
+  // Send only 2 bytes instead of sizeof(int)==4 — corrupted / partial message
+  uint16_t garbage = 0xBEEF;
+  send_raw_bytes(sock, &garbage, sizeof(garbage));
+
+  // Give the consumer time to process (or not)
+  spy.wait(500);
+  consumer.stop();
+
+  // No signal should have been emitted
+  EXPECT_EQ(spy.count(), 0);
+}
+
+TEST(ConsumerThreadTest, DropsEmptyConnection)
+{
+  const std::string sock = "/tmp/test_ct_corrupt_empty.sock";
+
+  ConsumerThread consumer(sock);
+  QSignalSpy spy(&consumer, &ConsumerThread::valueReceived);
+  consumer.start();
+
+  // Connect and immediately close — zero bytes sent
+  send_raw_bytes(sock, nullptr, 0);
+
+  spy.wait(500);
+  consumer.stop();
+
+  EXPECT_EQ(spy.count(), 0);
+}
+
+TEST(ConsumerThreadTest, SurvivesCorruptedThenReceivesValid)
+{
+  const std::string sock = "/tmp/test_ct_corrupt_then_valid.sock";
+
+  ConsumerThread consumer(sock);
+  QSignalSpy spy(&consumer, &ConsumerThread::valueReceived);
+  consumer.start();
+
+  // First: send corrupted (1 byte)
+  uint8_t one_byte = 0xFF;
+  send_raw_bytes(sock, &one_byte, sizeof(one_byte));
+  std::this_thread::sleep_for(std::chrono::milliseconds(50));
+
+  // Then: send a valid int via the normal bridge
+  UnixIpcBridge bridge(sock);
+  bridge.send(777);
+
+  // Wait for the valid signal
+  for (int attempt = 0; spy.count() < 1 && attempt < 20; ++attempt)
+  {
+    spy.wait(100);
+  }
+  consumer.stop();
+
+  // The corrupted message must have been dropped, valid one received
+  ASSERT_EQ(spy.count(), 1);
+  EXPECT_EQ(spy.at(0).at(0).toInt(), 777);
+}

tests/test_race_conditions.cxx

@@ -0,0 +1,177 @@
+// test_race_conditions.cxx
+// SPDX-License-Identifier: GPL-3.0-or-later
+// Author: Unai Blazquez <unaibg2000@gmail.com>
+
+#include <gtest/gtest.h>
+
+#include <QCoreApplication>
+#include <QSignalSpy>
+#include <atomic>
+#include <chrono>
+#include <fstream>
+#include <iostream>
+#include <stdexcept>
+#include <thread>
+#include <vector>
+
+#include "Consumer.hpp"
+#include "Producer.hpp"
+#include "UnixIpcBridge.hpp"
+
+static int argc_ = 0;
+static QCoreApplication app_(argc_, nullptr);
+
+
+TEST(RaceConditionTest, RepeatedStartStopWhileProducerSends)
+{
+  const std::string sock = "/tmp/test_race.sock";
+  constexpr int kCycles = 20;
+
+  // Watchdog: if the test takes longer than 15s, declare deadlock.
+  std::atomic<bool> test_done{false};
+  std::thread watchdog([&test_done]() {
+    for (int i = 0; i < 150 && !test_done.load(); ++i)
+    {
+      std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    }
+    if (!test_done.load())
+    {
+      std::cerr
+          << "DEADLOCK DETECTED: RepeatedStartStopWhileProducerSends timed out"
+          << std::endl;
+      std::abort();
+    }
+  });
+
+  std::atomic<bool> producer_running{true};
+  std::thread producer([&]() {
+    while (producer_running.load())
+    {
+      try
+      {
+        UnixIpcBridge bridge(sock);
+        bridge.send(42);
+      }
+      catch (const std::runtime_error&)
+      {
+        // Expected: consumer socket not ready or just torn down.
+      }
+      std::this_thread::sleep_for(std::chrono::milliseconds(5));
+    }
+  });
+
+  for (int i = 0; i < kCycles; ++i)
+  {
+    ConsumerThread consumer(sock);
+    consumer.start();
+
+    // Let it run briefly so the producer can connect during some cycles.
+    std::this_thread::sleep_for(std::chrono::milliseconds(10 + (i % 5) * 5));
+
+    // stop() must return without deadlock every single time.
+    consumer.stop();
+  }
+
+  producer_running.store(false);
+  producer.join();
+
+  test_done.store(true);
+  watchdog.join();
+
+  // If we reach here, no deadlock across kCycles start/stop cycles.
+  SUCCEED();
+}
+
+TEST(RaceConditionTest, ProducerSurvivesConsumerCrash)
+{
+  const std::string sock = "/tmp/test_crash.sock";
+  const std::string sysfs = "./fake_sysfs_race";
+
+  // Prepare sysfs file so the producer is in Enabled state.
+  { std::ofstream(sysfs) << "1\n"; }
+
+  // Track what the producer sends.
+  std::vector<int> sent_values;
+  std::mutex sent_mutex;
+  std::vector<std::string> logs;
+  std::mutex log_mutex;
+
+  auto make_safe_send = [&](const std::string& path) {
+    return [&, path](int value) {
+      try
+      {
+        UnixIpcBridge bridge(path);
+        bridge.send(value);
+        std::lock_guard<std::mutex> lk(sent_mutex);
+        sent_values.push_back(value);
+      }
+      catch (const std::runtime_error&)
+      {
+        // Consumer is down — expected during the "crash" window.
+      }
+    };
+  };
+
+  Producer producer(
+      sysfs, make_safe_send(sock), []() { return 123; },
+      [&](const std::string& msg) {
+        std::lock_guard<std::mutex> lk(log_mutex);
+        logs.push_back(msg);
+      },
+      [](std::chrono::milliseconds) {
+        // Use a short sleep so the test runs fast.
+        std::this_thread::sleep_for(std::chrono::milliseconds(20));
+      });
+
+  // Phase 1: start consumer, start producer, let a few values flow.
+  {
+    ConsumerThread consumer(sock);
+    QSignalSpy spy(&consumer, &ConsumerThread::valueReceived);
+    consumer.start();
+    producer.start();
+
+    // Wait for at least 2 values to arrive.
+    for (int attempt = 0; spy.count() < 2 && attempt < 50; ++attempt)
+    {
+      spy.wait(100);
+    }
+    ASSERT_GE(spy.count(), 2) << "Phase 1: producer should have delivered values";
+
+    // "Crash" the consumer: stop + destroy.
+    consumer.stop();
+  }
+
+  // Phase 2: producer is still running with no consumer (sends will fail).
+  std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+  // Phase 3: bring up a fresh consumer. Producer should resume delivering.
+  {
+    ConsumerThread consumer2(sock);
+    QSignalSpy spy2(&consumer2, &ConsumerThread::valueReceived);
+    consumer2.start();
+
+    for (int attempt = 0; spy2.count() < 2 && attempt < 50; ++attempt)
+    {
+      spy2.wait(100);
+    }
+
+    consumer2.stop();
+
+    ASSERT_GE(spy2.count(), 2)
+        << "Phase 3: producer must deliver to a new consumer after crash";
+
+    // Values received by the second consumer should all be 123.
+    for (int i = 0; i < spy2.count(); ++i)
+    {
+      EXPECT_EQ(spy2.at(i).at(0).toInt(), 123);
+    }
+  }
+
+  producer.stop();
+
+  // Producer logged throughout all three phases.
+  {
+    std::lock_guard<std::mutex> lk(log_mutex);
+    EXPECT_GE(logs.size(), 3u) << "Producer should have kept logging";
+  }
+}