Repository: radiocosmology/alpenhorn

Issue:
Use stdlib copy for local file copies if rsync is not installed
See #14 

Discussion:

Patch:
diff --git a/alpenhorn/common/util.py b/alpenhorn/common/util.py
index ea1b052..efcca07 100644
--- a/alpenhorn/common/util.py
+++ b/alpenhorn/common/util.py
@@ -7,11 +7,16 @@ import hashlib
 import logging
 import socket
 import subprocess
+from typing import TYPE_CHECKING, Any
 
 import click
 
 from . import config, extensions, logger
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+del TYPE_CHECKING
+
 log = logging.getLogger(__name__)
 
 
@@ -193,6 +198,51 @@ def run_command(
     )
 
 
+def timeout_call(func: Callable, timeout: float, /, *args: Any, **kwargs: Any) -> Any:
+    """Call a (non-awaitable) function with a timeout.
+
+    Uses asyncio.to_thread to call a function in a thread that
+    will be killed if it runs over time.
+
+    Parameters
+    ----------
+    func : Callable
+        the function to call
+    timeout : float
+        timeout, in seconds
+    args, kwargs:
+        passed to `func`
+
+    Returns
+    -------
+    result:
+        The return value of func
+
+    Raises
+    ------
+    TimeoutError:
+        The call exceeded the timeout
+    """
+
+    # await-able wrapper
+    async def _async_wrapper(
+        func: Callable, timeout: float, args: tuple, kwargs: dict
+    ) -> Any:
+        try:
+            async with asyncio.timeout(timeout):
+                return await asyncio.to_thread(func, *args, **kwargs)
+        except TimeoutError:
+            log.error(f"Timeout after {pretty_deltat(timeout)} calling {func}.")
+            raise
+
+    # If timeout is not positive, don't even try
+    if timeout <= 0:
+        raise TimeoutError(f'Negative timeout for "{func}" in timeout_call.')
+
+    # Otherwise call via asyncio
+    return asyncio.run(_async_wrapper(func, timeout, args, kwargs))
+
+
 async def _md5sum_file(filename: str, hr: bool = True) -> str | None:
     """asyncio implementation of md5sum_file().
 
diff --git a/alpenhorn/io/_default_asyncs.py b/alpenhorn/io/_default_asyncs.py
index 87ea4b7..770901f 100644
--- a/alpenhorn/io/_default_asyncs.py
+++ b/alpenhorn/io/_default_asyncs.py
@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import errno
 import logging
 import pathlib
@@ -12,6 +11,7 @@ from typing import TYPE_CHECKING
 
 import peewee as pw
 
+from ..common.util import timeout_call
 from ..daemon.update import RemoteNode
 from ..db import ArchiveFileCopy, ArchiveFileCopyRequest, utcnow
 from . import ioutil
@@ -32,8 +32,9 @@ def pull_async(
 
     Things to try:
         - hard link (for nodes on the same filesystem)
-        - bbcp (if source is not on this host)
-        - rsync if all else fails
+        - bbcp (remote transfers only)
+        - rsync
+        - shutil.copy (local transfers only)
 
     Parameters
     ----------
@@ -154,8 +155,11 @@ def pull_async(
                 log.info(f"Pulling local file {req.file.path} using rsync")
                 ioresult = ioutil.rsync(from_path, to_file, req.file.size_b, local)
             else:
-                log.error("No commands available to complete local pull.")
-                ioresult = {"ret": -1, "check_src": False}
+                # No rsync?  Just use shutil.copy, I guess
+                log.warning("Falling back on shutil.copy to complete local pull.")
+                ioresult = ioutil.local_copy(
+                    from_path, to_dir, req.file.name, req.file.size_b
+                )
 
     # Delete the placeholder, if we created it
     placeholder.unlink(missing_ok=True)
@@ -183,21 +187,6 @@ def pull_async(
         io.node.update_avail_gb(new_avail)
 
 
-async def _size_from_stat(fullpath: pathlib.Path) -> int | None:
-    """Asynchronously stat a file to get its size."""
-
-    try:
-        async with asyncio.timeout(600):
-            stat = await asyncio.to_thread(fullpath.stat)
-    except TimeoutError:
-        log.error(f"Timeout trying to stat {fullpath}!")
-        return None
-
-    if stat:
-        return stat.st_size
-    return None
-
-
 def check_async(task: Task, io: BaseNodeIO, copy: ArchiveFileCopy) -> None:
     """Check a file copy.  This is asynchronous.
 
@@ -225,10 +214,12 @@ def check_async(task: Task, io: BaseNodeIO, copy: ArchiveFileCopy) -> None:
     # Does the copy exist?
     if fullpath.exists():
         # First check the size
-        size = asyncio.run(_size_from_stat(fullpath))
-        if size is None:
+        try:
+            size = timeout_call(fullpath.stat, 600).st_size
+        except (OSError, TimeoutError):
             # Abandon the check attempt if we can't stat
             return
+
         if copy.file.size_b and size != copy.file.size_b:
             log.error(
                 f"File {copyname} on node {io.node.name} is corrupt! "
diff --git a/alpenhorn/io/ioutil.py b/alpenhorn/io/ioutil.py
index b98b36f..e4b2d46 100644
--- a/alpenhorn/io/ioutil.py
+++ b/alpenhorn/io/ioutil.py
@@ -6,6 +6,7 @@ import errno
 import logging
 import pathlib
 import re
+import shutil
 import time
 from tempfile import TemporaryDirectory
 from typing import TYPE_CHECKING
@@ -302,7 +303,7 @@ def hardlink(
     ----------
     from_path : path-like
         Source location
-    to_dir : spath-like
+    to_dir : path-like
         Destination directory
     filename : str
         Destination filename
@@ -310,7 +311,8 @@ def hardlink(
     Returns
     -------
     ioresult : dict or None
-        None if hardlinking failed otherwise, the dict
+        `{"ret": 1}` if the attempt timed out;
+        None if hardlinking failed; otherwise, the dict
         `{"ret": 0, "md5sum": True}`
     """
 
@@ -321,22 +323,100 @@ def hardlink(
     # an existing file with a hardlink, so we have to do this ridiculousness:
     try:
         # Create a temporary directory as a subdirectory of the destination dir
-        with TemporaryDirectory(dir=to_dir) as tmpdir:
+        with TemporaryDirectory(dir=to_dir, prefix=".alpentemp") as tmpdir:
             # We can be certain that we can create anything we want in here
             tmp_path = pathlib.Path(tmpdir, filename)
+
             # Try to create the new hardlink in the tempdir
-            tmp_path.hardlink_to(from_path)
+            util.timeout_call(tmp_path.hardlink_to, 600, from_path)
+
             # Hardlink succeeded!  Overwrite any existing file atomically
             tmp_path.rename(dest_path)
     except OSError as e:
         # Link creation failed for some reason
         log.debug(f"hardlink failed: {e}")
         return None
+    except TimeoutError:
+        log.warning(f'Timeout trying to hardlink "{dest_path}".')
+        return {"ret": 1}
 
     # md5sum is obviously correct
     return {"ret": 0, "md5sum": True}
 
 
+def local_copy(
+    from_path: str | os.PathLike, to_dir: str | os.PathLike, filename: str, size_b: int
+) -> dict:
+    """Copy `from_path` to `to_dir/filename` using `shutil`
+
+    Atomically overwrites an existing `to_dir/filename`.
+
+    Copy attempt times out after `_pull_timeout(size_b)` seconds have elapsed.
+
+    After a successful copy, `common.util.md5sum_file` will be called to verify
+    the transfer was successful.
+
+    Parameters
+    ----------
+    from_path : path-like
+        Source location
+    to_dir : path-like
+        Destination directory
+    filename : str
+        Destination filename
+    size_b : int
+        Size in bytes of file
+
+    Returns
+    -------
+    ioresult : dict
+        Result of the transfer, with keys:
+        "ret": int
+            0 if copy succeeded; 1 if it failed
+        "md5sum": str
+            Only present if ret == 0: md5sum of the file computed via
+            `util.md5sum_file`
+        "stderr": str
+            Only present if ret == 1: a message indicating what went wrong.
+        "check_src": bool
+            True unless it's clear that a failure wasn't
+            due to a problem with the source file.
+    """
+
+    from_path = pathlib.Path(from_path)
+    dest_path = pathlib.Path(to_dir, filename)
+
+    # We create the copy in a temporary place so that the destination filename
+    # never points to a partially transferred file.
+    try:
+        # Create a temporary directory as a subdirectory of the destination dir
+        with TemporaryDirectory(dir=to_dir, prefix=".alpentemp") as tmpdir:
+            # Timeout for the pull
+            timeout = _pull_timeout(size_b)
+
+            # If no timeout, just directly call shutil
+            if timeout is None:
+                tmp_path = shutil.copy2(from_path, tmpdir)
+            else:
+                # Otherwise copy the source into the dest, with timeout.
+                # Raises OSError or TimeoutError on failure
+                tmp_path = util.timeout_call(shutil.copy2, timeout, from_path, tmpdir)
+
+            # Copy succeeded!  Overwrite any existing file atomically
+            pathlib.Path(tmp_path).rename(dest_path)
+
+        # Now MD5 the file to verify it.
+        log.info(f'verifying "{dest_path}" after local copy')
+        md5 = util.md5sum_file(dest_path)
+    except (OSError, TimeoutError) as e:
+        # Copy failed for some reason
+        log.warning(f"local copy failed: {e}")
+        return {"ret": 1, "stderr": str(e)}
+
+    # Copy succeeded
+    return {"ret": 0, "md5sum": md5}
+
+
 def post_add(node: StorageNode, file_: ArchiveFile) -> None:
     """Run any actions after adding `file` to `node`.
 


Test patch:
diff --git a/tests/io/test_defaultnode_pull.py b/tests/io/test_defaultnode_pull.py
index 4af4d5c..8b681dd 100644
--- a/tests/io/test_defaultnode_pull.py
+++ b/tests/io/test_defaultnode_pull.py
@@ -2,7 +2,7 @@
 
 import os
 import pathlib
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -345,16 +345,16 @@ def test_pull_async_link_arccontam(queue, pull_async):
 
     # Call the async
     task, key = queue.get()
-    task()
-    queue.task_done(key)
 
-    # Req isn't resolved because we have no rsync, so failure
-    afcr = ArchiveFileCopyRequest.get(id=req.id)
-    assert afcr.completed is False
-    assert afcr.cancelled is False
+    # Mock local_copy.  We'll end up falling back on this
+    # due to no other copy method working
+    mock = MagicMock()
+    with patch("alpenhorn.io.ioutil.local_copy", mock):
+        task()
+    queue.task_done(key)
 
-    # Source is not being re-checked
-    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+    # Local copy happened.
+    mock.assert_called_once()
 
 
 @pytest.mark.run_command_result(0, "", "stderr")
@@ -413,7 +413,8 @@ def test_pull_async_local_rsync_fail(queue, have_rsync, pull_async):
     assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file == "M"
 
 
-def test_pull_fail_unlink(xfs, queue, pull_async):
+@pytest.mark.run_command_result(1, "", "rsync_stderr")
+def test_pull_fail_unlink(xfs, queue, have_rsync, pull_async):
     """Test failure deleting the destination."""
 
     node, req = pull_async
diff --git a/tests/io/test_ioutil.py b/tests/io/test_ioutil.py
index 3c765ee..34df62b 100644
--- a/tests/io/test_ioutil.py
+++ b/tests/io/test_ioutil.py
@@ -208,57 +208,51 @@ def test_rsync_fail(mock_run_command):
     }
 
 
-def test_hardlink(tmp_path):
+def test_hardlink(xfs):
     """Test successful ioutil.hardlink() call."""
 
-    # Create src and dest in a temporary directory
-    srcdir = tmp_path.joinpath("src")
-    srcdir.mkdir()
-    file = srcdir.joinpath("file")
-    file.write_text("data")
-    dstdir = tmp_path.joinpath("dst")
-    dstdir.mkdir()
+    # Create src and dest
+    file = "/src/file"
+    xfs.create_file(file, contents="data")
+    dstdir = pathlib.Path("/dest")
+    xfs.create_dir(dstdir)
     destfile = dstdir.joinpath("file")
 
     assert ioutil.hardlink(file, dstdir, "file") == {"ret": 0, "md5sum": True}
     assert destfile.read_text() == "data"
 
 
-def test_hardlink_clobber(tmp_path):
+def test_hardlink_clobber(xfs):
     """Test successful overwrite in ioutil.hardlink() call."""
 
-    # Create src and dest in a temporary directory
-    srcdir = tmp_path.joinpath("src")
-    srcdir.mkdir()
-    file = srcdir.joinpath("file")
-    file.write_text("data")
-    dstdir = tmp_path.joinpath("dst")
-    dstdir.mkdir()
+    # Create src and dest
+    file = "/src/file"
+    xfs.create_file(file, contents="data")
+    dstdir = pathlib.Path("/dest")
     destfile = dstdir.joinpath("file")
-    destfile.write_text("other_data")
+    xfs.create_file(destfile, contents="other_data")
 
     assert destfile.read_text() == "other_data"
     assert ioutil.hardlink(file, dstdir, "file") == {"ret": 0, "md5sum": True}
     assert destfile.read_text() == "data"
 
 
-def test_hardlink_fail(tmp_path):
+def test_hardlink_fail(xfs):
     """Test failed ioutil.hardlink() call."""
 
-    # Create src but not dest
-    srcdir = tmp_path.joinpath("src")
-    srcdir.mkdir()
-    file = srcdir.joinpath("file")
-    file.write_text("data")
-    dstdir = tmp_path.joinpath("dst")
+    # Create src but not destdir
+    file = "/src/file"
+    xfs.create_file(file, contents="data")
+    dstdir = pathlib.Path("/dest")
     destfile = dstdir.joinpath("file")
 
     assert ioutil.hardlink(file, dstdir, "file") is None
     with pytest.raises(FileNotFoundError):
         destfile.read_text()
 
-    # Try with access error instead
-    dstdir.mkdir(mode=0o400)
+    # Try with access error instead.
+    xfs.create_file(destfile, contents="other_data")
+    xfs.chmod(dstdir, 0o400)
 
     assert ioutil.hardlink(file, dstdir, "file") is None
     with pytest.raises(PermissionError):
@@ -487,3 +481,66 @@ def test_remove_filedir_nonempty(simplenode, xfs):
     assert pathlib.Path(f"{simplenode.root}/a/b").exists()
     assert pathlib.Path(f"{simplenode.root}/a").exists()
     assert pathlib.Path(simplenode.root).exists()
+
+
+def test_local_copy(xfs, set_config):
+    """Test successful ioutil.local_copy() call."""
+
+    # Create src and dest
+    file = "/src/file"
+    xfs.create_file(file, contents="data")
+    dstdir = pathlib.Path("/dest")
+    xfs.create_dir(dstdir)
+    destfile = dstdir.joinpath("file")
+
+    assert ioutil.local_copy(file, dstdir, "file", 4) == {
+        "ret": 0,
+        "md5sum": "8d777f385d3dfec8815d20f7496026dc",
+    }
+    assert destfile.read_text() == "data"
+
+
+def test_local_copy_clobber(xfs, set_config):
+    """Test successful overwrite in ioutil.local_copy() call."""
+
+    # Create src and dest
+    file = "/src/file"
+    xfs.create_file(file, contents="data")
+    dstdir = pathlib.Path("/dest")
+    destfile = dstdir.joinpath("file")
+    xfs.create_file(destfile, contents="other_data")
+
+    assert destfile.read_text() == "other_data"
+    assert ioutil.local_copy(file, dstdir, "file", 4) == {
+        "ret": 0,
+        "md5sum": "8d777f385d3dfec8815d20f7496026dc",
+    }
+    assert destfile.read_text() == "data"
+
+
+def test_local_copy_fail(xfs, set_config):
+    """Test failed ioutil.local_copy() call."""
+
+    # Create src but not destdir
+    file = "/src/file"
+    xfs.create_file(file, contents="data")
+    dstdir = pathlib.Path("/dest")
+    destfile = dstdir.joinpath("file")
+
+    result = ioutil.local_copy(file, dstdir, "file", 4)
+    assert result["ret"] != 0
+    assert "stderr" in result
+
+    with pytest.raises(FileNotFoundError):
+        destfile.read_text()
+
+    # Try with access error instead.
+    xfs.create_file(destfile, contents="other_data")
+    xfs.chmod(dstdir, 0o400)
+
+    result = ioutil.local_copy(file, dstdir, "file", 4)
+    assert result["ret"] != 0
+    assert "stderr" in result
+
+    with pytest.raises(PermissionError):
+        destfile.read_text()