Synchronize Ethernet fixture wakeups with predicate updates

test/fixtures/rak4631_ethernet_runtime/test.cpp

@@ -264,7 +264,14 @@ static void finish_off() {
   assert(!ethernet_get_enabled() && !ethernet_is_running());
   assert(ethernet_task_handle == nullptr);
 }
-static void unblock(std::atomic<bool>& flag) { flag.store(false); library_changed.notify_all(); }
+static void unblock(std::atomic<bool>& flag) {
+  {
+    // Pair the predicate update with wait() so its notification cannot be lost.
+    std::lock_guard<std::mutex> lock(library_mutex);
+    flag.store(false);
+  }
+  library_changed.notify_all();
+}
 static std::string output() {
   std::lock_guard<std::mutex> lock(mock_output_mutex);
   return mock_ethernet_output;

test/test_rak4631_ethernet_runtime.py

@@ -9,6 +9,7 @@ import tempfile
 import unittest
 
 from test_serial_wifi_sessions import SANITIZER_FLAGS
+from test_radio_receive_contract import method
 
 ROOT = Path(__file__).resolve().parents[1]
 FIXTURE = ROOT / "test/fixtures/rak4631_ethernet_runtime"
@@ -46,6 +47,80 @@ SCENARIOS = (
 
 
 class Rak4631EthernetRuntimeTests(unittest.TestCase):
+    def test_unblock_holds_predicate_mutex_before_notifying(self):
+        compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
+        if compiler is None:
+            self.skipTest("a host C++17 compiler is required")
+        helper = method((FIXTURE / "test.cpp").read_text(), "static void unblock(")
+        guard = "std::lock_guard<std::mutex> lock(library_mutex);"
+        self.assertEqual(helper.count(guard), 1)
+        harness = r'''
+#include <atomic>
+#include <cassert>
+#include <chrono>
+#include <mutex>
+#include <thread>
+
+struct TrackedMutex {
+  std::mutex underlying;
+  std::atomic<bool> lock_attempted{false};
+  void lock() {
+    lock_attempted.store(true);
+    underlying.lock();
+  }
+  void unlock() { underlying.unlock(); }
+} library_mutex;
+struct Notification {
+  unsigned count = 0;
+  void notify_all() { ++count; }
+} library_changed;
+
+@UNBLOCK@
+
+int main() {
+  std::atomic<bool> blocked{true}, completed{false};
+  // Hold the waiter's mutex across its predicate-to-wait boundary. The
+  // notifier must attempt this mutex before it can change the predicate.
+  library_mutex.underlying.lock();
+  std::thread notifier([&] {
+    unblock(blocked);
+    completed.store(true);
+  });
+  const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3);
+  while (!library_mutex.lock_attempted.load() && !completed.load()) {
+    assert(std::chrono::steady_clock::now() < deadline && "notifier did not run");
+    std::this_thread::yield();
+  }
+  assert(library_mutex.lock_attempted.load() && "unblock bypassed predicate mutex");
+  assert(blocked.load() && "predicate changed before waiter released its mutex");
+  library_mutex.underlying.unlock();
+  notifier.join();
+  assert(!blocked.load() && library_changed.count == 1);
+}
+'''
+        with tempfile.TemporaryDirectory(prefix="meshcore-rak4631-unblock-") as directory:
+            for name, implementation in (
+                ("guarded", helper),
+                ("unguarded", helper.replace(guard, "")),
+            ):
+                with self.subTest(implementation=name):
+                    binary = Path(directory) / name
+                    # Track only mutex acquisition; execute the actual helper's
+                    # predicate update and notification in both variants.
+                    source = harness.replace("@UNBLOCK@", implementation.replace("std::mutex", "TrackedMutex"))
+                    compiled = subprocess.run(
+                        [compiler, "-std=c++17", "-pthread", "-Wall", "-Wextra", "-Werror",
+                         *SANITIZER_FLAGS, "-x", "c++", "-", "-o", str(binary)],
+                        input=source, capture_output=True, text=True, timeout=60,
+                    )
+                    self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
+                    checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+                    if name == "guarded":
+                        self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+                    else:
+                        self.assertNotEqual(checked.returncode, 0)
+                        self.assertIn("unblock bypassed predicate mutex", checked.stderr)
+
     def test_production_runtime_with_background_dhcp_and_spi_gate(self):
         compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
         if compiler is None: