Repository: radiocosmology/alpenhorn

Issue:
Copy request fails if `node_from` has a NULL `address`
Line 304 in [`update.py`](https://github.com/radiocosmology/alpenhorn/blob/master/alpenhorn/update.py#L304) is filtering out my list of archive file copy requests to nothing:

```py
    requests = requests.join(st.StorageNode).where(st.StorageNode.address != 'HPSS')
```

It's probably because the request's `node_from.address` is NULL. When I remounted the node with an explicit `--address` argument, the copy went ahead fine.

Discussion:

Patch:
diff --git a/alpenhorn/io/_default_asyncs.py b/alpenhorn/io/_default_asyncs.py
index 12665ad..e991885 100644
--- a/alpenhorn/io/_default_asyncs.py
+++ b/alpenhorn/io/_default_asyncs.py
@@ -2,19 +2,167 @@
 from __future__ import annotations
 from typing import TYPE_CHECKING
 
+import time
 import errno
+import shutil
 import logging
+import pathlib
 from datetime import datetime
 
-from ..archive import ArchiveFileCopy
+from . import ioutil
+from ..archive import ArchiveFileCopy, ArchiveFileCopyRequest
+from ..update import RemoteNode
 
 if TYPE_CHECKING:
     from .base import BaseNodeIO
+    from .updownlock import UpDownLock
     from ..task import Task
 
 log = logging.getLogger(__name__)
 
 
+def pull_async(
+    task: Task, io: BaseNodeIO, tree_lock: UpDownLock, req: ArchiveFileCopyRequest
+) -> None:
+    """Fulfill `req` by pulling a file onto the local node.
+
+    Things to try:
+        - hard link (for nodes on the same filesystem)
+        - bbcp (if source is not on this host)
+        - rsync if all else fails
+
+    Parameters
+    ----------
+    task : Task
+        The task instance containing this async.
+    io : Node I/O instance
+        The I/O instance for the pull destination node.
+    tree_lock : UpDownLock
+        The directory tree modificiation lock.
+    req : ArchiveFileCopyRequest
+        The request we're fulfilling.
+    """
+
+    # Before we were queued, NodeIO reserved space for this file.
+    # Automatically release bytes on task completion
+    task.on_cleanup(io.release_bytes, args=(req.file.size_b,))
+
+    # We know dest is local, so if source is too, this is a local transfer
+    local = req.node_from.local
+
+    # The Remote Node
+    remote = RemoteNode(req.node_from)
+
+    # Source spec
+    if local:
+        from_path = remote.io.file_path(req.file)
+    else:
+        try:
+            from_path = remote.io.file_addr(req.file)
+        except ValueError:
+            log.warning(
+                f"Skipping request for {req.file.path} "
+                f"due to unconfigured route to host for node {req.node_from.name}."
+            )
+            return
+
+    to_file = pathlib.Path(io.node.root, req.file.path)
+    to_dir = to_file.parent
+
+    # Placeholder file
+    placeholder = pathlib.Path(to_dir, f".{to_file.name}.placeholder")
+
+    # Create directories.  This must be done while locking up the tree lock
+    with tree_lock.up:
+        if not to_dir.exists():
+            log.info(f'Creating directory "{to_dir}".')
+            to_dir.mkdir(parents=True)
+
+        # If the file doesn't exist, create a placeholder so we can release
+        # the tree lock without having to wait for the transfer to complete
+        if not to_file.exists():
+            placeholder.touch(mode=0o600, exist_ok=True)
+
+    # Giddy up!
+    start_time = time.time()
+    log.info(f'Transferring file "{req.file.path}":')
+
+    # Attempt to transfer the file. Each of the methods below needs to return
+    # a dict with required key:
+    #  - ret : integer
+    #        return code (0 == success)
+    # optional keys:
+    #  - md5sum : string or True
+    #        If True, the sum is guaranteed to be right; otherwise, it's a
+    #        md5sum to check against the source.  Must be present if ret == 0
+    #  - stderr : string
+    #        if given, printed to the log when ret != 0
+    #  - check_src : bool
+    #        if given and False, the source file will _not_ be marked suspect
+    #        when ret != 0; otherwise, a failure results in a source check
+
+    # First we need to check if we are copying over the network
+    if not local:
+        if shutil.which("bbcp") is not None:
+            # First try bbcp which is a fast multistream transfer tool. bbcp can
+            # calculate the md5 hash as it goes, so we'll do that to save doing
+            # it at the end.
+            ioresult = ioutil.bbcp(from_path, to_dir, req.file.size_b)
+        elif shutil.which("rsync") is not None:
+            # Next try rsync over ssh.
+            ioresult = ioutil.rsync(from_path, to_dir, req.file.size_b, local)
+        else:
+            # We have no idea how to transfer the file...
+            log.error("No commands available to complete remote pull.")
+            ioresult = {"ret": -1, "check_src": False}
+
+    else:
+        # Okay, great we're just doing a local transfer.
+
+        # First try to just hard link the file. This will only work if we
+        # are on the same filesystem.  If it didn't work, ioresult will be None
+        #
+        # But don't do this if it creates a hardlink between an archive node and
+        # a non-archive node
+        if req.node_from.archive == io.node.archive:
+            ioresult = ioutil.hardlink(from_path, to_dir, req.file.name)
+        else:
+            ioresult = None
+
+        # If we couldn't just link the file, try copying it with rsync.
+        if ioresult is None:
+            if shutil.which("rsync") is not None:
+                ioresult = ioutil.rsync(from_path, to_dir, req.file.size_b, local)
+            else:
+                log.error("No commands available to complete local pull.")
+                ioresult = {"ret": -1, "check_src": False}
+
+    # Delete the placeholder, if we created it
+    placeholder.unlink(missing_ok=True)
+
+    if not ioutil.copy_request_done(
+        req,
+        io,
+        check_src=ioresult.get("check_src", True),
+        md5ok=ioresult.get("md5sum", None),
+        start_time=start_time,
+        stderr=ioresult.get("stderr", None),
+        success=(ioresult["ret"] == 0),
+    ):
+        # Remove file, on error
+        try:
+            to_file.unlink(missing_ok=True)
+        except OSError as e:
+            log.error(f"Error removing corrupt file {to_file}: {e}")
+
+    # Whatever has happened, update free space, if possible
+    new_avail = io.bytes_avail(fast=True)
+
+    # This was a fast update, so don't save "None" to the database
+    if new_avail is not None:
+        io.node.update_avail_gb(new_avail)
+
+
 def check_async(task: Task, io: BaseNodeIO, copy: ArchiveFileCopy) -> None:
     """Check a file copy.  This is asynchronous.
 
@@ -74,7 +222,9 @@ def check_async(task: Task, io: BaseNodeIO, copy: ArchiveFileCopy) -> None:
     copy.save()
 
 
-def delete_async(task: Task, copies: list[ArchiveFileCopy]) -> None:
+def delete_async(
+    task: Task, tree_lock: UpDownLock, copies: list[ArchiveFileCopy]
+) -> None:
     """Delete some file copies, if possible.
 
     Copies are only deleted if sufficient archived copies exist
@@ -86,6 +236,8 @@ def delete_async(task: Task, copies: list[ArchiveFileCopy]) -> None:
     ----------
     task : Task
         The task instance containing this async.
+    tree_lock : UpDownLock
+        The directory tree modificiation lock.
     copies : list of ArchiveFileCopy
         The list of copies to delete.  Never empty.
     """
@@ -128,23 +280,26 @@ def delete_async(task: Task, copies: list[ArchiveFileCopy]) -> None:
         # and remove if they are.
         dirname = fullpath.parent
 
-        # try to delete the directories.
-        while dirname != ".":
-            try:
-                dirname.rmdir()
-                log.info(f"Removed directory {dirname} on {name}")
-            except OSError as e:
-                if e.errno == errno.ENOTEMPTY:
-                    # This is fine, but stop trying to rmdir.
-                    break
-                elif e.errno == errno.ENOENT:
-                    # Already deleted, which is fine.
-                    pass
-                else:
-                    log.warning(f"Error deleting directory {dirname} on {name}: {e}")
-                    # Otherwise, let's try to soldier on
-
-            dirname = dirname.parent
+        # try to delete the directories.  This must be done while locking down the tree lock
+        with tree_lock.down:
+            while dirname != ".":
+                try:
+                    dirname.rmdir()
+                    log.info(f"Removed directory {dirname} on {name}")
+                except OSError as e:
+                    if e.errno == errno.ENOTEMPTY:
+                        # This is fine, but stop trying to rmdir.
+                        break
+                    elif e.errno == errno.ENOENT:
+                        # Already deleted, which is fine.
+                        pass
+                    else:
+                        log.warning(
+                            f"Error deleting directory {dirname} on {name}: {e}"
+                        )
+                        # Otherwise, let's try to soldier on
+
+                dirname = dirname.parent
 
         # Update the DB
         ArchiveFileCopy.update(
diff --git a/alpenhorn/io/base.py b/alpenhorn/io/base.py
index 98c287d..2004a25 100644
--- a/alpenhorn/io/base.py
+++ b/alpenhorn/io/base.py
@@ -10,10 +10,11 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 import logging
+import pathlib
 
 if TYPE_CHECKING:
-    import pathlib
-    from ..archive import ArchiveFileCopy
+    from ..acquisition import ArchiveFile
+    from ..archive import ArchiveFileCopy, ArchiveFileCopyRequest
     from ..queue import FairMultiFIFOQueue
     from ..storage import StorageNode, StorageGroup
     from ..update import UpdateableNode
@@ -21,6 +22,91 @@ if TYPE_CHECKING:
 log = logging.getLogger(__name__)
 
 
+# Comment from DVW:
+#
+# Separating BaseNodeIO and BaseNodeRemote is primarily to avoid the temptaiton
+# of accidentally writing code that tries to perform I/O operations on non-local
+# nodes.
+class BaseNodeRemote:
+    """Base I/O class for a remote StorageNode.
+
+    The remote I/O modules provide read-only information about a possibly
+    non-local StorageNode.
+
+    Parameters
+    ----------
+    node : StorageNode
+        the remote node
+    config : dict
+        the parsed `node.io_config`. If `node.io_config` is None,
+        this is an empty `dict`.
+    """
+
+    def __init__(self, node: StorageNode, config: dict) -> None:
+        self.node = node
+        self.config = config
+
+    def file_addr(self, file: ArchiveFile) -> str:
+        """Return an remote file address suitable for use with rsync.
+
+        i.e., a string of the form: <username>@<host>:<path>
+
+        Parameters
+        ----------
+        file : ArchiveFile
+            The file to return the address for.
+
+        Returns
+        -------
+        addr : str
+            The file address
+
+        Raises
+        ------
+        ValueError
+            `node.username` or `node.address` were not set.
+        """
+        if self.node.username is None:
+            raise ValueError("missing username")
+        if self.node.address is None:
+            raise ValueError("missing address")
+
+        return f"{self.node.username}@{self.node.address}:{self.file_path(file)}"
+
+    def file_path(self, file: ArchiveFile) -> str:
+        """Return a path on the remote node pointing to ArchiveFile `file`.
+
+        By default, returns the path contcatenation of `node.root` and
+        `file.path`.
+
+        Parameters:
+        -----------
+        file : ArchiveFile
+            The file to return the path for.
+
+        Returns:
+        --------
+        path : str
+            the remote path
+        """
+        return str(pathlib.PurePath(self.node.root, file.path))
+
+    def pull_ready(self, file: ArchiveFile) -> bool:
+        """Is `file` ready for pulling from this remote node?
+
+        Parameters
+        ----------
+        file : ArchiveFile
+            the file being checked
+
+        Returns
+        -------
+        ready : bool
+            True if `file` is ready on the node; False otherwise.
+        """
+        raise NotImplementedError("method must be re-implemented in subclass.")
+
+
 class BaseNodeIO:
     """Base class for StorageNode I/O modules in alpenhorn.
 
@@ -37,6 +123,9 @@ class BaseNodeIO:
 
     # SETUP
 
+    # Subclasses should set this to a BaseNodeRemote-derived class.
+    remote_class = BaseNodeRemote
+
     def __init__(
         self, node: StorageNode, config: dict, queue: FairMultiFIFOQueue
     ) -> None:
@@ -169,6 +258,16 @@ class BaseNodeIO:
         """
         raise NotImplementedError("method must be re-implemented in subclass.")
 
+    def exists(self, path: pathlib.PurePath) -> bool:
+        """Does `path` exist?
+
+        Parameters
+        ----------
+        path : pathlib.PurePath
+            path relative to `node.root`
+        """
+        raise NotImplementedError("method must be re-implemented in subclass.")
+
     def filesize(self, path: pathlib.Path, actual: bool = False) -> int:
         """Return size in bytes of the file given by `path`.
 
@@ -201,6 +300,37 @@ class BaseNodeIO:
         """
         raise NotImplementedError("method must be re-implemented in subclass.")
 
+    def pull(self, req: ArchiveFileCopyRequest) -> None:
+        """Pull file specified by copy request `req` onto `self.node`.
+
+        In this case, `node` is the destination.
+
+        Parameters
+        ----------
+        req : ArchiveFileCopyRequest
+            the copy request to fulfill.  We are the destination node (i.e.
+            `req.group_to == self.node.group`).
+        """
+        raise NotImplementedError("method must be re-implemented in subclass.")
+
+    def ready_pull(self, req: ArchiveFileCopyRequest) -> None:
+        """Ready a file to be pulled as specified by `req`.
+
+        This method is called for all pending requests, even ones that are
+        impossible due to the file being corrupt, missing, or some other calamity.
+
+        If such an impossibility arises, this method _may_ cancel the request,
+        but that's not required.  (It's the responsibility of the pulling
+        alpenhornd to resolve the request.)
+
+        Parameters
+        ----------
+        req : ArchiveFileCopyRequest
+            the copy request to ready.  We are the source node (i.e.
+            `req.node_from == self.node`).
+        """
+        raise NotImplementedError("method must be re-implemented in subclass.")
+
 
 class BaseGroupIO:
     """Base class for StorageGroup IO modules in alpenhorn.
@@ -266,7 +396,8 @@ class BaseGroupIO:
         Parameters
         ----------
         idle : boolean
-                If False, the group update is going to be skipped.
+                True if all the `nodes` were idle when the current
+                update loop started.
 
         Returns
         -------
@@ -299,3 +430,41 @@ class BaseGroupIO:
         """
         # Do nothing
         pass
+
+    # I/O METHODS
+
+    def exists(self, path: pathlib.PurePath) -> UpdateableNode | None:
+        """Check whether the file `path` exists in this group.
+
+        If the file exists on more than one node in the group,
+        implementations may use any method to choose which node
+        to return.
+
+        Parameters
+        ----------
+        path : pathlib.PurePath
+            the path, relative to a node `root` of the file to
+            search for.
+
+        Returns
+        -------
+        node : UpdateableNode or None
+            If the file exists, the node containing it.  This should be
+            one of the `UpdateableNode` instances provided to `set_nodes`.
+            If the file doesn't exist in the group, this is None.
+        """
+        raise NotImplementedError("method must be re-implemented in subclass.")
+
+    def pull(self, req: ArchiveFileCopyRequest) -> None:
+        """Handle ArchiveFileCopyRequest `req` by pulling to this group.
+
+        In general, implementations should choose which node in the group
+        to handle the request, and pass `req` to that node's `io.pull` method.
+
+        Parameters
+        ----------
+        req : ArchiveFileCopyRequest
+            the request to fulfill.  We are the destination group (i.e.
+            `req.group_to == self.group`).
+        """
+        raise NotImplementedError("method must be re-implemented in subclass.")
diff --git a/alpenhorn/io/default.py b/alpenhorn/io/default.py
index 2461fa4..a33837d 100644
--- a/alpenhorn/io/default.py
+++ b/alpenhorn/io/default.py
@@ -13,24 +13,69 @@ from typing import TYPE_CHECKING
 import os
 import logging
 import pathlib
+import threading
 
+from .base import BaseNodeIO, BaseGroupIO, BaseNodeRemote
+from .updownlock import UpDownLock
 from .. import util
-from .base import BaseNodeIO, BaseGroupIO
 from ..task import Task
 
 # The asyncs are over here:
-from ._default_asyncs import check_async, delete_async
+from ._default_asyncs import pull_async, check_async, delete_async
 
 if TYPE_CHECKING:
-    from ..archive import ArchiveFileCopy
+    from ..acquisition import ArchiveFile
+    from ..archive import ArchiveFileCopy, ArchiveFileCopyRequest
+    from ..queue import FairMultiFIFOQueue
+    from ..storage import StorageNode
     from ..update import UpdateableNode
 
 log = logging.getLogger(__name__)
 
+# Reserved byte counts are stored here, indexed by node name and protected
+# by the mutex
+_mutex = threading.Lock()
+_reserved_bytes = dict()
+
+
+class DefaultNodeRemote(BaseNodeRemote):
+    """I/O class for a remote DefaultIO StorageNode."""
+
+    def pull_ready(self, file: ArchiveFile) -> bool:
+        """Is `file` ready for pulling from this remote node?
+
+        Parameters
+        ----------
+        file : ArchiveFile
+            the file being checked
+
+        Returns
+        -------
+        True
+            Files on Default nodes are always ready.
+        """
+        return True
+
 
 class DefaultNodeIO(BaseNodeIO):
     """A simple StorageNode backed by a regular POSIX filesystem."""
 
+    # SETUP
+
+    remote_class = DefaultNodeRemote
+
+    def __init__(
+        self, node: StorageNode, queue: FairMultiFIFOQueue, config: dict
+    ) -> None:
+        super().__init__(node, queue, config)
+
+        # The directory tree modification lock
+        self.tree_lock = UpDownLock()
+
+        # Set up a reservation for ourself if necessary
+        with _mutex:
+            _reserved_bytes.setdefault(node.name, 0)
+
     # I/O METHODS
 
     def bytes_avail(self, fast: bool = False) -> int | None:
@@ -114,12 +159,22 @@ class DefaultNodeIO(BaseNodeIO):
             func=delete_async,
             queue=self._queue,
             key=self.node.name,
-            args=(copies,),
+            args=(self.tree_lock, copies),
             name="Delete copies "
             + str([copy.id for copy in copies])
             + f" from {self.node.name}",
         )
 
+    def exists(self, path: pathlib.PurePath) -> bool:
+        """Does `path` exist?
+
+        Parameters
+        ----------
+        path : pathlib.PurePath
+            path relative to `node.root`
+        """
+        return pathlib.Path(self.node.root, path).is_file()
+
     def filesize(self, path: pathlib.Path, actual: bool = False) -> int:
         """Return size in bytes of the file given by `path`.
 
@@ -163,6 +218,109 @@ class DefaultNodeIO(BaseNodeIO):
         """
         return util.md5sum_file(pathlib.Path(self.node.root, path, *segments))
 
+    def pull(self, req: ArchiveFileCopyRequest) -> None:
+        """Pull file specified by copy request `req` onto `self.node`.
+
+        Most of the work happens in an asynchronous I/O task.
+
+        Parameters
+        ----------
+        req : ArchiveFileCopyRequest
+            the copy request to fulfill.  We are the destination node (i.e.
+            `req.group_to == self.node.group`).
+        """
+
+        if self.node.under_min:
+            log.info(
+                f"Skipping pull for StorageNode {self.node.name}: hit minimum free space: "
+                f"({self.node.avail_gb:.2f} GiB < {self.node.min_avail_gb:.2f} GiB)"
+            )
+            return
+
+        if self.node.check_over_max():
+            log.info(
+                f"Skipping pull for StorageNode {self.node.name}: node full. "
+                f"({self.node.get_total_gb():.2f} GiB >= {self.node.max_total_gb:.2f} GiB)"
+            )
+            return
+
+        # Check that there is enough space available (and reserve what we need)
+        if not self.reserve_bytes(req.file.size_b):
+            log.warning(
+                f"Skipping request for {req.file.acq.name}/{req.file.name}: "
+                f"insufficient space on node {self.node.name}."
+            )
+            return
+
+        Task(
+            func=pull_async,
+            queue=self._queue,
+            key=self.node.name,
+            args=(self, self.tree_lock, req),
+            name=f"AFCR#{req.id}: {req.node_from.name} -> {self.node.name}",
+        )
+
+    # This is the reservation fudge factor.  XXX Is it correct?
+    reserve_factor = 2
+
+    def release_bytes(self, size: int) -> None:
+        """Release space previously reserved with `reserve_bytes`.
+
+        Parameters
+        ----------
+        size : integer
+            the number of bytes to release
+
+        Raises
+        ------
+        ValueError
+            `size` was greater than the total amount of reserved
+            space.
+        """
+        size *= self.reserve_factor
+        with _mutex:
+            if _reserved_bytes[self.node.name] < size:
+                raise ValueError(
+                    f"attempted to release too many bytes: {_reserved_bytes[self.node.name]} < {size}"
+                )
+            _reserved_bytes[self.node.name] -= size
+
+    def reserve_bytes(self, size: int, check_only: bool = False) -> bool:
+        """Attempt to reserve `size` bytes of space on the filesystem.
+
+        Parameters
+        ----------
+        size : int
+            the number of bytes to reserve
+        check_only : bool, optional
+            If True, no reservation is made, and the only effect is
+            the return value.
+
+        Returns
+        -------
+        success : bool
+            False if there was insufficient space to make the reservation.
+            True otherwise.
+        """
+        size *= self.reserve_factor
+        with _mutex:
+            bavail = self.bytes_avail()
+            if bavail is not None and bavail - _reserved_bytes[self.node.name] < size:
+                return False  # Insufficient space
+
+            if not check_only:
+                _reserved_bytes[self.node.name] += size
+
+            return True
+
+    def ready_pull(self, req: ArchiveFileCopyRequest) -> None:
+        """Ready a file to be pulled as specified by `req`.
+
+        This method does nothing: DefaultIO file copies are
+        always ready.
+        """
+        pass
+
 
 class DefaultGroupIO(BaseGroupIO):
     """A simple StorageGroup.
@@ -200,3 +358,39 @@ class DefaultGroupIO(BaseGroupIO):
 
         self.node = nodes[0]
         return nodes
+
+    # I/O METHODS
+
+    def exists(self, path: pathlib.PurePath) -> UpdateableNode | None:
+        """Check whether the file `path` exists in this group.
+
+
+        Parameters
+        ----------
+        path : pathlib.PurePath
+            the path, relative to node `root` of the file to
+            search for.
+
+        Returns
+        -------
+        node : UpdateableNode or None
+            If the file exists, returns the node in the group.
+            If the file doesn't exist in the group, this is None.
+        """
+        if self.node.io.exists(path):
+            return self.node
+
+        return None
+
+    def pull(self, req: ArchiveFileCopyRequest) -> None:
+        """Handle ArchiveFileCopyRequest `req` by pulling to this group.
+
+        Simply passes the `req` on to the node in the group.
+
+        Parameters
+        ----------
+        req : ArchiveFileCopyRequest
+            the request to fulfill.  We are the destination group (i.e.
+            `req.group_to == self.group`).
+        """
+        self.node.io.pull(req)
diff --git a/alpenhorn/io/updownlock.py b/alpenhorn/io/updownlock.py
new file mode 100644
index 0000000..185e604
--- /dev/null
+++ b/alpenhorn/io/updownlock.py
@@ -0,0 +1,267 @@
+"""UpDownLock: A two-way threading lock.
+
+This is a shared two-way lock used for thread synchronization.
+
+This lock behaves similarly to a regular `threading.RLock`, but
+can be in one of three states: locked up, locked down, or
+unlocked.
+
+All threads wanting to acquire the lock must choose one of the
+two locked states to acquire the lock in.  When the lock is
+in one of the locked states, any other threads wanting to
+acquire the lock in that same state may do so (i.e. multiple
+threads can hold the lock in a particular state), but threads
+wanting the other lock state must wait for the lock to become
+unlocked before they can aquire the lock.
+
+Alpenhorn uses this lock to prevent race conditions while
+modifying the directory tree of a StorageNode: threads which
+are in the process of creating directories must do so while
+holding the up lock and threads which are in the process of
+removing directories must do so while holding the down lock.
+
+The lock ensures only one operation (creation or removal) can
+happen at any given time.
+"""
+
+import time
+import threading
+from collections import defaultdict
+
+
+class _UpDownLock:
+    """UpDownLock internals.
+
+    Used to prevent a reference loop.
+    """
+
+    __slots__ = ["count", "_lock", "_owners", "_is_unlocked"]
+
+    def __init__(self) -> None:
+        # This tracks the state of the lock:
+        #  > 0: the lock is in the "up" state
+        #  = 0: the lock is unlocked
+        #  < 0: the lock is in the "down" state
+        #
+        # The absolute value indicates how many times
+        # the lock is held
+        self.count = 0
+
+        # Protects self._count
+        self._lock = threading.Lock()
+
+        # Dict of threads holding the lock.  Keys are the
+        # thread IDs.  Values are how many times that thread
+        # owns the lock.
+        self._owners = defaultdict(int)
+
+        # Is the lock unlocked?  Handles threads waiting for
+        # unlock.
+        self._is_unlocked = threading.Condition(threading.RLock())
+
+    def acquire(self, blocking: bool, timeout: float, is_down: bool) -> bool:
+        """Acquire the lock in the specified state."""
+
+        now = time.monotonic()
+        me = threading.get_ident()
+
+        # Try to acquire the lock
+        with self._lock:
+            if is_down:
+                ok_to_lock = self.count <= 0
+            else:
+                ok_to_lock = self.count >= 0
+
+            if ok_to_lock:
+                if is_down:
+                    self.count -= 1
+                else:
+                    self.count += 1
+                self._owners[me] += 1
+                return True
+
+            # If we already hold the lock in the other state, fail
+            if self._owners[me] > 0:
+                raise RuntimeError("Can't acquire both locking states.")
+
+        # If not blocking, fail
+        if not blocking:
+            return False
+
+        # Otherwise, wait until unlocked
+        with self._is_unlocked:
+            if timeout < 0:
+                self._is_unlocked.wait()
+            else:
+                end_at = None
+                notified = False
+
+                # Wait until either we're notified or else we timeout
+                while not notified and (end_at is None or time.monotonic() < end_at):
+                    # We do it this way to ensure this loop runs at least once
+                    if end_at is None:
+                        end_at = now + timeout
+
+                    # This might be negative; threading.condition is okay
+                    # with that (essentially it makes wait() non-blocking)
+                    timeout = end_at - time.monotonic()
+                    notified = self._is_unlocked.wait(timeout)
+
+                # If we're out of time, fail
+                if not notified:
+                    return False
+
+                # If we were notfied, but we're also out of time, convert to
+                # non blocking, to try one last time
+                if timeout <= 0:
+                    blocking = False
+
+        # If we got here, we were notified of the lock unlocking; try again,
+        # possibly with a reduced timeout and/or converted to non-blocking
+        return self.acquire(blocking, timeout, is_down)
+
+    def release(self, is_down: bool) -> None:
+        """Release the lock with the given state.
+
+        Raises RuntimeError if the lock was not held in the state.
+        """
+
+        me = threading.get_ident()
+
+        with self._lock:
+            if is_down:
+                ok_to_unlock = self.count < 0
+            else:
+                ok_to_unlock = self.count > 0
+
+            # Check ownership
+            if ok_to_unlock:
+                if self._owners[me] == 0:
+                    ok_to_unlock = False
+
+            if not ok_to_unlock:
+                raise RuntimeError(
+                    f"Lock not held in {'down' if is_down else 'up'} state."
+                )
+
+            # Otherwise, unlock
+            if is_down:
+                self.count += 1
+            else:
+                self.count -= 1
+            self._owners[me] -= 1
+
+            # If we're now unlocked, notifiy waiters
+            if self.count == 0:
+                with self._is_unlocked:
+                    self._is_unlocked.notify_all()
+
+
+class _UpDownAccessor:
+    """Lock accessor for one of the lock states.
+
+    To acquire the lock in this state, use the `acquire` method.
+
+    To release the lock in this state, use the `release` method.
+
+    This accessor can also be used as a context manager.
+    """
+
+    __slots__ = ["_is_down", "_internals"]
+
+    def __init__(self, is_down: bool, internals: _UpDownLock) -> None:
+        self._is_down = is_down
+        self._internals = internals
+
+    def acquire(self, blocking: bool = True, timeout: float = -1) -> bool:
+        """Acquire the lock once.
+
+        Parameters
+        ----------
+        blocking : bool
+            If True, block until the lock can be acquired (or timeout expires).
+            If False, this method immediately returns False if the lock cannot
+            be acquired.
+        timeout : float
+            If set to a positive value, and `blocking` is True, wait at most
+            `timeout` seconds before failing to acquire the lock (returning
+            False).
+
+        Returns
+        -------
+        success : bool
+            True if the lock was acquired.  False otherwise.
+
+        Raises
+        ------
+        RuntimeError:
+            Lock is already being held in the opposite state.
+        """
+        return self._internals.acquire(
+            blocking=blocking, timeout=timeout, is_down=self._is_down
+        )
+
+    def release(self) -> None:
+        """Release the lock once.
+
+        Raises
+        ------
+        RuntimeError:
+            The lock was not being held in this state.
+        """
+        self._internals.release(is_down=self._is_down)
+
+    # Context manager
+    __enter__ = acquire
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.release()
+        return False  # Re-raise exception, if any
+
+
+class UpDownLock:
+    """A two-way thread lock.
+
+    This is a threading synchronization primitive, similar to
+    a standard `threading.RLock`, but one which can be locked in
+    two different states ("up" and "down).
+
+    When locked in one of the states, any number of threads can
+    hold the lock in that state, but threads attempting to
+    acquire the lock in the other state must wait for the
+    lock to unlock first.
+
+    If a thread holds the lock in one a particular state, it may
+    re-acquire the lock in that state again without blocking
+    (i.e. the lock is re-entrant in this case), but a thread holding
+    the lock in one state may not attempt to also acquire the lock
+    in the opposite state as well. (This will result in a
+    RuntimeError.) A thread must release the lock once for each time
+    it has acquired the lock.
+
+    The two states of the lock are accessed via `UpDownLock.up`
+    and `UpDownLock.down`. These accessors provide `acquire` and
+    `release` methods which work as they do with the standard
+    `threading.Lock`.  The accessors may also be used as context
+    managers.
+    """
+
+    __slots__ = ["up", "down", "_internals"]
+
+    def __init__(self) -> None:
+        self._internals = _UpDownLock()
+        self.up = _UpDownAccessor(is_down=False, internals=self._internals)
+        self.down = _UpDownAccessor(is_down=True, internals=self._internals)
+
+    def __repr__(self) -> str:
+        count = self._internals.count
+        if count == 0:
+            state = "unlocked"
+        elif count > 0:
+            state = "locked up"
+        else:
+            state = "locked down"
+        return (
+            f"<UpDownLock object state={state} count={abs(count)} "
+            f"at {hex(id(self))}>"
+        )
diff --git a/alpenhorn/update.py b/alpenhorn/update.py
index fc77458..7ad3f80 100644
--- a/alpenhorn/update.py
+++ b/alpenhorn/update.py
@@ -3,19 +3,13 @@
 from __future__ import annotations
 from typing import TYPE_CHECKING
 
-import os
 import json
 import time
 import logging
 import peewee as pw
-import datetime as dt
-from peewee import fn
 
-from . import acquisition as ac
-from . import archive as ar
-from . import storage as st
 from . import config, util
-from .archive import ArchiveFileCopy
+from .archive import ArchiveFileCopy, ArchiveFileCopyRequest
 from .extensions import io_module
 from .pool import global_abort, WorkerPool, EmptyPool
 from .storage import StorageNode, StorageGroup
@@ -25,12 +19,6 @@ if TYPE_CHECKING:
 
 log = logging.getLogger(__name__)
 
-# Parameters.
-max_time_per_node_operation = 300  # Don't let node operations hog time.
-
-# Globals.
-done_transport_this_cycle = False
-
 
 class updateable_base:
     """Abstract base class for UpdateableNode and UpdateableGroup.
@@ -191,6 +179,27 @@ class updateable_base:
         return False
 
 
+class RemoteNode(updateable_base):
+    """Remote storage node.
+
+    This represents a (potentially) non-local node used
+    as the source-side of a pull request.
+
+    Parameters
+    ----------
+    node : StorageNode
+        The underlying StorageNode instance
+    """
+
+    is_group = False
+
+    def __init__(self, node: StorageNode) -> None:
+        self.db = node
+        self.io_class = self._get_io_class()
+        config = self._parse_io_config(node.io_config)
+        self.io = self.io_class.remote_class(node, config)
+
+
 class UpdateableNode(updateable_base):
     """Updateable storage node
 
@@ -348,9 +357,13 @@ class UpdateableNode(updateable_base):
             # Delete any unwanted files to cleanup space
             self.update_delete()
 
-            # Process any regular transfers requests onto this node
-            # XXX Commented out until future PR.
-            # update_node_requests(node)
+            # Prepare files for pulls out from this node
+            for req in ArchiveFileCopyRequest.select().where(
+                ArchiveFileCopyRequest.completed == 0,
+                ArchiveFileCopyRequest.cancelled == 0,
+                ArchiveFileCopyRequest.node_from == self.db,
+            ):
+                self.io.ready_pull(req)
 
             self._updated = True
         else:
@@ -436,6 +449,119 @@ class UpdateableGroup(updateable_base):
 
         return True
 
+    def update_pull(self, req: ArchiveFileCopyRequest) -> None:
+        """Process pull request `req`.
+
+        Parameters
+        ----------
+        req : ArchiveFileCopyRequest
+            The pull request to process.
+        """
+        # What's the current situation on the destination?
+        copy_state = self.db.filecopy_state(req.file)
+        if copy_state == "Y":
+            # We mark the AFCR cancelled rather than complete becase
+            # _this_ AFCR clearly hasn't been responsible for creating
+            # the file copy.
+            log.info(
+                f"Cancelling pull request for "
+                f"{req.file.acq.name}/{req.file.name}: "
+                f"already present in group {self.name}."
+            )
+            ArchiveFileCopyRequest.update(cancelled=True).where(
+                ArchiveFileCopyRequest.id == req.id
+            ).execute()
+            return
+        elif copy_state == "M":
+            log.warning(
+                f"Skipping pull request for "
+                f"{req.file.acq.name}/{req.file.name}: "
+                f"existing copy in group {self.name} needs check."
+            )
+            return
+        elif copy_state == "X":
+            # If the file is corrupt, we continue with the
+            # pull to overwrite the corrupt file
+            pass
+        elif copy_state == "N":
+            # Check whether an actual file exists on the target.
+            node = self.io.exists(req.file.path)
+            if node is not None:
+                # file on disk: create/update the ArchiveFileCopy
+                # to force a check next pass
+                log.warning(
+                    f"Skipping pull request for "
+                    f"{req.file.acq.name}/{req.file.name}: "
+                    f"file already on disk in group {self.name}."
+                )
+                log.info(
+                    f"Requesting check of "
+                    f"{req.file.acq.name}/{req.file.name} on node "
+                    f"{node.name}."
+                )
+
+                # Update/create ArchiveFileCopy to force a check.
+
+                # ready == False is the safe option here: copy will be readied
+                # during the subsequent check if needed.
+                count = (
+                    ArchiveFileCopy.update(has_file="M", wants_file="Y", ready=False)
+                    .where(
+                        ArchiveFileCopy.file == req.file,
+                        ArchiveFileCopy.node == node.db,
+                    )
+                    .execute()
+                )
+                if count == 0:
+                    # Create new copy
+                    ArchiveFileCopy.create(
+                        file=req.file,
+                        node=node.db,
+                        has_file="M",
+                        wants_file="Y",
+                        ready=False,
+                        size_b=node.io.filesize(req.file.path, actual=True),
+                    )
+                return
+        else:
+            # Shouldn't get here
+            log.error(
+                f"Unexpected copy state: '{copy_state}' "
+                f"for file ID={req.file.id} in group {self.name}."
+            )
+            return
+
+        # Skip request unless the source node is active
+        if not req.node_from.active:
+            log.warning(
+                f"Skipping request for {req.file.acq.name}/{req.file.name}:"
+                f" source node {req.node_from.name} is not active."
+            )
+            return
+
+        # If the source file doesn't exist, skip the request.
+        #
+        # XXX Cancel instead?
+        if not req.node_from.filecopy_present(req.file):
+            log.warning(
+                f"Skipping request for {req.file.acq.name}/{req.file.name}:"
+                f" not available on node {req.node_from.name}. "
+                f"[file_id={req.file.id}]"
+            )
+            return
+
+        # If the source file is not ready, skip the request.
+        node_from = RemoteNode(req.node_from)
+        if not node_from.io.pull_ready(req.file):
+            log.info(
+                f"Skipping request for {req.file.acq.name}/{req.file.name}:"
+                f" not ready on node {req.node_from.name}."
+            )
+            return
+
+        # Early checks passed: dispatch this request to the Group I/O layer
+        self.io.pull(req)
+
     def update(self) -> None:
         """Perform I/O updates on the group"""
 
@@ -453,7 +579,13 @@ class UpdateableGroup(updateable_base):
         if self._init_idle and do_update:
             log.info(f'Updating group "{self.name}".')
 
-            # TODO: do group I/O here.
+            # Process pulls into this group
+            for req in ArchiveFileCopyRequest.select().where(
+                ArchiveFileCopyRequest.completed == 0,
+                ArchiveFileCopyRequest.cancelled == 0,
+                ArchiveFileCopyRequest.group_to == self.db,
+            ):
+                self.update_pull(req)
 
             # Check for idleness at the end
             self._do_idle_updates = self.idle
@@ -501,7 +633,6 @@ def update_loop(
     During a clean exit, alpenhornd will try to finish in-progress tasks before
     shutting down.
     """
-    global done_transport_this_cycle
 
     # The nodes and groups we're working on.  These will be updated
     # each time through the main loop, whenever the underlying storage objects
@@ -512,7 +643,6 @@ def update_loop(
 
     while not global_abort.is_set():
         loop_start = time.time()
-        done_transport_this_cycle = False
 
         # Nodes are re-queried every loop iteration so we can
         # detect changes in available storage media
@@ -674,203 +804,3 @@ def serial_io(queue: FairMultiFIFOQueue) -> None:
         task()
         queue.task_done(key)
         log.info(f"Finished task {task}")
-
-
-def update_node_requests(node):
-    """Process file copy requests onto this node."""
-    import shutil
-    from .io import ioutil
-
-    global done_transport_this_cycle
-
-    # TODO: Fix up HPSS support
-    # Ensure we are not on an HPSS node
-    # if is_hpss_node(node):
-    #     log.error("Cannot process HPSS node here.")
-    #     return
-
-    avail_gb = node.avail_gb
-
-    # Skip if node is too full
-    if avail_gb < (node.min_avail_gb + 10):
-        log.info("Node %s is nearly full. Skip transfers." % node.name)
-        return
-
-    # Calculate the total archive size from the database
-    size_query = (
-        ac.ArchiveFile.select(fn.Sum(ac.ArchiveFile.size_b))
-        .join(ar.ArchiveFileCopy)
-        .where(ar.ArchiveFileCopy.node == node, ar.ArchiveFileCopy.has_file == "Y")
-    )
-
-    size = size_query.scalar(as_tuple=True)[0]
-    current_size_gb = float(0.0 if size is None else size) / 2**30.0
-
-    # Stop if the current archive size is bigger than the maximum (if set, i.e. > 0)
-    if current_size_gb > node.max_total_gb and node.max_total_gb > 0.0:
-        log.info(
-            "Node %s has reached maximum size (current: %.1f GB, limit: %.1f GB)"
-            % (node.name, current_size_gb, node.max_total_gb)
-        )
-        return
-
-    # ... OR if this is a transport node quit if the transport cycle is done.
-    if node.storage_type == "T" and done_transport_this_cycle:
-        log.info("Ignoring transport node %s" % node.name)
-        return
-
-    start_time = time.time()
-
-    # Fetch requests to process from the database
-    requests = ar.ArchiveFileCopyRequest.select().where(
-        ~ar.ArchiveFileCopyRequest.completed,
-        ~ar.ArchiveFileCopyRequest.cancelled,
-        ar.ArchiveFileCopyRequest.group_to == node.group,
-    )
-
-    # Add in constraint that node_from cannot be an HPSS node
-    requests = requests.join(st.StorageNode).where(st.StorageNode.address != "HPSS")
-
-    for req in requests:
-        if time.time() - start_time > max_time_per_node_operation:
-            break  # Don't hog all the time.
-
-        # By default, if a copy fails, we mark the source file as suspect
-        # so it gets re-MD5'd on the source node.
-        check_source_on_err = True
-
-        # Only continue if the node is actually active
-        if not req.node_from.active:
-            continue
-
-        # For transport disks we should only copy onto the transport
-        # node if the from_node is local, this should prevent pointlessly
-        # rsyncing across the network
-        if node.storage_type == "T" and node.host != req.node_from.host:
-            log.debug(
-                "Skipping request for %s/%s from remote node [%s] onto local "
-                "transport disks"
-                % (req.file.acq.name, req.file.name, req.node_from.name)
-            )
-            continue
-
-        # Only proceed if the destination file does not already exist.
-        try:
-            ar.ArchiveFileCopy.get(
-                ar.ArchiveFileCopy.file == req.file,
-                ar.ArchiveFileCopy.node == node,
-                ar.ArchiveFileCopy.has_file == "Y",
-            )
-            log.info(
-                "Skipping request for %s/%s since it already exists on "
-                'this node ("%s"), and updating DB to reflect this.'
-                % (req.file.acq.name, req.file.name, node.name)
-            )
-            ar.ArchiveFileCopyRequest.update(completed=True).where(
-                ar.ArchiveFileCopyRequest.file == req.file
-            ).where(ar.ArchiveFileCopyRequest.group_to == node.group).execute()
-            continue
-        except pw.DoesNotExist:
-            pass
-
-        # Only proceed if the source file actually exists (and is not corrupted).
-        try:
-            ar.ArchiveFileCopy.get(
-                ar.ArchiveFileCopy.file == req.file,
-                ar.ArchiveFileCopy.node == req.node_from,
-                ar.ArchiveFileCopy.has_file == "Y",
-            )
-        except pw.DoesNotExist:
-            log.error(
-                "Skipping request for %s/%s since it is not available on "
-                'node "%s". [file_id=%i]'
-                % (req.file.acq.name, req.file.name, req.node_from.name, req.file.id)
-            )
-            continue
-
-        # Check that there is enough space available.
-        if node.avail_gb * 2**30.0 < 2.0 * req.file.size_b:
-            log.warning(
-                'Node "%s" is full: not adding datafile "%s/%s".'
-                % (node.name, req.file.acq.name, req.file.name)
-            )
-            continue
-
-        # Constuct the origin and destination paths.
-        from_path = "%s/%s/%s" % (req.node_from.root, req.file.acq.name, req.file.name)
-        if req.node_from.host != node.host:
-            if req.node_from.username is None or req.node_from.address is None:
-                log.error(
-                    "Source node (%s) not properly configured (username=%s, address=%s)",
-                    req.node_from.name,
-                    req.node_from.username,
-                    req.node_from.address,
-                )
-                continue
-
-            from_path = "%s@%s:%s" % (
-                req.node_from.username,
-                req.node_from.address,
-                from_path,
-            )
-
-        to_file = os.path.join(node.root, req.file.acq.name, req.file.name)
-        to_dir = os.path.dirname(to_file)
-        if not os.path.isdir(to_dir):
-            log.info('Creating directory "%s".' % to_dir)
-            os.makedirs(to_dir)
-
-        # For the potential error message later
-        stderr = None
-
-        # Giddy up!
-        log.info('Transferring file "%s/%s".' % (req.file.acq.name, req.file.name))
-        start_time = time.time()
-        req.transfer_started = dt.datetime.fromtimestamp(start_time)
-        req.save(only=req.dirty_fields)
-
-        # Attempt to transfer the file. Each of the methods below needs to set a
-        # return code `ret` and give an `md5sum` of the transferred file.
-
-        # First we need to check if we are copying over the network
-        if req.node_from.host != node.host:
-            # First try bbcp which is a fast multistream transfer tool. bbcp can
-            # calculate the md5 hash as it goes, so we'll do that to save doing
-            # it at the end.
-            if shutil.which("bbcp") is not None:
-                ioresult = ioutil.bbcp(from_path, to_dir, req.file.size_b)
-            # Next try rsync over ssh.
-            elif shutil.which("rsync") is not None:
-                ioresult = ioutil.rsync(from_path, to_dir, req.file.size_b, False)
-            # If we get here then we have no idea how to transfer the file...
-            else:
-                log.warn("No commands available to complete this transfer.")
-                check_source_on_err = False
-                ret = -1
-
-        # Okay, great we're just doing a local transfer.
-        else:
-            # First try to just hard link the file. This will only work if we
-            # are on the same filesystem. As there's no actual copying it's
-            # probably unecessary to calculate the md5 check sum, so we'll just
-            # fake it.
-            ioresult = ioutil.hardlink(from_path, to_dir, req.file.name)
-
-            # If we couldn't just link the file, try copying it with rsync.
-            if ioresult is None:
-                if shutil.which("rsync") is not None:
-                    ioresult = ioutil.rsync(from_path, to_dir, req.file.size_b, True)
-                else:
-                    log.warn("No commands available to complete this transfer.")
-                    check_source_on_err = False
-                    ret = -1
-
-        ioutil.copy_request_done(
-            req,
-            node.io,
-            check_src=ioresult.get("check_src", True),
-            md5ok=ioresult.get("md5sum", None),
-            start_time=start_time,
-            stderr=ioresult.get("stderr", None),
-            success=(ioresult["ret"] == 0),
-        )


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 7570e17..53df0ac 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -438,7 +438,7 @@ def simplecopy(
     acq = archiveacq(name="simplecopy_acq")
     file = archivefile(name="simplecopy_file", acq=acq, size_b=2**20)
     group = storagegroup(name="simplecopy_group")
-    node = storagenode(name="simplecopy_node", group=group)
+    node = storagenode(name="simplecopy_node", group=group, root="/simplecopy_node")
     return archivefilecopy(file=file, node=node)
 
 
diff --git a/tests/io/test_defaultgroup.py b/tests/io/test_defaultgroup.py
new file mode 100644
index 0000000..14fafb7
--- /dev/null
+++ b/tests/io/test_defaultgroup.py
@@ -0,0 +1,72 @@
+"""Test DefaultGroupIO."""
+
+import pytest
+from unittest.mock import MagicMock
+
+from alpenhorn.update import UpdateableNode, UpdateableGroup
+
+
+@pytest.fixture
+def groupnode(xfs, queue, storagegroup, storagenode):
+    """Fixture setting up a default test group.
+
+    Returns both the group and the node."""
+    stgroup = storagegroup(name="group")
+    node = UpdateableNode(
+        queue, storagenode(name="node", group=stgroup, active=True, root="/node")
+    )
+    group = UpdateableGroup(group=stgroup, nodes=[node], idle=True)
+
+    # Create the directory
+    xfs.create_dir("/node")
+
+    return group, node
+
+
+def test_too_many_nodes(storagegroup, storagenode):
+    """Test DefaultGroupIO rejecting more than one node."""
+
+    stgroup = storagegroup(name="group")
+    node1 = UpdateableNode(None, storagenode(name="node1", group=stgroup, active=True))
+    node2 = UpdateableNode(None, storagenode(name="node2", group=stgroup, active=True))
+
+    group = UpdateableGroup(group=stgroup, nodes=[node1, node2], idle=True)
+    assert group._nodes is None
+
+
+def test_just_enough_nodes(storagegroup, storagenode):
+    """Test DefaultGroupIO accepting one node."""
+
+    stgroup = storagegroup(name="group")
+    node = UpdateableNode(None, storagenode(name="node1", group=stgroup, active=True))
+
+    group = UpdateableGroup(group=stgroup, nodes=[node], idle=True)
+    assert group._nodes == [node]
+
+
+def test_exists(xfs, groupnode):
+    """Test DefaultGroupIO.exists."""
+
+    group, node = groupnode
+
+    # Make something on the node
+    xfs.create_file("/node/acq/file")
+
+    # Check
+    assert group.io.exists("acq/file") == node
+    assert group.io.exists("acq/no-file") is None
+    assert group.io.exists("no-acq/no-file") is None
+
+
+def test_pull_handoff(groupnode, simplerequest):
+    """Test DefaultGroupIO.pull()."""
+
+    # We mock the DefaultNodeIO.pull() method because we don't need to test it here.
+    group, node = groupnode
+    node.io.pull = MagicMock(return_value=None)
+
+    # Call pull
+    group.io.pull(simplerequest)
+
+    # Check for hand-off to the node
+    node.io.pull.assert_called_with(simplerequest)
diff --git a/tests/io/test_defaultnode.py b/tests/io/test_defaultnode.py
index 5020fb2..e66d19f 100644
--- a/tests/io/test_defaultnode.py
+++ b/tests/io/test_defaultnode.py
@@ -1,4 +1,5 @@
 """Test DefaultNodeIO."""
+import pytest
 
 
 def test_bytes_avail(xfs, unode):
@@ -24,6 +25,16 @@ def test_check_active(xfs, unode):
     assert unode.io.check_active() is True
 
 
+def test_exists(unode, xfs):
+    """test DefaultNodeIO.exists()"""
+
+    xfs.create_file("/node/dir/file")
+
+    assert unode.io.exists("dir/file") is True
+    assert unode.io.exists("dir/no-file") is False
+    assert unode.io.exists("no-dir/no-file") is False
+
+
 def test_filesize(unode, xfs):
     """test DefaultNodeIO.filesize()"""
     xfs.create_file("/node/dir/file1", st_size=1000)
@@ -46,3 +57,36 @@ def test_md5(unode, xfs):
         "/node/dir/file2", contents="The quick brown fox jumps over the lazy dog"
     )
     assert unode.io.md5("dir/file2") == "9e107d9d372bb6826bd81d3542a419d6"
+
+
+def test_reserve_bytes(unode, xfs):
+    """test byte reservations"""
+
+    xfs.create_dir("/node")
+    xfs.set_disk_usage(10000)
+
+    # We start off with 10,000 bytes available
+    #
+    # The reservation system reserves twice as much space
+    # as requested (see DefaultNodeIO.reserve_factor), so
+    # we only have enough space to concurrenlty reserve 5000 bytes.
+    assert unode.io.reserve_bytes(4000, check_only=True) is True
+    assert unode.io.reserve_bytes(40000) is False
+
+    # Now reserve some bytes
+    assert unode.io.reserve_bytes(30000) is False
+
+    # Each of these uses up 4,000 bytes (twice the amount requested)
+    assert unode.io.reserve_bytes(2000) is True  # 4k bytes reserved
+    assert unode.io.reserve_bytes(2000) is True  # 8k bytes reserved
+    assert unode.io.reserve_bytes(2000) is False  # can't reserve 12k bytes
+
+    # Release
+    unode.io.release_bytes(2000)  # 4k bytes reserved
+    assert unode.io.reserve_bytes(4000) is False  # needs 8k free (only have 6k)
+
+    with pytest.raises(ValueError):
+        unode.io.release_bytes(5000)  # too much
+
+    # Release the rest
+    unode.io.release_bytes(2000)
diff --git a/tests/io/test_defaultnode_delete.py b/tests/io/test_defaultnode_delete.py
index 672129a..777e72a 100644
--- a/tests/io/test_defaultnode_delete.py
+++ b/tests/io/test_defaultnode_delete.py
@@ -6,6 +6,7 @@ from unittest.mock import patch
 
 from alpenhorn.archive import ArchiveFileCopy
 from alpenhorn.io._default_asyncs import delete_async
+from alpenhorn.io.updownlock import UpDownLock
 
 
 @pytest.fixture
@@ -185,8 +186,8 @@ def test_delete_dirs(
     delete_copies = copies.copy()
     del delete_copies[2]
 
-    # Call async directly for simplicity.
-    delete_async(None, delete_copies)
+    # Call async directly with a fake UpDownLock
+    delete_async(None, UpDownLock(), delete_copies)
 
     # Only copies[2] remains
     assert ArchiveFileCopy.select().where(ArchiveFileCopy.has_file == "Y").count() == 1
diff --git a/tests/io/test_defaultnode_pull.py b/tests/io/test_defaultnode_pull.py
new file mode 100644
index 0000000..5adacd7
--- /dev/null
+++ b/tests/io/test_defaultnode_pull.py
@@ -0,0 +1,443 @@
+"""Test DefaultNodeIO.pull()"""
+
+import os
+import pytest
+import pathlib
+from unittest.mock import patch
+
+from alpenhorn.archive import ArchiveFileCopy, ArchiveFileCopyRequest
+from alpenhorn.update import UpdateableNode
+
+
+@pytest.fixture
+def have_bbcp(mock_run_command):
+    """Pretend to have bbcp
+
+    Mocks shutil.which to indicate bbcp is present.
+
+    Also mocks util.run_command to emulate running bbcp.  Use the
+    run_command_result marker to specify the result of running bbcp."""
+
+    from shutil import which as real_which
+
+    def _mocked_which(cmd, mode=os.F_OK | os.X_OK, path=None):
+        nonlocal real_which
+        if cmd == "bbcp":
+            return "BBCP"
+        return real_which(cmd, mode, path)
+
+    with patch("shutil.which", _mocked_which):
+        yield mock_run_command
+
+
+@pytest.fixture
+def have_rsync(mock_run_command):
+    """Pretend to have rsync
+
+    Mocks shutil.which to indicate rsync is present.
+
+    Also mocks util.run_command to emulate running rsync.  Use the
+    run_command_result marker to specify the result of running rsync."""
+
+    from shutil import which as real_which
+
+    def _mocked_which(cmd, mode=os.F_OK | os.X_OK, path=None):
+        nonlocal real_which
+        if cmd == "rsync":
+            return "RSYNC"
+        return real_which(cmd, mode, path)
+
+    with patch("shutil.which", _mocked_which):
+        yield mock_run_command
+
+
+@pytest.fixture
+def test_req(
+    xfs,
+    hostname,
+    queue,
+    storagegroup,
+    storagenode,
+    simplefile,
+    archivefilecopy,
+    archivefilecopyrequest,
+):
+    """Create a test ArchiveFileCopyRequest.
+
+    Returns a two-element tuple: (node_to, copy_request)"""
+
+    group_to = storagegroup(name="group_to")
+    node_to = UpdateableNode(
+        queue,
+        storagenode(name="node_to", group=group_to, host=hostname, root="/node_to"),
+    )
+    node_from = storagenode(
+        name="node_from",
+        group=storagegroup(name="group_from"),
+        root="/node_from",
+        host=hostname,  # By default the transfer is local
+        username="user",
+        address="addr",
+    )
+
+    copy = archivefilecopy(file=simplefile, node=node_from, has_file="Y")
+
+    # Create source file and dest root
+    xfs.create_dir("/node_to")
+    xfs.create_file(copy.path, st_size=copy.size_b)
+    return (
+        node_to,
+        archivefilecopyrequest(file=simplefile, node_from=node_from, group_to=group_to),
+    )
+
+
+@pytest.fixture
+def pull_async(test_req):
+    """Put a pull_async Task on the queue.
+
+    Returns a two-element tuple: (node_to, copy_request)"""
+    node, req = test_req
+    node.io.pull(req)
+
+    return test_req
+
+
+def test_pull_sync_undermin(queue, test_req):
+    """test DefaultNodeIO.pull synchronous under_min check"""
+
+    node, req = test_req
+
+    # set up node to fail under_min check
+    node.db.avail_gb = 1.0
+    node.db.min_avail_gb = 2.0
+    node.io.pull(req)
+
+    # No job should be queued and req isn't resolved.
+    assert queue.qsize == 0
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+
+def test_pull_sync_overmax(queue, test_req, archivefile, archivefilecopy):
+    """test DefaultNodeIO.pull synchronous over_max check"""
+
+    # Init I/O layer
+    node, req = test_req
+
+    # set up node to fail over_max check
+    file = archivefile(name="file2", acq=req.file.acq, size_b=100000)
+    archivefilecopy(file=file, node=node.db, has_file="Y")
+    node.db.max_total_gb = file.size_b / 2**21  # i.e. half of file.size_b
+    node.io.pull(req)
+
+    # No job should be queued and req isn't resolved.
+    assert queue.qsize == 0
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+
+def test_pull_sync_fit(xfs, queue, test_req):
+    """test DefaultNodeIO.pull synchronous fit check"""
+
+    node, req = test_req
+
+    # set up node to fail reserve_bytes check
+    xfs.set_disk_usage(10000)
+    node.io.reserve_bytes(4000)
+    node.io.pull(req)
+
+    # No job should be queued and req isn't resolved.
+    assert queue.qsize == 0
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+
+def test_pull_async_noroute(queue, pull_async):
+    """Test no route for remote pull."""
+
+    node, req = pull_async
+
+    # Make the request non-local
+    req.node_from.host = "other-host"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Req isn't resolved.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+    # Source is not being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+
+
+@pytest.mark.run_command_result(1, "", "bbcp_stderr")
+def test_pull_async_bbcp_fail(queue, have_bbcp, pull_async):
+    """Test an unsuccessful bbcp remote pull."""
+
+    node, req = pull_async
+
+    # Make the request non-local
+    req.node_from.host = "other-host"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Verify that bbcp ran
+    assert "bbcp" in have_bbcp()["cmd"]
+
+    # Req isn't resolved.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+    # Source is being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file == "M"
+
+
+@pytest.mark.run_command_result(0, "", "md5 d41d8cd98f00b204e9800998ecf8427e")
+def test_pull_async_bbcp_succeed(queue, have_bbcp, mock_filesize, pull_async):
+    """Test a successful bbcp remote pull."""
+
+    node, req = pull_async
+
+    # Make the request non-local
+    req.node_from.host = "other-host"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Req is complete.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is True
+    assert afcr.cancelled is False
+
+    # Verify that bbcp ran
+    assert "bbcp" in have_bbcp()["cmd"]
+
+    # Target copy exists
+    afc = ArchiveFileCopy.get(node=node.db, file=req.file)
+    assert afc.has_file == "Y"
+
+    # Source is not being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+
+
+@pytest.mark.run_command_result(1, "", "rsync_stderr")
+def test_pull_async_remote_rsync_fail(queue, have_rsync, pull_async):
+    """Test an unsuccessful rsync remote pull."""
+
+    node, req = pull_async
+
+    # Make the request non-local
+    req.node_from.host = "other-host"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Verify that rsync ran
+    assert "rsync" in have_rsync()["cmd"]
+
+    # Req isn't resolved.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+    # Source is being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file == "M"
+
+
+@pytest.mark.run_command_result(0, "", "md5 d41d8cd98f00b204e9800998ecf8427e")
+def test_pull_async_remote_rsync_succeed(queue, have_rsync, mock_filesize, pull_async):
+    """Test a successful rsync remote pull."""
+
+    node, req = pull_async
+
+    # Make the request non-local
+    req.node_from.host = "other-host"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Verify that rsync ran
+    assert "rsync" in have_rsync()["cmd"]
+
+    # Req is complete.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is True
+    assert afcr.cancelled is False
+
+    # Target copy exists
+    afc = ArchiveFileCopy.get(node=node.db, file=req.file)
+    assert afc.has_file == "Y"
+
+    # Source is not being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+
+
+def test_pull_async_remote_nomethod(queue, pull_async):
+    """Test remote pull with no command available."""
+
+    node, req = pull_async
+
+    # Make the request non-local
+    req.node_from.host = "other-host"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Req isn't resolved.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+    # Source is not being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+
+
+def test_pull_async_link(queue, pull_async):
+    """Test creating hardlink."""
+
+    node, req = pull_async
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Req is resolved after successful hardlink
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is True
+    assert afcr.cancelled is False
+
+    # Target copy exists
+    afc = ArchiveFileCopy.get(node=node.db, file=req.file)
+    assert afc.has_file == "Y"
+
+    # Source is not being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+
+
+def test_pull_async_link_arccontam(queue, pull_async):
+    """Test not creating hardlinks between archive nodes."""
+
+    node, req = pull_async
+
+    # Make one node non-archival
+    node.db.storage_type = "F"
+
+    # 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
+
+    # Source is not being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+
+
+@pytest.mark.run_command_result(0, "", "stderr")
+def test_pull_async_local_rsync_succeed(queue, have_rsync, mock_filesize, pull_async):
+    """Test a successful rsync remote pull."""
+
+    node, req = pull_async
+
+    # Make one node non-archival to avoid hardlinking
+    node.db.storage_type = "F"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Verify that rsync ran
+    assert "rsync" in have_rsync()["cmd"]
+
+    # Req is complete.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is True
+    assert afcr.cancelled is False
+
+    # Target copy exists
+    afc = ArchiveFileCopy.get(node=node.db, file=req.file)
+    assert afc.has_file == "Y"
+
+    # Source is not being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file != "M"
+
+
+@pytest.mark.run_command_result(1, "", "rsync_stderr")
+def test_pull_async_local_rsync_fail(queue, have_rsync, pull_async):
+    """Test an unsuccessful rsync local pull."""
+
+    node, req = pull_async
+
+    # Make one node non-archival to avoid hardlinking
+    node.db.storage_type = "F"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Verify that rsync ran
+    assert "rsync" in have_rsync()["cmd"]
+
+    # Req isn't resolved.
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+    # Source is being re-checked
+    assert ArchiveFileCopy.get(node=req.node_from, file=req.file).has_file == "M"
+
+
+def test_pull_fail_unlink(xfs, queue, pull_async):
+    """Test failure deleting the destination."""
+
+    node, req = pull_async
+
+    # Destination path
+    path = pathlib.Path(node.db.root, req.file.path)
+
+    # Create destination file
+    xfs.create_file(path)
+
+    # Verify
+    assert path.exists()
+
+    # Force hardlinking to fail
+    node.db.storage_type = "T"
+
+    # Call the async
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # Pull failure
+    afcr = ArchiveFileCopyRequest.get(id=req.id)
+    assert afcr.completed is False
+    assert afcr.cancelled is False
+
+    # File is gone
+    assert not path.exists()
diff --git a/tests/io/test_defaultnoderemote.py b/tests/io/test_defaultnoderemote.py
new file mode 100644
index 0000000..e3110cb
--- /dev/null
+++ b/tests/io/test_defaultnoderemote.py
@@ -0,0 +1,37 @@
+"""Test DefaultNodeRemote."""
+
+import pytest
+
+from alpenhorn.update import RemoteNode
+
+
+def test_file_addr(simplenode, simplefile):
+    """Test DefaultNodeRemote.file_addr"""
+
+    simplenode.address = None
+    simplenode.username = "user"
+
+    with pytest.raises(ValueError):
+        assert RemoteNode(simplenode).io.file_addr(simplefile) is None
+
+    simplenode.address = "addr"
+    simplenode.username = None
+
+    with pytest.raises(ValueError):
+        assert RemoteNode(simplenode).io.file_addr(simplefile) is None
+
+    simplenode.address = "addr"
+    simplenode.username = "user"
+
+    assert (
+        RemoteNode(simplenode).io.file_addr(simplefile)
+        == "user@addr:/node/simplefile_acq/simplefile"
+    )
+
+
+def test_file_path(simplecopy):
+    """Test DefaultNodeRemote.file_name"""
+
+    assert RemoteNode(simplecopy.node).io.file_path(simplecopy.file) == str(
+        simplecopy.path
+    )
diff --git a/tests/io/test_updownlock.py b/tests/io/test_updownlock.py
new file mode 100644
index 0000000..f4bbc4e
--- /dev/null
+++ b/tests/io/test_updownlock.py
@@ -0,0 +1,343 @@
+"""Test the UpDownLock"""
+
+import pytest
+import threading
+from time import sleep
+
+from alpenhorn.io.updownlock import UpDownLock
+
+
+@pytest.fixture
+def udlock():
+    """Provide and check an UpDownLock
+
+    Yields a newly-created lock and after the test
+    ensures it's unlocked.
+    """
+
+    lock = UpDownLock()
+
+    # Verify we're unlocked to begin with
+    assert "unlocked" in repr(lock)
+
+    yield lock
+
+    # Must be unlocked again after the test
+    assert "unlocked" in repr(lock)
+
+
+def test_lock_up(udlock):
+    """Test the up lock."""
+
+    assert udlock.up.acquire()
+
+    # Check
+    assert "locked up" in repr(udlock)
+    assert "count=1 " in repr(udlock)
+
+    udlock.up.release()
+
+
+def test_lock_down(udlock):
+    """Test the down lock."""
+
+    assert udlock.down.acquire()
+
+    # Check
+    assert "locked down" in repr(udlock)
+    assert "count=1 " in repr(udlock)
+
+    udlock.down.release()
+
+
+def test_unlock_release(udlock):
+    """Releasing an unlocked lock raises RuntimeError."""
+
+    with pytest.raises(RuntimeError):
+        udlock.up.release()
+
+    with pytest.raises(RuntimeError):
+        udlock.down.release()
+
+
+def test_antirelease(udlock):
+    """Releasing the wrong state raises RuntimeError."""
+
+    assert udlock.up.acquire()
+    with pytest.raises(RuntimeError):
+        udlock.down.release()
+    udlock.up.release()
+
+    assert udlock.down.acquire()
+    with pytest.raises(RuntimeError):
+        udlock.up.release()
+    udlock.down.release()
+
+
+def test_acquire_both(udlock):
+    """Attempting to acquire both lock states raises RuntimeError."""
+
+    assert udlock.up.acquire()
+    with pytest.raises(RuntimeError):
+        udlock.down.acquire()
+    udlock.up.release()
+
+    assert udlock.down.acquire()
+    with pytest.raises(RuntimeError):
+        udlock.up.acquire()
+    udlock.down.release()
+
+
+def test_reentry_up(udlock):
+    """Test lock re-entry."""
+
+    # Test the up lock
+    assert udlock.up.acquire()
+    assert udlock.up.acquire()
+    assert udlock.up.acquire()
+
+    # Check
+    assert "locked up" in repr(udlock)
+    assert "count=3" in repr(udlock)
+
+    udlock.up.release()
+    udlock.up.release()
+
+    # Still locked
+    assert "locked up" in repr(udlock)
+    assert "count=1" in repr(udlock)
+
+    udlock.up.release()
+
+
+def test_reentry_down(udlock):
+    """Test lock re-entry."""
+
+    # Test the down lock
+    assert udlock.down.acquire()
+    assert udlock.down.acquire()
+    assert udlock.down.acquire()
+
+    # Check
+    assert "locked down" in repr(udlock)
+    assert "count=3" in repr(udlock)
+
+    udlock.down.release()
+    udlock.down.release()
+
+    # Still locked
+    assert "locked down" in repr(udlock)
+    assert "count=1" in repr(udlock)
+
+    udlock.down.release()
+
+
+def test_threads_up(udlock):
+    """Test acquiring the lock from multiple threads."""
+
+    # Number of threads to run
+    n = 3
+
+    # Synchronisation barrier
+    barrier = threading.Barrier(1 + n)
+
+    def thread(udlock, barrier):
+        udlock.up.acquire()
+
+        # Wait for check to complete
+        barrier.wait()
+        barrier.wait()
+
+        # Now release
+        udlock.up.release()
+
+    # Create threads
+    threads = list()
+    for _ in range(n):
+        threads.append(
+            threading.Thread(target=thread, args=(udlock, barrier), daemon=True)
+        )
+
+    # Start threads
+    for thread in threads:
+        thread.start()
+
+    # Wait for lock acquisition
+    barrier.wait()
+
+    # Check
+    assert f"count={n}" in repr(udlock)
+
+    # Clean up
+    barrier.wait()
+    for thread in threads:
+        thread.join()
+
+
+def test_threads_down(udlock):
+    """Test acquiring the lock from multiple threads."""
+
+    # Number of threads to run
+    n = 3
+
+    # Synchronisation barriers
+    barrier = threading.Barrier(1 + n)
+
+    def thread(udlock, barrier):
+        udlock.down.acquire()
+
+        # Wait for check to complete
+        barrier.wait()
+        barrier.wait()
+
+        # Now release
+        udlock.down.release()
+
+    # Create threads
+    threads = list()
+    for _ in range(n):
+        threads.append(
+            threading.Thread(target=thread, args=(udlock, barrier), daemon=True)
+        )
+
+    # Start threads
+    for thread in threads:
+        thread.start()
+
+    # Wait for lock acquisition
+    barrier.wait()
+
+    # Check
+    assert f"count={n}" in repr(udlock)
+
+    # Clean up
+    barrier.wait()
+    for thread in threads:
+        thread.join()
+
+
+def test_nonblocking(udlock):
+    """Test a non-blocking acquire."""
+
+    barrier = threading.Barrier(2)
+
+    def thread(udlock, barrier):
+        # Lock up
+        udlock.up.acquire()
+
+        # Wait for main to test
+        barrier.wait()
+        barrier.wait()
+
+        udlock.up.release()
+
+    t = threading.Thread(target=thread, args=(udlock, barrier), daemon=True)
+    t.start()
+
+    # Wait for lock
+    barrier.wait()
+
+    # Try locking
+    assert not udlock.down.acquire(blocking=False)
+
+    # Test done
+    barrier.wait()
+    t.join()
+
+
+def test_waiting(udlock):
+    """Test waiting for a lock."""
+
+    # Synchronisation barrier between main and one test thread
+    barrier2 = threading.Barrier(2)
+
+    # Synchronisation barrier between all threads
+    barrier3 = threading.Barrier(3)
+
+    def upthread(udlock, barrier2, barrier3):
+        udlock.up.acquire()
+
+        # Signal we have the up lock
+        barrier3.wait()  # Sync point 1
+
+        # Wait for main to detect downthread waiting
+        barrier2.wait()  # Sync point 2
+
+        # Release the lock
+        udlock.up.release()
+
+        # Wait for downthread to acquire down lock
+        barrier3.wait()  # Sync point 3
+
+        # Attempt to acquire uplock (this blocks now)
+        assert udlock.up.acquire()
+
+        # Signal we have the up lock
+        barrier3.wait()  # Sync point 5
+
+        # Wait for main thread to check lock state
+        barrier2.wait()  # Sync point 6
+
+        # Release the lock again
+        udlock.up.release()
+
+    def downthread(udlock, barrier2, barrier3):
+        # Wait for upthread to lock up
+        barrier3.wait()  # Sync point 1
+
+        # Acquire the down lock (this blocks)
+        assert udlock.down.acquire()
+
+        # Signal we have the down lock
+        barrier3.wait()  # Sync point 3
+
+        # Wait for main to detect upthread waiting
+        barrier2.wait()  # Sync point 4
+
+        # Release the lock
+        udlock.down.release()
+
+        # Wait for upthread to re-acquire
+        barrier3.wait()  # Sync point 5
+
+    up = threading.Thread(
+        target=upthread, args=(udlock, barrier2, barrier3), daemon=True
+    )
+    down = threading.Thread(
+        target=downthread, args=(udlock, barrier2, barrier3), daemon=True
+    )
+
+    # Start
+    up.start()
+    down.start()
+
+    # Wait for lock up
+    barrier3.wait()  # Sync point 1
+
+    assert "locked up" in repr(udlock)
+
+    # Wait for a bit to give downthread a chance to block
+    sleep(0.05)
+
+    # It's probably fine now
+    barrier2.wait()  # Sync point 2
+
+    # Wait for lock change
+    barrier3.wait()  # Sync point 3
+
+    assert "locked down" in repr(udlock)
+
+    # Wait for a bit to give upthread a chance to block
+    sleep(0.05)
+    barrier2.wait()  # Sync point 4
+
+    # Wait for lock up
+    barrier3.wait()  # Sync point 5
+
+    assert "locked up" in repr(udlock)
+    # We're done
+    barrier2.wait()  # Sync point 6
+
+    # Clean up
+    up.join()
+    down.join()
diff --git a/tests/test_service.py b/tests/test_service.py
index 428f7bd..831e948 100644
--- a/tests/test_service.py
+++ b/tests/test_service.py
@@ -1,6 +1,7 @@
 """An end-to-end test of the alpenhorn daemon.
 
 Things that this end-to-end test does:
+    - pulls a file
     - checks a corrupt file
     - deletes a file
 
@@ -9,10 +10,14 @@ of the daemon, but to check the connectivity of the high-level blocks
 of the main update loop.
 """
 
+import os
 import sys
 import yaml
 import pytest
+import shutil
+import pathlib
 import peewee as pw
+from unittest.mock import patch
 from click.testing import CliRunner
 from urllib.parse import quote as urlquote
 
@@ -101,6 +106,14 @@ def e2e_db(xfs, hostname):
         patterns='["*.me"]',
     )
 
+    # A file that needs to be pulled onto the Transport group
+    pullme = ArchiveFile.create(name="pull.me", type=filetype, acq=acq1, size_b=0)
+    ArchiveFileCopy.create(
+        file=pullme, node=dftnode, has_file="Y", wants_file="Y", size_b=0, ready=False
+    )
+    ArchiveFileCopyRequest.create(file=pullme, node_from=dftnode, group_to=fleet)
+    xfs.create_file("/dft/acq1/pull.me")
+
     # A file to check
     checkme = ArchiveFile.create(name="check.me", acq=acq1, size_b=0, md5sum="0")
     ArchiveFileCopy.create(
@@ -126,6 +139,38 @@ def e2e_db(xfs, hostname):
     yield
 
 
+@pytest.fixture
+def mock_rsync(xfs):
+    """Mocks rsync."""
+
+    original_which = shutil.which
+
+    def _mocked_which(cmd, mode=os.F_OK | os.X_OK, path=None):
+        """A mock of shutil.which that fakes having rsync."""
+
+        nonlocal original_which
+        if cmd == "rsync":
+            return "RSYNC"
+
+        return original_which(cmd, mode, path)
+
+    def _mocked_rsync(from_path, to_dir, size_b, local):
+        """An ioutil.rsync mock."""
+
+        nonlocal xfs
+
+        filename = pathlib.PurePath(from_path).name
+        destpath = pathlib.Path(to_dir, filename)
+
+        xfs.create_file(destpath)
+
+        return {"ret": 0, "stdout": "", "md5sum": True}
+
+    with patch("shutil.which", _mocked_which):
+        with patch("alpenhorn.io.ioutil.rsync", _mocked_rsync):
+            yield
+
+
 @pytest.fixture
 def e2e_config(xfs, hostname):
     """Fixture creating the config file for the end-to-end test."""
@@ -150,7 +195,7 @@ def e2e_config(xfs, hostname):
     xfs.create_file("/etc/alpenhorn/alpenhorn.conf", contents=yaml.dump(config))
 
 
-def test_cli(e2e_db, e2e_config, loop_once):
+def test_cli(e2e_db, e2e_config, mock_rsync, loop_once):
     runner = CliRunner()
 
     result = runner.invoke(cli, catch_exceptions=False)
@@ -159,7 +204,14 @@ def test_cli(e2e_db, e2e_config, loop_once):
 
     # Check results
 
+    # pull.me has been pulled
     tp1 = StorageNode.get(name="tp1")
+    pullme = ArchiveFile.get(name="pull.me")
+    afcr = ArchiveFileCopyRequest.get(file=pullme)
+    assert afcr.completed
+    assert not afcr.cancelled
+    copy = ArchiveFileCopy(file=pullme, node=tp1)
+    assert copy.path.exists()
 
     # check.me is corrupt
     checkme = ArchiveFile.get(name="check.me")
diff --git a/tests/test_update_group.py b/tests/test_update_group.py
index 158305c..e7e476f 100644
--- a/tests/test_update_group.py
+++ b/tests/test_update_group.py
@@ -1,4 +1,223 @@
 """Tests for UpdateableGroup."""
+import pytest
+from unittest.mock import patch, call
+
+from alpenhorn.archive import ArchiveFileCopy, ArchiveFileCopyRequest
+
+
+def make_afcr(
+    name,
+    acq,
+    srcnode,
+    srcstate,
+    dstgroup,
+    dstnode,
+    dststate,
+    archivefile,
+    archivefilecopy,
+    archivefilecopyrequest,
+):
+    """Create an ArchiveFileCopyRequest with ArchiveFileCopies on src and dest.
+
+    Returns a 4-tuple: ArchiveFile, source-ArchiveFileCopy, dest-ArchiveFileCopy, Request
+    """
+    file = archivefile(name=name, acq=acq)
+    srccopy = archivefilecopy(node=srcnode, file=file, has_file=srcstate)
+    if dststate is None:
+        dstcopy = None
+    else:
+        dstcopy = archivefilecopy(node=dstnode, file=file, has_file=dststate)
+    afcr = archivefilecopyrequest(node_from=srcnode, group_to=dstgroup, file=file)
+
+    return file, srccopy, dstcopy, afcr
+
+
+@pytest.fixture
+def pull(
+    mockgroupandnode,
+    simpleacq,
+    simplenode,
+    archivefile,
+    archivefilecopy,
+    archivefilecopyrequest,
+):
+    """A simple ArchiveFileCopyRequest for update_group.
+
+    The group and node form `mockedgroupandnode` are the
+    destination.  `simplenode` is the source.
+    """
+
+    # Make source active
+    simplenode.active = True
+    simplenode.save()
+
+    mockio, group, node = mockgroupandnode
+
+    mockio.group.before_update.return_value = True
+    # File doesn't exist on dest
+    mockio.group.exists.return_value = None
+
+    result = make_afcr(
+        "file",
+        simpleacq,
+        simplenode,
+        "Y",
+        group.db,
+        node.db,
+        None,
+        archivefile,
+        archivefilecopy,
+        archivefilecopyrequest,
+    )
+
+    # Don't bother returning the dstcopy, which is None
+    return result[0], result[1], result[3]
+
+
+def test_update_group_inactive(mockgroupandnode, hostname, queue, pull, simplenode):
+    """Test running update.update_group with source node inactive."""
+
+    mockio, group, node = mockgroupandnode
+    file, copy, afcr = pull
+
+    # Source not active
+    simplenode.active = False
+    simplenode.save()
+
+    # run update
+    group.update()
+
+    # afcr is not handled
+    assert not ArchiveFileCopyRequest.get(file=file).completed
+    assert not ArchiveFileCopyRequest.get(file=file).cancelled
+    assert call.pull(file) not in mockio.group.mock_calls
+
+
+def test_update_group_nosrc(mockgroupandnode, hostname, queue, pull):
+    """Test running update.update_group with source file missing."""
+
+    mockio, group, node = mockgroupandnode
+    file, copy, afcr = pull
+
+    # Source not present
+    copy.has_file = "N"
+    copy.save()
+
+    # run update
+    group.update()
+
+    # afcr is not handled
+    assert not ArchiveFileCopyRequest.get(file=afcr.file).completed
+    assert not ArchiveFileCopyRequest.get(file=afcr.file).cancelled
+    assert call.pull(afcr) not in mockio.group.mock_calls
+
+
+def test_update_group_notready(mockgroupandnode, hostname, queue, pull):
+    """Test running update.update_group with source file not ready."""
+
+    mockio, group, node = mockgroupandnode
+    file, copy, afcr = pull
+
+    # Force source not ready
+    with patch(
+        "alpenhorn.io.default.DefaultNodeRemote.pull_ready", lambda self, file: False
+    ):
+        # run update
+        group.update()
+
+    # afcr is not handled
+    assert not ArchiveFileCopyRequest.get(file=file).completed
+    assert not ArchiveFileCopyRequest.get(file=file).cancelled
+    assert call.pull(afcr) not in mockio.group.mock_calls
+
+
+def test_update_group_path_exists(mockgroupandnode, hostname, queue, pull):
+    """Test running update.update_group with unexpected existing dest."""
+
+    mockio, group, node = mockgroupandnode
+    file, copy, afcr = pull
+
+    # File exists on dest
+    mockio.group.exists.return_value = node
+
+    # This runs twice, to test two possibilities:
+    # - no destination ArchiveFileCopy record
+    # - desitination ArchiveFileCopy present with has_file='N'
+    # It should behave the same both times
+    for missing in [True, False]:
+        # run update
+        group.update()
+
+        # Dest file copy needs checking
+        dst = ArchiveFileCopy.get(file=file, node=node.db)
+        assert dst.has_file == "M"
+
+        # Request is still pending
+        assert not ArchiveFileCopyRequest.get(file=file).completed
+        assert not ArchiveFileCopyRequest.get(file=file).cancelled
+
+        # Pull was not executed
+        assert call.pull(afcr) not in mockio.group.mock_calls
+
+        # Only need to do this the first time
+        if missing:
+            # "Delete" dest
+            dst.has_file = "N"
+            dst.save()
+
+
+def test_update_group_copy_state(
+    mockgroupandnode,
+    hostname,
+    queue,
+    simplenode,
+    simpleacq,
+    archivefile,
+    archivefilecopy,
+    archivefilecopyrequest,
+):
+    """Test running update.update_group with different copy states."""
+
+    mockio, group, node = mockgroupandnode
+
+    mockio.group.before_update.return_value = True
+    # Files don't exist on dest
+    mockio.group.exists.return_value = None
+
+    # Source is active
+    simplenode.active = True
+    simplenode.save()
+
+    # Make some pull requests
+    commonargs = (simpleacq, simplenode, "Y", group.db, node.db)
+    factories = (archivefile, archivefilecopy, archivefilecopyrequest)
+    fileY, srcY, dstY, afcrY = make_afcr("fileY", *commonargs, "Y", *factories)
+    fileM, srcM, dstM, afcrM = make_afcr("fileM", *commonargs, "M", *factories)
+    fileX, srcX, dstX, afcrX = make_afcr("fileX", *commonargs, "X", *factories)
+    fileN, srcN, dstN, afcrN = make_afcr("fileN", *commonargs, "N", *factories)
+
+    # run update
+    group.update()
+
+    # afcrY is cancelled
+    assert not ArchiveFileCopyRequest.get(file=fileY).completed
+    assert ArchiveFileCopyRequest.get(file=fileY).cancelled
+    assert call.pull(afcrY) not in mockio.group.mock_calls
+
+    # afcrM is ongoing, but not handled
+    assert not ArchiveFileCopyRequest.get(file=fileM).completed
+    assert not ArchiveFileCopyRequest.get(file=fileM).cancelled
+    assert call.pull(afcrM) not in mockio.group.mock_calls
+
+    # afcrX was executed
+    assert not ArchiveFileCopyRequest.get(file=fileX).completed
+    assert not ArchiveFileCopyRequest.get(file=fileX).cancelled
+    assert call.pull(afcrX) in mockio.group.mock_calls
+
+    # afcrN was executed
+    assert not ArchiveFileCopyRequest.get(file=fileX).completed
+    assert not ArchiveFileCopyRequest.get(file=fileX).cancelled
+    assert call.pull(afcrN) in mockio.group.mock_calls
 
 
 def test_group_idle(queue, mockgroupandnode):
diff --git a/tests/test_update_node.py b/tests/test_update_node.py
index f7e571e..faa67a0 100644
--- a/tests/test_update_node.py
+++ b/tests/test_update_node.py
@@ -2,7 +2,7 @@
 
 import pytest
 import datetime
-from unittest.mock import patch, MagicMock
+from unittest.mock import call, patch, MagicMock
 
 from alpenhorn.storage import StorageNode
 from alpenhorn.update import UpdateableNode
@@ -218,3 +218,34 @@ def test_update_delete_over_min(unode, simpleacq, archivefile, archivefilecopy):
     with patch.object(unode.io, "delete", mock_delete):
         unode.update_delete()
     mock_delete.assert_called_once_with([copyN])
+
+
+def test_update_node_run(
+    unode, queue, simplegroup, simplefile, archivefilecopy, archivefilecopyrequest
+):
+    """Test running UpdateableNode.update_node."""
+
+    # Make something to check
+    copy = archivefilecopy(node=unode.db, file=simplefile, has_file="M")
+
+    # And something to pull
+    afcr = archivefilecopyrequest(
+        node_from=unode.db, group_to=simplegroup, file=simplefile
+    )
+
+    mock = MagicMock()
+    mock.before_update.return_value = True
+    mock.bytes_avail.return_value = None
+    with patch.object(unode, "io", mock):
+        # update runs
+        unode.update()
+
+    assert unode._updated is True
+
+    # Check I/O calls
+    calls = list(mock.mock_calls)
+    assert len(calls) == 5
+    assert call.bytes_avail(fast=False) in calls
+    assert call.check(copy) in calls
+    assert call.delete([]) in calls
+    assert call.ready_pull(afcr) in calls