Bound Ethernet receive work to the entry byte budget

src/helpers/ethernet/SerialEthernetInterface.cpp

@@ -165,10 +165,12 @@ size_t SerialEthernetInterface::checkRecvFrame(uint8_t dest[]) {
         send_queue[i] = send_queue[i + 1];
       }
     } else {
-      while (true) {
+      // Newly arriving bytes belong to the next mesh loop. A continuously
+      // writing peer must not monopolize radio and watchdog service.
+      int pending = available();
+      while (pending-- > 0) {
         // A read may likewise hand its receive-ACK command to the worker;
         // leave the rest of the frame untouched until hardware is available.
-        if (updateIOPause((uint32_t)millis()) || !available()) break;
         if (updateIOPause((uint32_t)millis())) break;
         int c = read();
         if (c < 0) break;
@@ -231,6 +233,9 @@ size_t SerialEthernetInterface::checkRecvFrame(uint8_t dest[]) {
         }
 #endif
       }
+      // Snapshot availability or the final budgeted read can hand IO to the
+      // worker. Observe that pause after the byte has changed parser state.
+      (void)updateIOPause((uint32_t)millis());
     }
   }
 

test/fixtures/serial_ethernet_sessions/test.cpp

@@ -42,6 +42,9 @@ struct Transport : SerialEthernetInterface {
   bool pause_after_write = false;
   bool pause_after_read = false;
   bool pause_after_available = false;
+  bool replenish_read = false;
+  size_t read_guard = 0;
+  Bytes append_after_read;
   int available_calls = 0;
   int write_calls = 0;
   int disconnect_calls = 0;
@@ -67,9 +70,16 @@ struct Transport : SerialEthernetInterface {
   int read() override {
     assert(!io_unavailable);
     if (!client || client->received.empty()) return -1;
+    assert((read_guard == 0 || client->read_count < read_guard)
+        && "Ethernet RX invocation exceeded bounded read guard");
     const int value = client->received.front();
     client->received.pop_front();
     ++client->read_count;
+    if (replenish_read) client->received.push_back(uint8_t(value));
+    if (!append_after_read.empty()) {
+      client->received.insert(client->received.end(), append_after_read.begin(), append_after_read.end());
+      append_after_read.clear();
+    }
     if (pause_after_read) {
       io_unavailable = true;
       pause_after_read = false;
@@ -500,6 +510,52 @@ int main(int argc, char** argv) {
     assert(f.tick() == 0 && !socket->connected && f.callbacks == 1);
 #endif
     assert(other_socket->connected);
+  } else if (scenario >= 30 && scenario <= 35) {
+    const bool replenish = scenario <= 32;
+    static constexpr size_t budgets[] = {1, 7, 257};
+    const size_t budget = budgets[(scenario - 30) % 3];
+    const Bytes request = command(Bytes{'a', 'b'});
+#if ETHERNET_RAW_LINE
+    const uint8_t junk = '\n';
+#else
+    const uint8_t junk = 'x';
+#endif
+    socket->received.assign(budget, junk);
+    f.transport.read_guard = 4096; // The live-availability negative control must terminate.
+    f.transport.replenish_read = replenish;
+    if (!replenish) f.transport.append_after_read = request;
+    assert(f.tick() == 0 && "new arrivals must wait for the next mesh loop");
+    assert(socket->read_count == budget && f.transport.available_calls == 1);
+    assert(socket->received.size() == (replenish ? budget : request.size()));
+    assert(socket->connected && !f.transport.isReadBusy() && f.callbacks == 0);
+    f.transport.replenish_read = false;
+    if (replenish) socket->received.insert(socket->received.end(), request.begin(), request.end());
+    // Nothing was dropped: existing idle bytes and a newly arrived complete
+    // command remain in wire order and parse on the following mesh loop.
+    assert(f.tick() == 2 && f.input[0] == 'a' && f.input[1] == 'b');
+    assert(socket->received.empty() && f.transport.available_calls == 2);
+  } else if (scenario >= 36 && scenario <= 39) {
+    const bool zero_available = scenario >= 38;
+    mock_millis = (scenario & 1) ? UINT32_MAX - 1000U : 0;
+    const Bytes request = command(Bytes{'a', 'b'});
+    socket->received.push_back(request[0]);
+    if (!zero_available) f.transport.pause_after_read = true;
+    assert(f.tick() == 0 && f.transport.isReadBusy());
+    if (zero_available) {
+      mock_millis += 2000;
+      f.transport.pause_after_available = true;
+      assert(f.tick() == 0);
+    }
+    assert(f.transport.io_unavailable && socket->read_count == 1);
+    const int available_calls = f.transport.available_calls;
+    mock_millis += 12000;
+    assert(f.tick() == 0 && f.transport.available_calls == available_calls);
+    f.transport.io_unavailable = false;
+    f.transport.loop();
+    mock_millis += zero_available ? 2999 : 4999;
+    socket->received.insert(socket->received.end(), request.begin() + 1, request.end());
+    assert((f.tick() == 2 && socket->connected && f.input[0] == 'a' && f.input[1] == 'b'
+        && f.callbacks == 0) && "worker handoff must preserve remaining frame deadline");
   } else {
     assert(false && "unknown scenario");
   }

test/test_serial_ethernet_sessions.py

@@ -46,10 +46,30 @@ SCENARIOS = (
     "read command ownership pauses before polling the socket again",
     "availability command ownership prevents a following socket read",
     "main-loop ownership service preserves timeout behind another locked route",
+    "continuously replenished idle input yields after entry budget 1",
+    "continuously replenished idle input yields after entry budget 7",
+    "continuously replenished idle input yields after entry budget 257",
+    "new command waits behind entry budget 1 until next loop",
+    "new command waits behind entry budget 7 until next loop",
+    "new command waits behind entry budget 257 until next loop",
+    "last budgeted read records worker pause after parser state changes",
+    "last budgeted read worker pause preserves deadline across rollover",
+    "zero-available snapshot records worker pause immediately",
+    "zero-available worker pause preserves deadline across rollover",
 )
 
 
 class SerialEthernetSessionTests(unittest.TestCase):
+    def compile_transport(self, compiler, raw, binary, source):
+        compiled = subprocess.run(
+            [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+             *SANITIZER_FLAGS, f"-DETHERNET_RAW_LINE={raw}",
+             f"-I{MOCKS}", f"-I{ROOT / 'src'}",
+             f"-I{ROOT / 'src/helpers/ethernet'}", str(FIXTURE / "test.cpp"),
+             str(source), "-o", str(binary)], capture_output=True, text=True, timeout=60,
+        )
+        self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
+
     def test_production_transport_sessions(self):
         compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
         if compiler is None:
@@ -57,15 +77,8 @@ class SerialEthernetSessionTests(unittest.TestCase):
         with tempfile.TemporaryDirectory(prefix="meshcore-ethernet-sessions-") as directory:
             binary = Path(directory) / "ethernet-sessions.exe"
             for raw in (0, 1):
-                compiled = subprocess.run(
-                    [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
-                     *SANITIZER_FLAGS, f"-DETHERNET_RAW_LINE={raw}",
-                     f"-I{MOCKS}", f"-I{ROOT / 'src'}",
-                     str(FIXTURE / "test.cpp"),
-                     str(ROOT / "src/helpers/ethernet/SerialEthernetInterface.cpp"),
-                     "-o", str(binary)], capture_output=True, text=True, timeout=60,
-                )
-                self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
+                self.compile_transport(compiler, raw, binary,
+                                       ROOT / "src/helpers/ethernet/SerialEthernetInterface.cpp")
                 for scenario, description in enumerate(SCENARIOS):
                     with self.subTest(raw=raw, scenario=description):
                         checked = subprocess.run(
@@ -75,6 +88,44 @@ class SerialEthernetSessionTests(unittest.TestCase):
                         self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
                         self.assertIn(f"PASS: Ethernet session scenario {scenario}", checked.stdout)
 
+    def test_live_available_and_missing_final_pause_negative_controls(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")
+        source = (ROOT / "src/helpers/ethernet/SerialEthernetInterface.cpp").read_text()
+        snapshot = '''      int pending = available();
+      while (pending-- > 0) {
+        // A read may likewise hand its receive-ACK command to the worker;
+        // leave the rest of the frame untouched until hardware is available.
+        if (updateIOPause((uint32_t)millis())) break;'''
+        live = '''      while (true) {
+        if (updateIOPause((uint32_t)millis()) || !available()) break;
+        if (updateIOPause((uint32_t)millis())) break;'''
+        pause = '''      // Snapshot availability or the final budgeted read can hand IO to the
+      // worker. Observe that pause after the byte has changed parser state.
+      (void)updateIOPause((uint32_t)millis());'''
+        self.assertEqual(source.count(snapshot), 1)
+        self.assertEqual(source.count(pause), 1)
+        with tempfile.TemporaryDirectory(prefix="meshcore-ethernet-rx-negative-") as directory:
+            work = Path(directory)
+            for name, modified, modes, scenarios in (
+                    ("live", source.replace(snapshot, live), (0, 1), range(30, 36)),
+                    ("naive", source.replace(pause, ""), (0,), range(36, 40))):
+                variant = work / (name + ".cpp")
+                variant.write_text(modified)
+                for raw in modes:
+                    binary = work / (name + str(raw) + ".exe")
+                    self.compile_transport(compiler, raw, binary, variant)
+                    for scenario in scenarios:
+                        with self.subTest(variant=name, raw=raw, scenario=SCENARIOS[scenario]):
+                            checked = subprocess.run([str(binary), str(scenario)], capture_output=True,
+                                                     text=True, timeout=10)
+                            self.assertNotEqual(checked.returncode, 0, "unsafe RX loop escaped regression")
+                            expected = ("bounded read guard" if scenario <= 32 else
+                                        "new arrivals must wait" if scenario <= 35 else
+                                        "worker handoff must preserve remaining frame deadline")
+                            self.assertIn(expected, checked.stderr)
+
     def test_hardware_drivers_cancel_before_exposing_new_client(self):
         for driver in ("ch390/CH390EthernetInterface", "RAK13800/RAK13800EthernetInterface"):
             source = (ROOT / f"src/helpers/ethernet/{driver}.cpp").read_text(encoding="utf-8")