Retry nonblocking Ethernet operations in runtime regression fixture

test/fixtures/rak4631_ethernet_runtime/test.cpp

@@ -60,6 +60,16 @@ static void wait_until(const std::function<bool()>& condition) {
     std::this_thread::sleep_for(std::chrono::milliseconds(1));
   }
 }
+static void write_all(const uint8_t* data, size_t size) {
+  size_t accepted = 0;
+  wait_until([&] {
+    const size_t remaining = size - accepted;
+    const size_t written = ethernet_client.write(data + accepted, remaining);
+    assert(written <= remaining);
+    accepted += written;
+    return accepted == size;
+  });
+}
 void mock_bus_touch() {
   assert(bus_owner.load() && spi_active.load());
   assert(ethernet_spi_access.load() != EthernetSpiAccess::Idle);
@@ -392,23 +402,27 @@ int main(int argc, char** argv) {
       start_online();
       assert(command("get eth") == "> on");
       assert(command("eth on").find("already on") != std::string::npos);
-      const int before_status = hardware_operations.load();
+      const int before_status = main_hardware_operations.load();
       assert(command("eth.status") == "ETH: 192.168.1.42:23");
-      assert(hardware_operations.load() == before_status);
+      assert(main_hardware_operations.load() == before_status);
       accept_connection();
       assert(ethernet_take_session_reset() && !ethernet_take_session_reset());
-      assert(ethernet_client.connected());
+      wait_until([] { return ethernet_client.connected(); });
       mock_ethernet_input = "eth.status\r\n";
       char line[32] = {};
-      assert(ethernet_read_line(line, sizeof(line)) && std::string(line) == "eth.status");
-      wait_until([] { return ethernet_is_running(); });
-      assert(ethernet_client.write(reinterpret_cast<const uint8_t*>("abcde"), 5) == 5);
-      wait_until([] { return output().find("abcde") != std::string::npos && ethernet_is_running(); });
+      // Online is only a snapshot, not a reservation against the SPI worker.
+      // Keep parser state and retry only bytes the Stream has not accepted.
+      wait_until([&] { return ethernet_read_line(line, sizeof(line)); });
+      assert(std::string(line) == "eth.status");
+      write_all(reinterpret_cast<const uint8_t*>("abcde"), 5);
+      const std::string expected = std::string(ETHERNET_CLI_BANNER) + "\r\n\r\nabcde";
+      wait_until([&] { return output() == expected; });
       mock_socket_connected[0].store(false);
       wait_until([] { return ethernet_session_reset.load() && !ethernet_client_attached.load(); });
       wait_until([] { return ethernet_is_running(); });
       assert(ethernet_take_session_reset() && !ethernet_client.connected());
       finish_off();
+      assert(output() == expected); // No delayed duplicate writes after drain.
       break;
     }
     case 8: { // Renewal owns SPI; every retained Stream call yields without raw IO.

test/test_rak4631_ethernet_runtime.py

@@ -53,6 +53,64 @@ SCENARIOS = (
 
 
 class Rak4631EthernetRuntimeTests(unittest.TestCase):
+    def test_fixture_retries_busy_and_short_writes_without_duplicates(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")
+        fixture = (FIXTURE / "test.cpp").read_text(encoding="ascii")
+        waiter = method(fixture, "static void wait_until(")
+        writer = method(fixture, "static void write_all(")
+        harness = r'''
+#include <cassert>
+#include <chrono>
+#include <cstdint>
+#include <functional>
+#include <string>
+#include <thread>
+struct Client {
+  unsigned calls = 0;
+  std::string output;
+  size_t write(const uint8_t* data, size_t size) {
+    const size_t capacity[] = {0, 2, 0, 1, 1, 1};
+    assert(calls < sizeof(capacity) / sizeof(capacity[0]));
+    const size_t offered = capacity[calls++];
+    const size_t accepted = size < offered ? size : offered;
+    output.append(reinterpret_cast<const char*>(data), accepted);
+    return accepted;
+  }
+} ethernet_client;
+@WAITER@
+@WRITER@
+int main() {
+  const uint8_t payload[] = {'a', 'b', 'c', 'd', 'e'};
+  write_all(payload, sizeof(payload));
+  assert(ethernet_client.output == "abcde" && "busy/short retries must preserve exact bytes");
+  assert(ethernet_client.calls == 6);
+}
+'''
+        with tempfile.TemporaryDirectory(prefix="meshcore-ethernet-fixture-write-") as directory:
+            for name, implementation in (
+                ("retry-unsent", writer),
+                ("single-attempt", "static void write_all(const uint8_t* data, size_t size) {\n"
+                 "wait_until([] { return true; });\n"
+                 "assert(ethernet_client.write(data, size) == size);\n}"),
+                ("duplicate-prefix", writer.replace("data + accepted", "data")),
+            ):
+                with self.subTest(implementation=name):
+                    binary = Path(directory) / name
+                    compiled = subprocess.run(
+                        [compiler, "-std=c++17", "-pthread", "-Wall", "-Wextra", "-Werror",
+                         *SANITIZER_FLAGS, "-x", "c++", "-", "-o", str(binary)],
+                        input=harness.replace("@WAITER@", waiter).replace("@WRITER@", implementation),
+                        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 == "retry-unsent":
+                        self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+                    else:
+                        self.assertNotEqual(checked.returncode, 0)
+                        self.assertIn("Assertion", checked.stderr)
+
     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:
@@ -141,7 +199,9 @@ int main() {
                 capture_output=True, text=True, timeout=60,
             )
             self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
-            for scenario, description in enumerate(SCENARIOS):
+            cases = list(enumerate(SCENARIOS))
+            cases.extend((7, f"online ownership race repeat {index}") for index in range(32))
+            for scenario, description in cases:
                 with self.subTest(scenario=description):
                     checked = subprocess.run([str(binary), str(scenario)],
                                              capture_output=True, text=True, timeout=10)