Repository: radiocosmology/alpenhorn

Issue:
Add timeout to `LustreHSMNodeIO._restore_wait`
To avoid the potential for locking-up all the workers due to restore requests being abandonned/forgotten about/ignored by HSM without alpenhorn's knowledge.

https://github.com/radiocosmology/alpenhorn/blob/4670ec285f810bef19199fd4a9cc2848d81e539e/alpenhorn/io/lustrehsm.py#L113

Probably best to have something configurable via conf file, but I want to run as-is for a while to see what timescale would be reasonable as a default here.

Discussion:

Patch:
diff --git a/alpenhorn/io/lfs.py b/alpenhorn/io/lfs.py
index 06e2777..9ec2d6a 100644
--- a/alpenhorn/io/lfs.py
+++ b/alpenhorn/io/lfs.py
@@ -36,6 +36,7 @@ from alpenhorn import util
 
 if TYPE_CHECKING:
     import os
+del TYPE_CHECKING
 
 
 log = logging.getLogger(__name__)
@@ -113,7 +114,7 @@ class LFS:
         if self._lfs is None:
             raise RuntimeError("lfs command not found.")
 
-    def run_lfs(self, *args: str) -> str | None:
+    def run_lfs(self, *args: str, timeout: float | None = None) -> str | False | None:
         """Run the lfs command with the `args` provided.
 
         Parameters
@@ -121,26 +122,35 @@ class LFS:
         *args : strings
             The list of command-line arguments to pass to the
             lfs command.
+        timeout : float, optional
+            If not None, stop waiting for the command after `timeout` seconds
 
         Retunrs
         -------
-        output : str or None
+        output : str or False or None
             If the command succeeded, returns standard output of
-            the command.  If the command failed, returns None and
-            logs the failure.
+            the command.  If the command failed or timed out,
+            returns False (failed) or None (timed out) and logs
+            the failure.
         """
         # Stringify args
         args = [str(arg) for arg in args]
 
-        ret, stdout, stderr = util.run_command([self._lfs] + args)
+        ret, stdout, stderr = util.run_command([self._lfs] + args, timeout=timeout)
 
-        if ret != 0:
-            log.warning(f"LFS command failed (ret={ret}): " + " ".join(args))
+        # Failure or timeout
+        if ret is None or ret != 0:
+            if ret is None:
+                result = "timed out"
+            else:
+                result = f"failed (ret={ret})"
+                ret = False
+            log.warning(f"LFS command {result}: " + " ".join(args))
             if stderr:
                 log.debug(f"LFS stderr: {stderr}")
             if stdout:
                 log.debug(f"LFS stdout: {stdout}")
-            return None
+            return ret
 
         return stdout
 
@@ -259,7 +269,7 @@ class LFS:
         path = str(path)
 
         stdout = self.run_lfs("hsm_state", path)
-        if stdout is None:
+        if stdout is False:
             return None  # Command returned error
 
         # The output of hsm_state looks like this:
@@ -304,7 +314,7 @@ class LFS:
         """Is `path` released to external storage?"""
         return self.hsm_state(path) == HSMState.RELEASED
 
-    def hsm_restore(self, path: os.PathLike) -> bool:
+    def hsm_restore(self, path: os.PathLike) -> bool | None:
         """Trigger restore of `path` from external storage.
 
         If `path` is already restored or is missing, this does nothing.
@@ -316,9 +326,11 @@ class LFS:
 
         Returns
         -------
-        restored : bool
+        restored : bool or None
+            None if the request timed out.
+            False if the request failed.
             True if a successful restore request was made, or if
-            `path` was already restored.  False otherwise.
+            `path` was already restored.
         """
         state = self.hsm_state(path)
 
@@ -331,7 +343,10 @@ class LFS:
         if state != HSMState.RELEASED:
             return True
 
-        return self.run_lfs("hsm_restore", path) is not None
+        result = self.run_lfs("hsm_restore", path, timeout=60)
+        if result is None or result is False:
+            return result
+        return True
 
     def hsm_release(self, path: os.PathLike) -> bool:
         """Trigger release of `path` from disk.
@@ -363,4 +378,4 @@ class LFS:
             return False
 
         # Otherwise send the request
-        return self.run_lfs("hsm_release", path) is not None
+        return self.run_lfs("hsm_release", path) is not False
diff --git a/alpenhorn/io/lustrehsm.py b/alpenhorn/io/lustrehsm.py
index e4959da..3d3a160 100644
--- a/alpenhorn/io/lustrehsm.py
+++ b/alpenhorn/io/lustrehsm.py
@@ -36,9 +36,14 @@ if TYPE_CHECKING:
     from ..archive import ArchiveFileCopyRequest
     from ..queue import FairMultiFIFOQueue
     from ..update import UpdateableNode, UpdateableGroup
+del TYPE_CHECKING
 
 log = logging.getLogger(__name__)
 
+# Retry delays (in seconds) after a successful or timed-out hsm_restore request
+RESTORE_TIMEOUT_RETRY = 1 * 3600  # 1 hour
+RESTORE_SUCCESS_RETRY = 4 * 3600  # 4 hours
+
 
 class LustreHSMNodeRemote(BaseNodeRemote):
     """LustreHSMNodeRemote: information about a LustreHSM remote node."""
@@ -100,6 +105,10 @@ class LustreHSMNodeIO(LustreQuotaNodeIO):
         # `ArchiveFile.id`s
         self._restoring = set()
 
+        # The time.monotonic value after which we should try to restore again.
+        # Keys are elements in self._restoring.
+        self._restore_retry = dict()
+
         # For informational purposes.  Keys are elements in self._restoring.
         self._restore_start = dict()
 
@@ -136,6 +145,7 @@ class LustreHSMNodeIO(LustreQuotaNodeIO):
         if state == self._lfs.HSM_MISSING:
             log.warning(f"Unable to restore {copy.path}: missing.")
             self._restore_start.pop(copy.file.id, None)
+            self._restore_retry.pop(copy.file.id, None)
             self._restoring.discard(copy.file.id)
             return None
 
@@ -150,20 +160,53 @@ class LustreHSMNodeIO(LustreQuotaNodeIO):
                     f"after {pretty_deltat(deltat)}"
                 )
                 self._restore_start.pop(copy.file.id, None)
+                self._restore_retry.pop(copy.file.id, None)
                 self._restoring.discard(copy.file.id)
             return False
 
         # If we got here, copy is released.
 
-        # Add this copy to the list of copies we're waiting on.  Other tasks
-        # can use this to see if the file they're interested in is already
-        # "in use".
-        if copy.file.id not in self._restoring:
-            self._restoring.add(copy.file.id)
-            self._restore_start[copy.file.id] = time.monotonic()
+        # Have we hit the retry time for an in-progress restore?
+        retry_restore = (
+            copy.file.id in self._restoring
+            and self._restore_retry[copy.file.id] <= time.monotonic()
+        )
+
+        if retry_restore or copy.file.id not in self._restoring:
+            # Add this copy to the list of copies we're waiting on.  Other tasks
+            # can use this to see if the file they're interested in is already
+            # "in use".
+            if not retry_restore:
+                self._restoring.add(copy.file.id)
+                self._restore_start[copy.file.id] = time.monotonic()
+
+            # Try to restore it.
+            result = self._lfs.hsm_restore(copy.path)
+            log.warning(f"restore result: {result}")
+
+            if result is False:
+                # Reqeust failed. Abandon the restore attempt entirely,
+                # in case it was deleted from the node.
+                self._restore_retry.pop(copy.file.id, None)
+                self._restore_start.pop(copy.file.id, None)
+                self._restoring.discard(copy.file.id)
 
-            # Restore it.
-            self._lfs.hsm_restore(copy.path)
+                # Report failure
+                return None
+            elif result is None:
+                # Request timeout.  This seems to happen when the file
+                # is already being restored, but let's try again in a
+                # little while, just to be safe.
+                self._restore_retry[copy.file.id] = (
+                    time.monotonic() + RESTORE_TIMEOUT_RETRY
+                )
+            else:
+                # Request success; restore should be in-progress, but
+                # we'll still schedule a retry for some time in the future
+                # to guard against HSM forgetting/abandonning our request
+                self._restore_retry[copy.file.id] = (
+                    time.monotonic() + RESTORE_SUCCESS_RETRY
+                )
 
         # Tell the caller to wait
         return True
diff --git a/alpenhorn/util.py b/alpenhorn/util.py
index b41f22d..1a28ca8 100644
--- a/alpenhorn/util.py
+++ b/alpenhorn/util.py
@@ -51,9 +51,7 @@ def run_command(
     except subprocess.TimeoutExpired:
         log.warning(f"Process overrun [timeout={timeout}]: " + " ".join(cmd))
         proc.kill()
-        stdout_val = ""
-        stderr_val = ""
-        retval = None
+        return (None, "", "")
 
     return (
         retval,


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index a057a8a..0608f86 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -39,6 +39,13 @@ def pytest_configure(config):
         "to indicate the desired quota that LFS.quota_remaining "
         "should return.",
     )
+    config.addinivalue_line(
+        "markers",
+        "lfs_hsm_restore_result(result): "
+        "used on tests which mock alpenhorn.io.lfs.LFS "
+        "to indicate the result of the hsm_restore call.  result "
+        "may be 'fail', 'timeout', or 'wait'",
+    )
     config.addinivalue_line(
         "markers",
         "lfs_dont_mock(*method_names): "
@@ -225,7 +232,20 @@ def mock_lfs(have_lfs, request):
         raise ValueError("Bad state in lfs_hsm_state marker: {state} for path {path}")
 
     def _mocked_lfs_hsm_restore(self, path):
-        nonlocal lfs_hsm_state
+        nonlocal request, lfs_hsm_state
+
+        marker = request.node.get_closest_marker("lfs_hsm_restore_result")
+        if marker:
+            if marker.args[0] == "fail":
+                # Return failure
+                return False
+            if marker.args[0] == "timeout":
+                # Return timeout
+                return None
+            if marker.args[0] == "wait":
+                # Return true (successful request)
+                # without changing state to "restored"
+                return True
 
         # de-pathlib-ify
         path = str(path)
diff --git a/tests/io/test_lfs.py b/tests/io/test_lfs.py
index d4e93e7..9a54c3f 100644
--- a/tests/io/test_lfs.py
+++ b/tests/io/test_lfs.py
@@ -40,7 +40,7 @@ def test_run_lfs_fail(have_lfs, mock_run_command):
 
     lfs = LFS(None)
 
-    assert lfs.run_lfs("arg1", "arg2") is None
+    assert lfs.run_lfs("arg1", "arg2") is False
     assert mock_run_command() == {
         "cmd": ["LFS", "arg1", "arg2"],
         "kwargs": dict(),
@@ -255,7 +255,7 @@ def test_hsm_restore(mock_lfs, mock_run_command):
 
     lfs = LFS("qgroup")
 
-    assert not lfs.hsm_restore("/missing")
+    assert lfs.hsm_restore("/missing") is False
     assert lfs.hsm_restore("/unarchived")
     assert lfs.hsm_restore("/restored")
 
@@ -268,6 +268,17 @@ def test_hsm_restore(mock_lfs, mock_run_command):
     assert "/released" in mock_run_command()["cmd"]
 
 
+@pytest.mark.run_command_result(None, "", "")
+@pytest.mark.lfs_hsm_state({"/released": "released"})
+@pytest.mark.lfs_dont_mock("hsm_restore")
+def test_hsm_restore_timeout(mock_lfs, mock_run_command):
+    """Test hsm_restore() timeout."""
+
+    lfs = LFS("qgroup")
+
+    assert lfs.hsm_restore("/released") is None
+
+
 @pytest.mark.lfs_hsm_state(
     {
         "/missing": "missing",
diff --git a/tests/io/test_lustrehsmnode.py b/tests/io/test_lustrehsmnode.py
index 9ded3fa..25992db 100644
--- a/tests/io/test_lustrehsmnode.py
+++ b/tests/io/test_lustrehsmnode.py
@@ -262,7 +262,6 @@ def test_check_released(xfs, queue, mock_lfs, node):
     """Test check on a non-ready, released file."""
 
     xfs.create_file("/node/simpleacq/file1")
-
     copy = ArchiveFileCopy.get(id=1)
     copy.ready = False
     copy.save()
@@ -536,3 +535,111 @@ def test_idle_update_not_ready(xfs, queue, mock_lfs, node):
     assert not ArchiveFileCopy.get(id=4).ready
     assert ArchiveFileCopy.get(id=4).last_update >= before
     assert ArchiveFileCopy.get(id=4).has_file == "N"
+
+
+@pytest.mark.lfs_hsm_state({"/node/simpleacq/file1": "released"})
+@pytest.mark.lfs_hsm_restore_result("wait")
+def test_hsm_restore_twice(xfs, queue, mock_lfs, node):
+    """Test that only one restore request is made."""
+
+    # File is not ready, and maybe corrupt
+    xfs.create_file("/node/simpleacq/file1")
+    copy = ArchiveFileCopy.get(id=1)
+    copy.has_file = "M"
+    copy.ready = False
+    copy.save()
+
+    # Task will restore the file
+    node.io.check(copy)
+
+    # One item in queue now
+    assert queue.qsize == 1
+
+    # Run the task
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Task has been requeued and is deferred
+    assert queue.deferred_size == 1
+    assert queue.qsize == 0
+
+    # Check the internal bookkeeping
+    assert copy.id in node.io._restoring
+    assert copy.id in node.io._restore_start
+    assert copy.id in node.io._restore_retry
+
+    # Try to add another task
+    node.io.check(copy)
+
+    # Second attempt shouldn't make a new task
+    # because the first one added copy.id to the
+    # list of in-progress restores
+    assert queue.deferred_size == 1
+    assert queue.qsize == 0
+
+
+@pytest.mark.lfs_hsm_state({"/node/simpleacq/file1": "released"})
+@pytest.mark.lfs_hsm_restore_result("timeout")
+def test_hsm_restore_timeout(xfs, queue, mock_lfs, node):
+    """Test handling of timeout in hsm_restore"""
+
+    # File is not ready, and maybe corrupt
+    xfs.create_file("/node/simpleacq/file1")
+    copy = ArchiveFileCopy.get(id=1)
+    copy.has_file = "M"
+    copy.ready = False
+    copy.save()
+
+    # Task will restore the file
+    node.io.check(copy)
+
+    # One item in queue now
+    assert queue.qsize == 1
+
+    # Run the task
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Task has been requeued and is deferred
+    assert queue.deferred_size == 1
+    assert queue.qsize == 0
+
+    # Check the internal bookkeeping
+    assert copy.id in node.io._restoring
+    assert copy.id in node.io._restore_start
+    assert copy.id in node.io._restore_retry
+
+
+@pytest.mark.lfs_hsm_state({"/node/simpleacq/file1": "released"})
+@pytest.mark.lfs_hsm_restore_result("fail")
+def test_hsm_restore_fail(xfs, queue, mock_lfs, node):
+    """Test handling of hsm_restore failure"""
+
+    # File is not ready, and maybe corrupt
+    xfs.create_file("/node/simpleacq/file1")
+    copy = ArchiveFileCopy.get(id=1)
+    copy.has_file = "M"
+    copy.ready = False
+    copy.save()
+
+    # Task will restore the file
+    node.io.check(copy)
+
+    # One item in queue now
+    assert queue.qsize == 1
+
+    # Run the task
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Task has been abandonned
+    assert queue.deferred_size == 0
+    assert queue.qsize == 0
+
+    # Check the internal bookkeeping
+    assert copy.id not in node.io._restoring
+    assert copy.id not in node.io._restore_start
+    assert copy.id not in node.io._restore_retry
diff --git a/tests/test_util.py b/tests/test_util.py
index 6ad06d6..ddaa9c1 100644
--- a/tests/test_util.py
+++ b/tests/test_util.py
@@ -26,7 +26,7 @@ def test_run_stdout():
 
 
 def test_run_stderr():
-    """Test getting stdout from run_command."""
+    """Test getting stderr from run_command."""
     retval, stdout, stderr = util.run_command(
         ["python3", "-c", "import os; os.write(2, b'stderr')"]
     )
@@ -35,6 +35,12 @@ def test_run_stderr():
     assert retval == 0
 
 
+def test_run_timeout():
+    """Test run_command timing out."""
+    retval, stdout, stderr = util.run_command(["sleep", "10"], timeout=0.1)
+    assert retval is None
+
+
 def test_md5sum_file(tmp_path):
     """Test util.md5sum_file"""