Merge pull request #226 from AllenNeuralDynamics/feat-improve-timeout-strategy

Monitor long-running RPC spawned processes to prevent timeout errors

Author: bruno-f-cruz

src/clabe/xml_rpc/_client.py

@@ -37,6 +37,12 @@ class XmlRpcClientSettings(ServiceSettings):
     timeout: float = Field(default=30.0, description="Default timeout for RPC calls in seconds")
     poll_interval: float = Field(default=0.5, description="Polling interval for job status checks in seconds")
     max_file_size: int = Field(default=5 * 1024 * 1024, description="Maximum file size in bytes (default 5MB)")
+    monitor: bool = Field(
+        default=True,
+        description="If True, timeout becomes a liveness check window. As long as the job is confirmed "
+        "running within the timeout period, waiting continues indefinitely. If False, timeout is the "
+        "total execution time limit.",
+    )
 
 
 class XmlRpcClient:
@@ -52,7 +58,9 @@ def __init__(self, settings: XmlRpcClientSettings):
         self.settings = settings
         self._client = xmlrpc.client.ServerProxy(str(settings.server_url), allow_none=True)
         self._token = settings.token.get_secret_value()
-        self._executor = XmlRpcExecutor(self, timeout=settings.timeout, poll_interval=settings.poll_interval)
+        self._executor = XmlRpcExecutor(
+            self, timeout=settings.timeout, poll_interval=settings.poll_interval, monitor=settings.monitor
+        )
 
         logger.info(f"RPC client initialized for server: {settings.server_url}")
 
@@ -152,7 +160,8 @@ def wait_for_result(self, job_id: str, timeout: Optional[float] = None) -> JobRe
             JobResult object with execution details
 
         Raises:
-            TimeoutError: If the command doesn't complete within the timeout
+            TimeoutError: If the command doesn't complete within the timeout (or if
+                monitor mode is enabled and the job stops responding within the timeout window)
 
         Example:
             ```python
@@ -166,13 +175,21 @@ def wait_for_result(self, job_id: str, timeout: Optional[float] = None) -> JobRe
 
         start_time = time.time()
 
-        while time.time() - start_time < timeout:
+        while True:
+            elapsed = time.time() - start_time
+            if elapsed >= timeout:
+                raise TimeoutError(f"Job {job_id} did not complete within {timeout} seconds")
+
             result = self.get_result(job_id)
             if result.status == JobStatus.DONE:
                 return result
-            time.sleep(self.settings.poll_interval)
 
-        raise TimeoutError(f"Job {job_id} did not complete within {timeout} seconds")
+            # In monitor mode, reset timer if job is still running
+            if self.settings.monitor and self.is_running(job_id):
+                logger.debug("Job %s is still running; resetting timeout timer", job_id)
+                start_time = time.time()
+
+            time.sleep(self.settings.poll_interval)
 
     def run_command(self, cmd_args: list[str] | str, timeout: Optional[float] = None) -> JobResult:
         """

src/clabe/xml_rpc/_executor.py

@@ -47,7 +47,11 @@ class XmlRpcExecutor:
     """
 
     def __init__(
-        self, client: "XmlRpcClient", timeout: Optional[float] = None, poll_interval: Optional[float] = None
+        self,
+        client: "XmlRpcClient",
+        timeout: Optional[float] = None,
+        poll_interval: Optional[float] = None,
+        monitor: bool = True,
     ) -> None:
         """
         Initialize the RPC executor.
@@ -56,6 +60,9 @@ def __init__(
             client: RPC client instance for communication with the remote server
             timeout: Maximum time to wait for command completion (uses client default if None)
             poll_interval: Polling interval for job status checks (uses client default if None)
+            monitor: If True, timeout becomes a liveness check window. As long as the job
+                is confirmed running within the timeout period, waiting continues indefinitely.
+                If False, timeout is the total execution time limit.
 
         Example:
             ```python
@@ -72,6 +79,7 @@ def __init__(
         self.client = client
         self.timeout = timeout
         self.poll_interval = poll_interval
+        self.monitor = monitor
 
         logger.info(f"RPC executor initialized for server: {client.settings.server_url}")
 
@@ -152,18 +160,26 @@ async def _wait_for_result_async(self, job_id: str):
             JobResult with execution details
 
         Raises:
-            TimeoutError: If job doesn't complete within timeout
+            TimeoutError: If job doesn't complete within timeout (or if monitor mode
+                is enabled and the job stops responding within the timeout window)
         """
 
         timeout = self.timeout or self.client.settings.timeout
         poll_interval = self.poll_interval or self.client.settings.poll_interval
         start_time = time.time()
 
-        while time.time() - start_time < timeout:
+        while True:
+            elapsed = time.time() - start_time
+            if elapsed >= timeout:
+                raise TimeoutError(f"Job {job_id} did not complete within {timeout} seconds")
+
             await asyncio.sleep(poll_interval)
             # this is synchronous but should be fast
             result = self.client.get_result(job_id)
             if result.status == JobStatus.DONE:
                 return result
 
-        raise TimeoutError(f"Job {job_id} did not complete within {timeout} seconds")
+            # In monitor mode, reset timer if job is still running
+            if self.monitor and self.client.is_running(job_id):
+                logger.debug("Job %s is still running; resetting timeout timer", job_id)
+                start_time = time.time()

tests/apps/test_rpc_executors.py

@@ -17,6 +17,7 @@ def mock_client(self):
         mock_client.settings.timeout = 30.0
         mock_client.settings.poll_interval = 0.5
         mock_client.settings.server_url = "http://localhost:8000"
+        mock_client.settings.monitor = True
         return mock_client
 
     @pytest.fixture
@@ -29,10 +30,19 @@ def test_init(self):
         mock_client = Mock()
         mock_client.settings.server_url = "http://localhost:8000"
 
-        executor = XmlRpcExecutor(mock_client, timeout=60.0, poll_interval=1.0)
+        executor = XmlRpcExecutor(mock_client, timeout=60.0, poll_interval=1.0, monitor=False)
         assert executor.client == mock_client
         assert executor.timeout == 60.0
         assert executor.poll_interval == 1.0
+        assert executor.monitor is False
+
+    def test_init_default_monitor(self):
+        """Test XmlRpcExecutor initialization with default monitor=True."""
+        mock_client = Mock()
+        mock_client.settings.server_url = "http://localhost:8000"
+
+        executor = XmlRpcExecutor(mock_client)
+        assert executor.monitor is True
 
     def test_run_success(self, executor):
         """Test successful synchronous command execution."""
@@ -96,3 +106,37 @@ async def test_run_async_no_job_id(self, executor):
 
         with pytest.raises(Exception, match="Job submission failed: no job ID returned"):
             await executor.run_async(cmd)
+
+    @pytest.mark.asyncio
+    async def test_run_async_with_monitor_enabled(self, mock_client):
+        """Test asynchronous command execution with monitor mode enabled (default)."""
+        # Create executor with monitor=True (default)
+        executor = XmlRpcExecutor(mock_client, monitor=True)
+
+        submission_response = JobSubmissionResponse(success=True, job_id="monitor-job")
+        # Simulate job running initially, then completing
+        running_result = JobResult(
+            job_id="monitor-job", status=JobStatus.RUNNING, stdout=None, stderr=None, returncode=None, error=None
+        )
+        done_result = JobResult(
+            job_id="monitor-job",
+            status=JobStatus.DONE,
+            stdout="Monitored async output",
+            stderr="",
+            returncode=0,
+            error=None,
+        )
+
+        executor.client.submit_command.return_value = submission_response
+        # First call returns running, second call returns done
+        executor.client.get_result.side_effect = [running_result, done_result]
+        executor.client.is_running.return_value = True
+
+        cmd = Command(cmd="long_async_command", output_parser=identity_parser)
+        result = await executor.run_async(cmd)
+
+        assert isinstance(result, CommandResult)
+        assert result.stdout == "Monitored async output"
+        assert result.exit_code == 0
+
+        executor.client.submit_command.assert_called_once_with("long_async_command")

tests/xml-rpc/test_client.py

@@ -94,6 +94,7 @@ def test_client_settings_creation(self):
         assert settings.timeout == 30.0
         assert settings.poll_interval == 0.5
         assert settings.max_file_size == 5 * 1024 * 1024
+        assert settings.monitor is True  # Default is True
 
     def test_client_settings_custom_values(self):
         """Test creating client settings with custom values."""
@@ -103,13 +104,15 @@ def test_client_settings_custom_values(self):
             timeout=60.0,
             poll_interval=1.0,
             max_file_size=10 * 1024 * 1024,
+            monitor=False,
         )
 
         assert str(settings.server_url) == "http://192.168.1.100:9000/"
         assert settings.token.get_secret_value() == "custom-token"
         assert settings.timeout == 60.0
         assert settings.poll_interval == 1.0
         assert settings.max_file_size == 10 * 1024 * 1024
+        assert settings.monitor is False
 
 
 class TestJobResult:
@@ -234,14 +237,25 @@ def test_wait_for_result_success(self, rpc_client: XmlRpcClient):
         assert result.returncode == 0
         assert "completed" in result.stdout
 
-    def test_wait_for_result_timeout(self, rpc_client: XmlRpcClient):
-        """Test timeout when waiting for command completion."""
+    def test_wait_for_result_timeout(self, test_server):
+        """Test timeout when waiting for command completion (with monitor=False)."""
         import sys
 
-        submission = rpc_client.submit_command([sys.executable, "-c", "import time; time.sleep(10)"])
+        # Create a client with monitor=False to test actual timeout behavior
+        _, port, token = test_server
+        settings = XmlRpcClientSettings(
+            server_url=HttpUrl(f"http://127.0.0.1:{port}"),
+            token=SecretStr(token),
+            timeout=5.0,
+            poll_interval=0.1,
+            monitor=False,  # Disable monitor mode to test hard timeout
+        )
+        client = XmlRpcClient(settings)
+
+        submission = client.submit_command([sys.executable, "-c", "import time; time.sleep(10)"])
 
         with pytest.raises(TimeoutError, match="did not complete within"):
-            rpc_client.wait_for_result(submission.job_id, timeout=0.5)
+            client.wait_for_result(submission.job_id, timeout=0.5)
 
     def test_run_command_success(self, rpc_client: XmlRpcClient):
         """Test running a command to completion."""
@@ -267,6 +281,49 @@ def test_is_running(self, rpc_client: XmlRpcClient):
         is_running_after = rpc_client.is_running(submission.job_id)
         assert is_running_after is False
 
+    def test_wait_for_result_with_monitor_default(self, rpc_client: XmlRpcClient):
+        """Test that monitor mode (default) completes successfully for long-running job."""
+        import sys
+
+        # Submit a job that runs longer than the timeout
+        submission = rpc_client.submit_command([sys.executable, "-c", "import time; time.sleep(0.5); print('done')"])
+
+        # The job takes 0.5s but timeout is 0.3s - with default monitor=True,
+        # the timeout resets each time is_running returns True
+        result = rpc_client.wait_for_result(submission.job_id, timeout=0.3)
+
+        assert result.status == JobStatus.DONE
+        assert result.returncode == 0
+        assert "done" in result.stdout
+
+    def test_wait_for_result_monitor_resets_timeout(self, rpc_client: XmlRpcClient):
+        """Test that monitor mode (default) resets timeout when job is confirmed running."""
+        import sys
+
+        # Submit a job that takes longer than the timeout
+        submission = rpc_client.submit_command(
+            [sys.executable, "-c", "import time; time.sleep(0.8); print('finished')"]
+        )
+
+        # With default monitor=True, the timeout resets each poll cycle while job is running
+        result = rpc_client.wait_for_result(submission.job_id, timeout=0.4)
+
+        assert result.status == JobStatus.DONE
+        assert "finished" in result.stdout
+
+    def test_run_command_with_monitor_default(self, rpc_client: XmlRpcClient):
+        """Test run_command with default monitor mode enabled."""
+        import sys
+
+        # With default monitor=True, long-running commands complete successfully
+        result = rpc_client.run_command(
+            [sys.executable, "-c", "import time; time.sleep(0.5); print('monitored')"],
+            timeout=0.3,
+        )
+
+        assert result.status == JobStatus.DONE
+        assert "monitored" in result.stdout
+
     def test_list_jobs(self, rpc_client: XmlRpcClient):
         """Test listing jobs."""
         # Submit a job