fix: raise TimeoutError on ZMQ retry exhaustion (#393)

send_json_with_retry() and recv_json_with_retry() now raise
TimeoutError instead of silently returning None when all 5 retries
are exhausted.  read() and write() catch the new exception so callers
never see an unhandled crash.

- send_json_with_retry: raise TimeoutError (was: return)
- recv_json_with_retry: raise TimeoutError (was: return None)
- read(): except TimeoutError -> (default, False), status TIMEOUT
- write(): except TimeoutError -> log and continue

Tests added in test_concore.py and test_concoredocker.py.

Closes #393

Author: GaneshPatil7517

concore_base.py

@@ -59,8 +59,7 @@ def send_json_with_retry(self, message):
             except zmq.Again:
                 logger.warning(f"Send timeout (attempt {attempt + 1}/5)")
                 time.sleep(0.5)
-        logger.error("Failed to send after retries.")
-        return
+        raise TimeoutError(f"ZMQ send failed after 5 retries on {self.address}")
 
     def recv_json_with_retry(self):
         """Receive JSON message with retries if timeout occurs."""
@@ -70,8 +69,7 @@ def recv_json_with_retry(self):
             except zmq.Again:
                 logger.warning(f"Receive timeout (attempt {attempt + 1}/5)")
                 time.sleep(0.5)
-        logger.error("Failed to receive after retries.")
-        return None
+        raise TimeoutError(f"ZMQ recv failed after 5 retries on {self.address}")
 
 
 def init_zmq_port(mod, port_name, port_type, address, socket_type_str):
@@ -282,6 +280,10 @@ def read(mod, port_identifier, name, initstr_val):
                     return message[1:], True
             last_read_status = "SUCCESS"
             return message, True
+        except TimeoutError as e:
+            logger.error(f"ZMQ read timeout on port {port_identifier} (name: {name}): {e}. Returning default.")
+            last_read_status = "TIMEOUT"
+            return default_return_val, False
         except zmq.error.ZMQError as e:
             logger.error(f"ZMQ read error on port {port_identifier} (name: {name}): {e}. Returning default.")
             last_read_status = "TIMEOUT"
@@ -384,6 +386,8 @@ def write(mod, port_identifier, name, val, delta=0):
                 # Mutation breaks cross-language determinism (see issue #385).
             else:
                 zmq_p.send_json_with_retry(zmq_val)
+        except TimeoutError as e:
+            logger.error(f"ZMQ write timeout on port {port_identifier} (name: {name}): {e}")
         except zmq.error.ZMQError as e:
             logger.error(f"ZMQ write error on port {port_identifier} (name: {name}): {e}")
         except Exception as e:

tests/test_concore.py

@@ -1,6 +1,7 @@
 import pytest
 import os
 import numpy as np
+from unittest.mock import patch
 
 
 class TestSafeLiteralEval:
@@ -450,3 +451,87 @@ def test_write_timestamp_matches_cpp_semantics(self, temp_dir):
             "After 3 writes with delta=1 simtime must remain 0 "
             "(matching C++/MATLAB/Verilog); got %s" % concore.simtime
         )
+
+
+class TestZMQRetryExhaustion:
+    """Issue #393 – recv_json_with_retry / send_json_with_retry must raise
+    TimeoutError instead of silently returning None when retries are
+    exhausted, and callers (read / write) must handle it gracefully."""
+
+    @patch("concore_base.time.sleep")
+    def test_recv_raises_timeout_error(self, _mock_sleep):
+        """recv_json_with_retry must raise TimeoutError after 5 failed attempts."""
+        import concore_base
+
+        port = concore_base.ZeroMQPort.__new__(concore_base.ZeroMQPort)
+        port.address = "tcp://127.0.0.1:9999"
+
+        class FakeSocket:
+            def recv_json(self, flags=0):
+                import zmq
+
+                raise zmq.Again("Resource temporarily unavailable")
+
+        port.socket = FakeSocket()
+        with pytest.raises(TimeoutError):
+            port.recv_json_with_retry()
+
+    @patch("concore_base.time.sleep")
+    def test_send_raises_timeout_error(self, _mock_sleep):
+        """send_json_with_retry must raise TimeoutError after 5 failed attempts."""
+        import concore_base
+
+        port = concore_base.ZeroMQPort.__new__(concore_base.ZeroMQPort)
+        port.address = "tcp://127.0.0.1:9999"
+
+        class FakeSocket:
+            def send_json(self, data, flags=0):
+                import zmq
+
+                raise zmq.Again("Resource temporarily unavailable")
+
+        port.socket = FakeSocket()
+        with pytest.raises(TimeoutError):
+            port.send_json_with_retry([42])
+
+    def test_read_returns_default_on_timeout(self, temp_dir):
+        """read() must return (default, False) when ZMQ recv times out."""
+        import concore
+        import concore_base
+
+        concore.inpath = os.path.join(temp_dir, "in")
+
+        class TimeoutPort:
+            address = "tcp://127.0.0.1:0"
+            socket = None
+
+            def recv_json_with_retry(self):
+                raise TimeoutError("ZMQ recv failed after 5 retries")
+
+        concore.zmq_ports["t_in"] = TimeoutPort()
+        concore.simtime = 0
+
+        result, ok = concore.read("t_in", "x", "[0.0]")
+
+        assert result == [0.0]
+        assert ok is False
+        assert concore_base.last_read_status == "TIMEOUT"
+
+    def test_write_does_not_crash_on_timeout(self, temp_dir):
+        """write() must not propagate TimeoutError to the caller."""
+        import concore
+
+        concore.outpath = os.path.join(temp_dir, "out")
+        os.makedirs(os.path.join(temp_dir, "out_t_out"), exist_ok=True)
+
+        class TimeoutPort:
+            address = "tcp://127.0.0.1:0"
+            socket = None
+
+            def send_json_with_retry(self, message):
+                raise TimeoutError("ZMQ send failed after 5 retries")
+
+        concore.zmq_ports["t_out"] = TimeoutPort()
+        concore.simtime = 0
+
+        concore.write("t_out", "y", [1.0], delta=1)

tests/test_concoredocker.py

@@ -247,3 +247,50 @@ def recv_json_with_retry(self):
 
         assert result == original
         assert ok is True
+
+
+class TestZMQRetryExhaustion:
+    """Issue #393 – concoredocker read/write must handle TimeoutError
+    raised by ZMQ retry exhaustion without crashing."""
+
+    def test_read_returns_default_on_timeout(self, temp_dir):
+        """read() must return (default, False) when ZMQ recv times out."""
+        import concoredocker
+        import concore_base
+
+        concoredocker.inpath = os.path.join(temp_dir, "in")
+
+        class TimeoutPort:
+            address = "tcp://127.0.0.1:0"
+            socket = None
+
+            def recv_json_with_retry(self):
+                raise TimeoutError("ZMQ recv failed after 5 retries")
+
+        concoredocker.zmq_ports["t_in"] = TimeoutPort()
+        concoredocker.simtime = 0
+
+        result, ok = concoredocker.read("t_in", "x", "[0.0]")
+
+        assert result == [0.0]
+        assert ok is False
+        assert concore_base.last_read_status == "TIMEOUT"
+
+    def test_write_does_not_crash_on_timeout(self, temp_dir):
+        """write() must not propagate TimeoutError to the caller."""
+        import concoredocker
+
+        concoredocker.outpath = os.path.join(temp_dir, "out")
+        os.makedirs(os.path.join(temp_dir, "out_t_out"), exist_ok=True)
+
+        class TimeoutPort:
+            address = "tcp://127.0.0.1:0"
+            socket = None
+
+            def send_json_with_retry(self, message):
+                raise TimeoutError("ZMQ send failed after 5 retries")
+
+        concoredocker.zmq_ports["t_out"] = TimeoutPort()
+        concoredocker.simtime = 0
+
+        concoredocker.write("t_out", "y", [1.0], delta=1)