Repository: radiocosmology/alpenhorn

Issue:
Ability to specify a list of files via text file to execute a CLI command on.
This is a normal things we do when dealing with files on cedar (i.e. have a list of files in a text file that we want to delete and or transfer).

Not sure if it would apply to everything in the `file` command group or a subset or comprise a separate command group with specific commands...

Hints from issue comments:

Solution patch:
diff --git a/alpenhorn/cli/cli.py b/alpenhorn/cli/cli.py
index 2db80c8a0..1b96258d9 100644
--- a/alpenhorn/cli/cli.py
+++ b/alpenhorn/cli/cli.py
@@ -26,7 +26,7 @@ def check_then_update(
     1. If `do_check` is False, skips to step 6
     2. Calls `func` in "check" mode.
     3. If `do_update` is False, program exits
-    4. `click.confirm("Continue?")
+    4. Asks user for confirmation.
     5. If user declines confirmation, program exits
     6. If `do_update` is True, calls `func` in "update" mode.
 
@@ -36,7 +36,7 @@ def check_then_update(
         "update" mode (step 6)
     2. `ctx`: the click context object
     3. any positional parameters given in in `args`
-    4. `first_time`: a bool keyword arguemnt which is True the first time
+    4. `first_time`: a bool keyword argument which is True the first time
         `func` is called, and False the second time (if called twice).
 
     Anything returned by `func` is ignored.
diff --git a/alpenhorn/cli/group/sync.py b/alpenhorn/cli/group/sync.py
index c690d95cf..721f4d07b 100644
--- a/alpenhorn/cli/group/sync.py
+++ b/alpenhorn/cli/group/sync.py
@@ -19,7 +19,9 @@
 )
 from ..cli import check_then_update, echo
 from ..options import (
+    check_if_from_stdin,
     cli_option,
+    files_from_file,
     files_in_groups,
     not_both,
     requires_other,
@@ -34,6 +36,7 @@ def _run_cancel(
     group: StorageGroup,
     node: StorageNode,
     acqs: list[ArchiveAcq],
+    listed_files: set[ArchiveFile],
     show_acqs: bool,
     show_files: bool,
     first_time: bool,
@@ -54,6 +57,8 @@ def _run_cancel(
         StorageNode soruce, or None if not given
     acqs:
         list of ArchiveAcqs to limit to.  May be the empty list for no limit
+    listed_files:
+        set of ArchiveFiles listed in a --file-list file.
     show_acqs:
         List acqusitions affected by the command
     show_files:
@@ -73,6 +78,10 @@ def _run_cancel(
     if group is not None:
         query = query.where(ArchiveFileCopyRequest.group_to == group)
 
+    # Limit to listed files, if given
+    if listed_files:
+        query = query.where(ArchiveFileCopyRequest.file << listed_files)
+
     # Apply acq constraint, if any
     if acqs:
         query = query.join(ArchiveFile).where(ArchiveFile.acq << acqs)
@@ -143,6 +152,7 @@ def _run_sync(
     group: StorageGroup,
     node: StorageNode,
     acqs: list[ArchiveAcq],
+    listed_files: set[ArchiveFile],
     show_acqs: bool,
     show_files: bool,
     target: list[str],
@@ -162,6 +172,8 @@ def _run_sync(
         StorageNode soruce
     acqs:
         list of ArchiveAcqs to limit to.  May be the empty list for no limit
+    listed_files:
+        set of ArchiveFiles listed in a --file-list file.
     show_acqs:
         List acqusitions affected by the command
     show_files:
@@ -200,6 +212,10 @@ def _run_sync(
         )
     )
 
+    # Limit to listed files, if given
+    if listed_files:
+        query = query.where(ArchiveFile.id << listed_files)
+
     # Limit to acqs, if needed
     if acqs:
         query = query.where(ArchiveFile.acq << acqs)
@@ -335,6 +351,7 @@ def run_query(
     node_name: str | None,
     acq: list,
     cancel: bool,
+    listed_files: set[ArchiveFile],
     show_acqs: bool,
     show_files: bool,
     target: list,
@@ -358,6 +375,8 @@ def run_query(
         list of --acq from command line
     cancel:
         True if we're cancelling transfers.
+    listed_files:
+        set of ArchiveFiles listed in a --file-list file.
     show_acqs:
         List acqusitions affected by the command
     show_files:
@@ -395,6 +414,7 @@ def run_query(
             group,
             node,
             acqs,
+            listed_files,
             show_acqs,
             show_files,
             first_time,
@@ -406,6 +426,7 @@ def run_query(
             group,
             node,
             acqs,
+            listed_files,
             show_acqs,
             show_files,
             target,
@@ -431,6 +452,7 @@ def run_query(
     "  Incompatible with --force",
     is_flag=True,
 )
+@cli_option("file_list")
 @click.option(
     "--force",
     help="Force update (skips confirmation).  Incompatible with --check",
@@ -453,6 +475,7 @@ def sync(
     cancel,
     check,
     force,
+    file_list,
     show_acqs,
     show_files,
     target,
@@ -464,7 +487,7 @@ def sync(
     running on the desination.  Only files which are not already in GROUP will
     be copied.
 
-    Which files are copied may be further limited with the --acq and
+    Which files are copied may be further limited with the --acq, --file-list and
     --target options.
 
     \b
@@ -474,8 +497,8 @@ def sync(
     Using the --cancel flag, this command can also be used to cancel pending
     transfers into GROUP.  With a NODE, transfers from NODE to GROUP are cancelled,
     but you can instead of NODE use the special flag --all to cancel all inbound
-    transfers into GROUP.  Cancelling can still be further limited by --acq,
-    but not --target.
+    transfers into GROUP.  Cancelling can still be further limited by --acq and
+    --file-list, but not --target.
     """
 
     # Usage checks
@@ -489,11 +512,26 @@ def sync(
     if all_ and node_name is not None:
         raise click.UsageError("Can't use --all with NODE.")
 
+    # Set check mode if --file-list=- was used to redirect stdin and no --force
+    check = check_if_from_stdin(file_list, check, force)
+
+    # Load file list, if any
+    listed_files = files_from_file(file_list)
+
     # Run the check-confirm-update loop
     check_then_update(
         not force,
         not check,
         run_query,
         ctx,
-        args=[group_name, node_name, acq, cancel, show_acqs, show_files, target],
+        args=[
+            group_name,
+            node_name,
+            acq,
+            cancel,
+            listed_files,
+            show_acqs,
+            show_files,
+            target,
+        ],
     )
diff --git a/alpenhorn/cli/node/clean.py b/alpenhorn/cli/node/clean.py
index da7665452..a9e6cbea7 100644
--- a/alpenhorn/cli/node/clean.py
+++ b/alpenhorn/cli/node/clean.py
@@ -11,7 +11,9 @@
 from ...db import ArchiveFile, ArchiveFileCopy, database_proxy, utcnow
 from ..cli import check_then_update, echo
 from ..options import (
+    check_if_from_stdin,
     cli_option,
+    files_from_file,
     files_in_groups,
     not_both,
     resolve_acq,
@@ -28,6 +30,7 @@ def _run_query(
     archive_ok,
     days: datetime.datetime,
     has_file: pw.Expression,
+    listed_files: set[ArchiveFile],
     size: int,
     target,
     clean_goal: str,
@@ -62,6 +65,8 @@ def _run_query(
         That is: an object which is the result of evaluating, say,
         (ArchiveFileCopy.has_file == 'Y'), which we'll pass on to the where()
         clause in the query.
+    listed_files:
+        set of ArchiveFiles listed in a --file-list file.
     size:
         Converted to bytes
     first_time:
@@ -100,10 +105,11 @@ def _run_query(
         if acqs or days or size:
             query = query.join(ArchiveFile)
 
-        # Add a wants_file constraint, if appropriate (with --size)
-        # we need to skip this: we need to scan through all the files
-        # so the totals are correct.  In that case, this constraint
-        # is handled when looping through copies.
+        # Add a wants_file constraint, if appropriate
+        #
+        # If --size is used, we need to skip this because we need to
+        # scan through all the files so the totals are correct.  In
+        # that case, this constraint is handled when looping through copies.
         if not size:
             # Without a size, we filter out everything that's not a
             # candidate for update
@@ -120,6 +126,10 @@ def _run_query(
         if acqs:
             query = query.where(ArchiveFile.acq << acqs)
 
+        # Limit by file-list, if given
+        if listed_files:
+            query = query.where(ArchiveFileCopy.file << listed_files)
+
         # Limit to registration time, if requested.  This will implicitly
         # drop any file copies where the registration time is unknown.
         if days:
@@ -286,6 +296,7 @@ def _run_query(
     type=int,
     help="Only clean files registered more than COUNT days ago.",
 )
+@cli_option("file_list")
 @click.option(
     "--force",
     help="Force cleaning (skips confirmation).  Incompatible with --check",
@@ -302,7 +313,7 @@ def _run_query(
     "-s",
     metavar="SIZE",
     type=float,
-    help="Stop cleaning files once the total size cleaned reaces SIZE GiB.",
+    help="Stop cleaning files once the total size cleaned reaches SIZE GiB.",
 )
 @cli_option("target")
 @click.pass_context
@@ -315,6 +326,7 @@ def clean(
     check,
     days,
     force,
+    file_list,
     include_bad,
     now,
     size,
@@ -347,7 +359,7 @@ def clean(
 
     In normal operation, this command will schedule *all* files on NODE for
     cleaning, but the operation may be limited with the "--acq", "--days",
-    "--size", and "--target" options.
+    "--file-list", "--size", and "--target" options.
 
     Multiple --acq options may be given to provide a list of acquisitions to
     restrict cleaning to.  Files not in one of the specified acqusitions will
@@ -375,8 +387,8 @@ def clean(
     -------------------
 
     You may unschedule files for cleaning using the "--cancel" flag, which can
-    be similarly restricted with "--acq", "--days", and "--target", but not
-    "--size".
+    be similarly restricted with "--acq", "--days", "--file-list", and
+    "--target", but not "--size".
 
     Both kinds of cleaning (discretionary and immediate) will be cancelled,
     but only for files not yet removed from the NODE by the daemon.
@@ -390,6 +402,9 @@ def clean(
     if size is not None and size <= 0:
         raise click.UsageError("--size must be positive")
 
+    # Set check mode if --file-list=- was used to redirect stdin and no --force
+    check = check_if_from_stdin(file_list, check, force)
+
     # Convert days to a datetime
     if days:
         days = utcnow() - datetime.timedelta(days=-days)
@@ -413,11 +428,24 @@ def clean(
     else:
         has_file = ArchiveFileCopy.has_file == "Y"
 
+    # Load file list, if any
+    listed_files = files_from_file(file_list)
+
     # Do a check-then-update with the `_run_query` function.
     check_then_update(
         not force,
         not check,
         _run_query,
         ctx,
-        args=[name, acq, archive_ok, days, has_file, size, target, clean_goal],
+        args=[
+            name,
+            acq,
+            archive_ok,
+            days,
+            has_file,
+            listed_files,
+            size,
+            target,
+            clean_goal,
+        ],
     )
diff --git a/alpenhorn/cli/node/sync.py b/alpenhorn/cli/node/sync.py
index 5c103b06c..756b5384b 100644
--- a/alpenhorn/cli/node/sync.py
+++ b/alpenhorn/cli/node/sync.py
@@ -7,7 +7,13 @@
 
 from ..cli import check_then_update
 from ..group.sync import run_query
-from ..options import cli_option, not_both, requires_other
+from ..options import (
+    check_if_from_stdin,
+    cli_option,
+    files_from_file,
+    not_both,
+    requires_other,
+)
 
 
 @click.command()
@@ -28,6 +34,7 @@
     "  Incompatible with --force",
     is_flag=True,
 )
+@cli_option("file_list")
 @click.option(
     "--force",
     help="Force update (skips confirmation).  Incompatible with --check",
@@ -50,6 +57,7 @@ def sync(
     cancel,
     check,
     force,
+    file_list,
     show_acqs,
     show_files,
     target,
@@ -61,7 +69,7 @@ def sync(
     running on the desination.  Only files which are not already in GROUP will
     be copied.
 
-    Which files are copied may be further limited with the --acq and
+    Which files are copied may be further limited with the --acq, --file-list, and
     --target options.
 
     \b
@@ -71,8 +79,8 @@ def sync(
     Using the --cancel flag, this command can also be used to cancel pending
     transfers out of NODE.  With a GROUP, transfers from NODE to GROUP are cancelled,
     but you can instead of GROUP use the special flag --all to cancel all outbound
-    transfers from NODE.  Cancelling can still be further limited by --acq,
-    but not --target.
+    transfers from NODE.  Cancelling can still be further limited by --acq and
+    --file-list, but not --target.
     """
 
     # Usage checks
@@ -86,6 +94,12 @@ def sync(
     if all_ and group_name is not None:
         raise click.UsageError("Can't use --all with GROUP.")
 
+    # Set check mode if --file-list=- was used to redirect stdin and no --force
+    check = check_if_from_stdin(file_list, check, force)
+
+    # Load file list, if any
+    listed_files = files_from_file(file_list)
+
     # We're using the "group sync" run_query function here, which is why
     # group_name and node_name are backwards
     check_then_update(
@@ -93,5 +107,14 @@ def sync(
         not check,
         run_query,
         ctx,
-        args=[group_name, node_name, acq, cancel, show_acqs, show_files, target],
+        args=[
+            group_name,
+            node_name,
+            acq,
+            cancel,
+            listed_files,
+            show_acqs,
+            show_files,
+            target,
+        ],
     )
diff --git a/alpenhorn/cli/node/verify.py b/alpenhorn/cli/node/verify.py
index 919e22b9d..3e852a90b 100644
--- a/alpenhorn/cli/node/verify.py
+++ b/alpenhorn/cli/node/verify.py
@@ -8,7 +8,15 @@
 from ...common.util import pretty_bytes
 from ...db import ArchiveFile, ArchiveFileCopy, database_proxy
 from ..cli import check_then_update, echo
-from ..options import cli_option, not_both, resolve_acq, resolve_node, state_constraint
+from ..options import (
+    check_if_from_stdin,
+    cli_option,
+    files_from_file,
+    not_both,
+    resolve_acq,
+    resolve_node,
+    state_constraint,
+)
 
 
 def _run_query(
@@ -17,10 +25,11 @@ def _run_query(
     name: str,
     acq: list,
     file_selection: pw.Expression,
+    listed_files: set[ArchiveFile],
     verify_goal: str,
     first_time: bool,
 ):
-    """Run the veroify query, either in check mode or update mode.
+    """Run the verify query, either in check mode or update mode.
 
     Must be done twice because the DB contents can change while we wait
     for the user's confirmation.
@@ -28,7 +37,7 @@ def _run_query(
     Parameters:
     -----------
     update:
-        True if in "update" mode.  False in "count" mode.
+        True if in "update" mode.  False in "check" mode.
     ctx:
         click context object
     name:
@@ -39,6 +48,8 @@ def _run_query(
         a peewee.Expression encapsulating the --corrupt, --missing, and
         --healthy optons.  This is a constraint which will be applied to
         the `where` clause of the `ArchiveFileCopy` query.
+    listed_files:
+        set of ArchiveFiles listed in a --file-list file.
     verify_goal:
         The value we want to set has_file to.  This is 'M' except in
         --cancel mode.
@@ -59,6 +70,10 @@ def _run_query(
         if acqs:
             query = query.join(ArchiveFile).where(ArchiveFile.acq << acqs)
 
+        # Limit by file-list, if given
+        if listed_files:
+            query = query.where(ArchiveFileCopy.file << listed_files)
+
         # Perform the query
         copies = list(query.execute())
 
@@ -129,7 +144,7 @@ def _run_query(
     help="Cancel existing verifcation requests by explicitly setting their status",
 )
 @click.option(
-    "--count",
+    "--check",
     is_flag=True,
     help="Don't actually perform the operation, just print how many files "
     "would be affected",
@@ -140,9 +155,10 @@ def _run_query(
     "verify known corrupt files",
     is_flag=True,
 )
+@cli_option("file_list")
 @click.option(
     "--force",
-    help="Force update (skips confirmation).  Incompatible with --count",
+    help="Force update (skips confirmation).  Incompatible with --check",
     is_flag=True,
 )
 @click.option(
@@ -158,7 +174,9 @@ def _run_query(
     is_flag=True,
 )
 @click.pass_context
-def verify(ctx, name, acq, all_, cancel, count, corrupt, force, healthy, missing):
+def verify(
+    ctx, name, acq, all_, cancel, check, corrupt, force, file_list, healthy, missing
+):
     """Verify files on a Storage Node.
 
     Use this command to request verification of files on NODE, or to
@@ -177,10 +195,9 @@ def verify(ctx, name, acq, all_, cancel, count, corrupt, force, healthy, missing
     --all, --corrupt, --healthy, and --missing flags.  If any of these flags
     are used, the default selection is ignored.
 
-    Files to verify may be further restricted by specifying one or
-    more acqusitions to limit the operation to (via the --acq option).
-    Files which have been released for immediate removal (via, say, the
-    "node clean" command) are always skipped.
+    Files to verify may be further restricted using the --acq and --file-list
+    options.  Files which have been released for immediate removal (via, say,
+    the "node clean" command) are always skipped.
 
     NOTE: Be careful when you request re-verification.  After this command
     sets an affected file to "suspect", that file will not be available for
@@ -205,7 +222,10 @@ def verify(ctx, name, acq, all_, cancel, count, corrupt, force, healthy, missing
     will override whatever status you set manually.
     """
     # usage checks
-    not_both(count, "count", force, "force")
+    not_both(check, "check", force, "force")
+
+    # Set check mode if --file-list=- was used to redirect stdin and no --force
+    check = check_if_from_stdin(file_list, check, force)
 
     # Cancel and non-cancel modes are fairly different about what most
     # of the flags mean
@@ -252,11 +272,14 @@ def verify(ctx, name, acq, all_, cancel, count, corrupt, force, healthy, missing
         if file_selection is None:
             raise RuntimeError("Something went wrong!  Selection expression is empty.")
 
+    # Load file list, if any
+    listed_files = files_from_file(file_list)
+
     # Do a check-then-update with the `_run_query` function.
     check_then_update(
         not force,
-        not count,
+        not check,
         _run_query,
         ctx,
-        args=[name, acq, file_selection, verify_goal],
+        args=[name, acq, file_selection, listed_files, verify_goal],
     )
diff --git a/alpenhorn/cli/options.py b/alpenhorn/cli/options.py
index afc1aee46..af65d62a3 100644
--- a/alpenhorn/cli/options.py
+++ b/alpenhorn/cli/options.py
@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import fileinput
 import json
+import logging
 import pathlib
 
 import click
@@ -74,6 +76,19 @@ def cli_option(option: str, **extra_kwargs):
             "help": "Make node a field node (i.e. neither an archive node "
             "nor a transport node).  Incompatible with --archive or --transport.",
         }
+    elif option == "file_list":
+        args = ("-F", "--file-list")
+        kwargs = {
+            "metavar": "PATH",
+            "type": click.Path(
+                exists=True, dir_okay=False, readable=True, allow_dash=True
+            ),
+            "help": (
+                "Limit operation to files listed in text file at PATH.  "
+                "If PATH is -, stdin will be read.  In that case, --check will be "
+                "assumed if --force isn't used."
+            ),
+        }
     elif option == "from_":
         args = ("from_", "--from")
         kwargs = {"metavar": "SOURCE_NODE", "help": "Source Node for the transfer."}
@@ -634,20 +649,36 @@ def set_io_config(
     return new_config if new_config else None
 
 
-def file_from_path(path: str) -> ArchiveFile:
+def file_from_path(path: str, source: str | None = None) -> ArchiveFile:
     """Get file record given a path
 
     Given an "acqname/filename" path-like string, find and return
     the file name.
 
-    If a file cannot be found, raises click.ClickException.
+    Parameters
+    ----------
+    path:
+        The path to resolve into an ArchiveFile
+    source:
+        If not None, a location (i.e. a line in a file) added
+        to an error string.
+
+    Raises:
+    -------
+    click.ClickException:
+        A record for `path` could not be found.
     """
 
     path = pathlib.PurePath(path)
 
+    if source:
+        errstr = "No such file " + source + ": " + str(path)
+    else:
+        errstr = "No such file: " + str(path)
+
     # An absolute path is not allowed
     if path.is_absolute():
-        raise click.ClickException("No such file: " + str(path))
+        raise click.ClickException(errstr)
 
     # The trick here is path can have multiple path components: a/b/c/d/e
     # and we don't know which components are part of the acq_name and which
@@ -676,7 +707,87 @@ def file_from_path(path: str) -> ArchiveFile:
         except pw.DoesNotExist:
             pass
 
-    raise click.ClickException("No such file: " + str(path))
+    raise click.ClickException(errstr)
+
+
+def check_if_from_stdin(path: str, check: bool, force: bool) -> bool:
+    """Check whether to automatically turn on check mode
+
+    The intent here is that the output of this function will be assigned to
+    `check` in the caller.  Emits a warning if check mode is to be turned on.
+
+    Parameters
+    ----------
+    path:
+        the argument to --file-list
+    check:
+        the state of the --check flag
+    force:
+        the state of the --force flag
+
+    Returns
+    -------
+    check:
+        True if `check` is True or if `check` and `force` are False
+        and `path` is "-".  False otherwise.
+    """
+
+    # If mode is already selected, just return the input `check` value
+    if check or force:
+        return check
+
+    # Warn in turning on check mode
+    if path == "-":
+        log = logging.getLogger(__name__)
+        log.warning(
+            "stdin has been redirected without --force.  "
+            "Automatically enabling --check mode."
+        )
+        return True
+
+    # Otherwise, continue
+    return False
+
+
+def files_from_file(path: str) -> None:
+    """Read a file list from a file given with --file-list
+
+    Returns a set of ArchiveFiles.  If path is None, the empty set is returned.
+    """
+
+    files = set()
+
+    # Not given; return empty set
+    if not path:
+        return files
+
+    name = "stdin" if path == "-" else path
+
+    try:
+        for line in fileinput.input(files=[path]):
+            # Skip comment lines.  Note this check happens before whiespace
+            # stripping, meaning the "#" _must_ appear in the first column
+            # and may not have whitespace before it.
+            if line[0] == "#":
+                continue
+
+            # Strip leading and trailing whitespace
+            line = line.strip()
+
+            # Skip empty lines
+            if not line:
+                continue
+
+            files.add(
+                file_from_path(
+                    line,
+                    source=f"on line {fileinput.filelineno()} of {name}",
+                )
+            )
+    except OSError as e:
+        raise click.ClickException("error reading {name}: {e}") from e
+
+    return files
 
 
 def validate_md5(md5: str | None) -> None:
diff --git a/alpenhorn/daemon/entry.py b/alpenhorn/daemon/entry.py
index 595cf16c2..9666ece94 100755
--- a/alpenhorn/daemon/entry.py
+++ b/alpenhorn/daemon/entry.py
@@ -51,7 +51,8 @@ def log_exception(*args):
 )
 @version_option
 @help_config_option
-def entry(conf, once, test_isolation):
+@click.pass_context
+def entry(ctx, conf, once, test_isolation):
     """Alpenhornd: data management daemon.
 
     The alpenhorn daemon can be used to manage Storage Nodes.  See the alpenhorn
@@ -93,11 +94,15 @@ def entry(conf, once, test_isolation):
 
     # Enter main loop
     try:
-        update.update_loop(queue, wpool, once)
+        result = update.update_loop(queue, wpool, once)
     # Catch keyboard interrupt
     except KeyboardInterrupt:
         log.info("Exiting due to SIGINT")
+        result = 1
 
     # Attempt to exit cleanly
     auto_import.stop_observers()
     wpool.shutdown()
+
+    # Exit with result
+    ctx.exit(result)
diff --git a/alpenhorn/daemon/update.py b/alpenhorn/daemon/update.py
index aa4cb8792..a28c2c62c 100644
--- a/alpenhorn/daemon/update.py
+++ b/alpenhorn/daemon/update.py
@@ -923,21 +923,11 @@ def update_idle(self) -> None:
 
 def update_loop(
     queue: FairMultiFIFOQueue, pool: WorkerPool | EmptyPool, once: bool
-) -> None:
+) -> int:
     """Main loop of alepnhornd.
 
     This is the main update loop for the alpenhorn daemon.
 
-    Parameters
-    ----------
-    queue : FairMultiFIFOQueue
-        the task manager
-    pool : WorkerPool
-        the pool of worker threads (may be empty)
-    once : bool
-        If True, only run the loop once, wait for the queue to empty,
-        and then exit.  If False, loop forever.
-
     If `once` is false, the daemon cycles through the update loop until it is
     terminated in one of three ways:
 
@@ -949,6 +939,21 @@ def update_loop(
 
     During a clean exit, alpenhornd will try to finish in-progress tasks before
     shutting down.
+
+    Parameters
+    ----------
+    queue : FairMultiFIFOQueue
+        the task manager
+    pool : WorkerPool
+        the pool of worker threads (may be empty)
+    once : bool
+        If True, only run the loop once, wait for the queue to empty,
+        and then exit.  If False, loop forever.
+
+    Return
+    ------
+    result:
+        0 if exiting after running once.  1 otherwise.
     """
 
     # Get the name of this host
@@ -1134,7 +1139,7 @@ def update_loop(
             while True:
                 if queue.qsize + queue.inprogress_size + queue.deferred_size == 0:
                     log.info("Update complete.  Exiting.")
-                    return
+                    return 0
 
                 if first_time:
                     first_time = False
@@ -1155,6 +1160,7 @@ def update_loop(
 
     # Warn on abnormal exit
     log.warning("Exiting due to global abort")
+    return 1
 
 
 def serial_io(queue: FairMultiFIFOQueue) -> None:


Test patch:
diff --git a/tests/cli/group/test_sync.py b/tests/cli/group/test_sync.py
index 21c4f0ebf..b2fb7b0d5 100644
--- a/tests/cli/group/test_sync.py
+++ b/tests/cli/group/test_sync.py
@@ -494,3 +494,163 @@ def test_sync_show_files(clidb, cli):
     assert "Acq1/File2" in result.output
     assert "Acq2/File3" in result.output
     assert "Acq2/File4" in result.output
+
+
+def test_file_list_missing(clidb, cli):
+    """Test sync --file-list with non-existent file."""
+
+    group = StorageGroup.create(name="Group")
+    StorageNode.create(name="Node", group=group)
+
+    cli(2, ["group", "sync", "Group", "Node", "--file-list=missing"])
+
+
+def test_file_list_bad_file(clidb, cli, xfs):
+    """Test sync --file-list with a bad entry."""
+
+    group_from = StorageGroup.create(name="GroupFrom")
+    node_from = StorageNode.create(name="NodeFrom", group=group_from)
+
+    group_to = StorageGroup.create(name="GroupTo")
+    StorageNode.create(name="NodeTo", group=group_to)
+
+    acq = ArchiveAcq.create(name="Acq")
+
+    file1 = ArchiveFile.create(name="File1", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file1, node=node_from, has_file="Y", wants_file="Y")
+
+    file2 = ArchiveFile.create(name="File2", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file2, node=node_from, has_file="Y", wants_file="Y")
+
+    xfs.create_file("/file_list", contents="Acq/File1\n# Comment\nAcq/File3")
+
+    cli(
+        1,
+        [
+            "group",
+            "sync",
+            "GroupTo",
+            "NodeFrom",
+            "--force",
+            "--file-list=/file_list",
+        ],
+    )
+
+    # No files synced
+    assert ArchiveFileCopyRequest.select().count() == 0
+
+
+def test_file_list(clidb, cli, xfs):
+    """Test sync --file-list."""
+
+    group_from = StorageGroup.create(name="GroupFrom")
+    node_from = StorageNode.create(name="NodeFrom", group=group_from)
+
+    group_to = StorageGroup.create(name="GroupTo")
+    StorageNode.create(name="NodeTo", group=group_to)
+
+    acq = ArchiveAcq.create(name="Acq")
+
+    file1 = ArchiveFile.create(name="File1", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file1, node=node_from, has_file="Y", wants_file="Y")
+
+    file2 = ArchiveFile.create(name="File2", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file2, node=node_from, has_file="Y", wants_file="Y")
+
+    file3 = ArchiveFile.create(name="File3", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file3, node=node_from, has_file="Y", wants_file="Y")
+
+    # No newline at the end of this file
+    xfs.create_file("/file_list", contents="Acq/File1\n# Comment\nAcq/File3")
+
+    cli(
+        0,
+        [
+            "group",
+            "sync",
+            "GroupTo",
+            "NodeFrom",
+            "--force",
+            "--file-list=/file_list",
+        ],
+    )
+
+    # File1 and File3 are transferred
+    assert ArchiveFileCopyRequest.select().count() == 2
+    assert ArchiveFileCopyRequest.get(id=1).file == file1
+    assert ArchiveFileCopyRequest.get(id=2).file == file3
+
+
+def test_from_stdin(clidb, cli):
+    """Test sync --file-list=-."""
+
+    group_from = StorageGroup.create(name="GroupFrom")
+    node_from = StorageNode.create(name="NodeFrom", group=group_from)
+
+    group_to = StorageGroup.create(name="GroupTo")
+    StorageNode.create(name="NodeTo", group=group_to)
+
+    acq = ArchiveAcq.create(name="Acq")
+
+    file1 = ArchiveFile.create(name="File1", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file1, node=node_from, has_file="Y", wants_file="Y")
+
+    file2 = ArchiveFile.create(name="File2", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file2, node=node_from, has_file="Y", wants_file="Y")
+
+    file3 = ArchiveFile.create(name="File3", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file3, node=node_from, has_file="Y", wants_file="Y")
+
+    cli(
+        0,
+        [
+            "group",
+            "sync",
+            "GroupTo",
+            "NodeFrom",
+            "--file-list=-",
+            "--force",
+        ],
+        input="Acq/File1\n# Comment\nAcq/File3",
+    )
+
+    # File1 and File3 are transferred
+    assert ArchiveFileCopyRequest.select().count() == 2
+    assert ArchiveFileCopyRequest.get(id=1).file == file1
+    assert ArchiveFileCopyRequest.get(id=2).file == file3
+
+
+def test_from_stdin_unforced(clidb, cli):
+    """Test sync --file-list=-. without --force"""
+
+    group_from = StorageGroup.create(name="GroupFrom")
+    node_from = StorageNode.create(name="NodeFrom", group=group_from)
+
+    group_to = StorageGroup.create(name="GroupTo")
+    StorageNode.create(name="NodeTo", group=group_to)
+
+    acq = ArchiveAcq.create(name="Acq")
+
+    file1 = ArchiveFile.create(name="File1", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file1, node=node_from, has_file="Y", wants_file="Y")
+
+    file2 = ArchiveFile.create(name="File2", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file2, node=node_from, has_file="Y", wants_file="Y")
+
+    file3 = ArchiveFile.create(name="File3", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file3, node=node_from, has_file="Y", wants_file="Y")
+
+    cli(
+        0,
+        [
+            "group",
+            "sync",
+            "GroupTo",
+            "NodeFrom",
+            "--file-list=-",
+        ],
+        input="Acq/File1\n# Comment\nAcq/File3",
+    )
+
+    # Nothing is transferred because --check was turned on
+    assert ArchiveFileCopyRequest.select().count() == 0
diff --git a/tests/cli/group/test_sync_cancel.py b/tests/cli/group/test_sync_cancel.py
index fa081e3c0..e3c1a3462 100644
--- a/tests/cli/group/test_sync_cancel.py
+++ b/tests/cli/group/test_sync_cancel.py
@@ -372,3 +372,41 @@ def test_all(clidb, cli):
     # Both requests are cancelled
     assert ArchiveFileCopyRequest.get(id=1).cancelled == 1
     assert ArchiveFileCopyRequest.get(id=2).cancelled == 1
+
+
+def test_file_list(clidb, cli, xfs):
+    """Test sync --cancel --file-list."""
+
+    group_to = StorageGroup.create(name="GroupTo")
+    StorageNode.create(name="NodeTo", group=group_to)
+
+    group_from = StorageGroup.create(name="GroupFrom")
+    node = StorageNode.create(name="Node1", group=group_from)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file1 = ArchiveFile.create(name="File1", acq=acq)
+    file2 = ArchiveFile.create(name="File2", acq=acq)
+    file3 = ArchiveFile.create(name="File3", acq=acq)
+
+    ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group_to, cancelled=0, completed=0
+    )
+    ArchiveFileCopyRequest.create(
+        file=file2, node_from=node, group_to=group_to, cancelled=0, completed=0
+    )
+    ArchiveFileCopyRequest.create(
+        file=file3, node_from=node, group_to=group_to, cancelled=0, completed=0
+    )
+
+    xfs.create_file("/file_list", contents="Acq/File1\n# Comment\nAcq/File3")
+
+    cli(
+        0,
+        ["group", "sync", "GroupTo", "--cancel", "--all", "--file-list=/file_list"],
+        input="Y\n",
+    )
+
+    # File1 and File3 are cancelled
+    assert ArchiveFileCopyRequest.get(file=file1).cancelled == 1
+    assert ArchiveFileCopyRequest.get(file=file2).cancelled == 0
+    assert ArchiveFileCopyRequest.get(file=file3).cancelled == 1
diff --git a/tests/cli/node/test_clean.py b/tests/cli/node/test_clean.py
index 84d02a40d..24c6ba70b 100644
--- a/tests/cli/node/test_clean.py
+++ b/tests/cli/node/test_clean.py
@@ -490,3 +490,31 @@ def test_include_bad(clidb, cli):
     assert ArchiveFileCopy.get(node=node, file=fileY).wants_file == "N"
     assert ArchiveFileCopy.get(node=node, file=fileM).wants_file == "N"
     assert ArchiveFileCopy.get(node=node, file=fileX).wants_file == "N"
+
+
+def test_file_list(clidb, cli, xfs):
+    """Test clean with --file-list"""
+
+    group = StorageGroup.create(name="Group1")
+    node = StorageNode.create(name="NODE", group=group, storage_type="F")
+
+    acq = ArchiveAcq.create(name="Acq")
+
+    file1 = ArchiveFile.create(name="File1", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file1, has_file="Y", wants_file="Y")
+    file2 = ArchiveFile.create(name="File2", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file2, has_file="Y", wants_file="Y")
+    file3 = ArchiveFile.create(name="File3", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file3, has_file="Y", wants_file="Y")
+    file4 = ArchiveFile.create(name="File4", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file4, has_file="Y", wants_file="Y")
+
+    xfs.create_file("/file_list", contents="Acq/File1\n\n# Comment\nAcq/File3\n")
+
+    cli(0, ["node", "clean", "NODE", "--force", "--file-list=/file_list"])
+
+    # Only File1 and File3 were cleaned
+    assert ArchiveFileCopy.get(node=node, file=file1).wants_file == "M"
+    assert ArchiveFileCopy.get(node=node, file=file2).wants_file == "Y"
+    assert ArchiveFileCopy.get(node=node, file=file3).wants_file == "M"
+    assert ArchiveFileCopy.get(node=node, file=file4).wants_file == "Y"
diff --git a/tests/cli/node/test_sync.py b/tests/cli/node/test_sync.py
index 64d345901..b5e47a072 100644
--- a/tests/cli/node/test_sync.py
+++ b/tests/cli/node/test_sync.py
@@ -1,7 +1,7 @@
 """Test CLI: alpenhorn node sync
 
 Most of the test of the functionality for this command
-happen in the "node sync" tests.
+happen in the "group sync" tests.
 """
 
 from alpenhorn.db import (
@@ -144,3 +144,44 @@ def test_all(clidb, cli):
     # Both requests are cancelled
     assert ArchiveFileCopyRequest.get(id=1).cancelled == 1
     assert ArchiveFileCopyRequest.get(id=2).cancelled == 1
+
+
+def test_file_list(clidb, cli, xfs):
+    """Test sync --file-list."""
+
+    group_from = StorageGroup.create(name="GroupFrom")
+    node_from = StorageNode.create(name="NodeFrom", group=group_from)
+
+    group_to = StorageGroup.create(name="GroupTo")
+    StorageNode.create(name="NodeTo", group=group_to)
+
+    acq = ArchiveAcq.create(name="Acq")
+
+    file1 = ArchiveFile.create(name="File1", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file1, node=node_from, has_file="Y", wants_file="Y")
+
+    file2 = ArchiveFile.create(name="File2", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file2, node=node_from, has_file="Y", wants_file="Y")
+
+    file3 = ArchiveFile.create(name="File3", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file3, node=node_from, has_file="Y", wants_file="Y")
+
+    # No newline at the end of this file
+    xfs.create_file("/file_list", contents="Acq/File1\n# Comment\nAcq/File3")
+
+    cli(
+        0,
+        [
+            "node",
+            "sync",
+            "NodeFrom",
+            "GroupTo",
+            "--force",
+            "--file-list=/file_list",
+        ],
+    )
+
+    # File1 and File3 are transferred
+    assert ArchiveFileCopyRequest.select().count() == 2
+    assert ArchiveFileCopyRequest.get(id=1).file == file1
+    assert ArchiveFileCopyRequest.get(id=2).file == file3
diff --git a/tests/cli/node/test_verify.py b/tests/cli/node/test_verify.py
index fe5d26a35..b5a1511e6 100644
--- a/tests/cli/node/test_verify.py
+++ b/tests/cli/node/test_verify.py
@@ -360,3 +360,26 @@ def test_verify_acq(clidb, cli):
     assert ArchiveFileCopy.get(node=node, file=file1).has_file == "M"
     assert ArchiveFileCopy.get(node=node, file=file2).has_file == "M"
     assert ArchiveFileCopy.get(node=node, file=file3).has_file == "X"
+
+
+def test_file_list(clidb, cli, xfs):
+    """Test verify with --file-list."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="NODE", group=group, storage_type="F")
+    acq = ArchiveAcq.create(name="Acq")
+    file1 = ArchiveFile.create(name="File1", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file1, has_file="X", wants_file="Y")
+    file2 = ArchiveFile.create(name="File2", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file2, has_file="X", wants_file="Y")
+    file3 = ArchiveFile.create(name="File3", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file3, has_file="X", wants_file="Y")
+
+    xfs.create_file("/file_list", contents="Acq/File1\nAcq/File3\n")
+
+    cli(0, ["node", "verify", "NODE", "--file-list=/file_list"], input="Y\n")
+
+    # only 1 and 3 should be updated
+    assert ArchiveFileCopy.get(node=node, file=file1).has_file == "M"
+    assert ArchiveFileCopy.get(node=node, file=file2).has_file == "X"
+    assert ArchiveFileCopy.get(node=node, file=file3).has_file == "M"
diff --git a/tests/cli/test_options.py b/tests/cli/test_options.py
index 48d26f46c..a6be9e3b1 100644
--- a/tests/cli/test_options.py
+++ b/tests/cli/test_options.py
@@ -3,7 +3,7 @@
 import click
 import pytest
 
-from alpenhorn.cli.options import set_io_config
+from alpenhorn.cli.options import check_if_from_stdin, set_io_config
 
 
 def test_sic_empty():
@@ -96,3 +96,17 @@ def test_sic_default_decode_override():
     io_config = set_io_config("", ("a=29",), "rawr")
 
     assert io_config == {"a": 29}
+
+
+def test_check_if_from_stdin():
+    """Test check_if_from_stdin"""
+
+    assert check_if_from_stdin("path", True, True) is True
+    assert check_if_from_stdin("path", True, False) is True
+    assert check_if_from_stdin("path", False, True) is False
+    assert check_if_from_stdin("path", False, False) is False
+
+    assert check_if_from_stdin("-", True, True) is True
+    assert check_if_from_stdin("-", True, False) is True
+    assert check_if_from_stdin("-", False, True) is False
+    assert check_if_from_stdin("-", False, False) is True
diff --git a/tests/conftest.py b/tests/conftest.py
index 629e567c9..5437071bf 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,9 +1,11 @@
 """Common fixtures"""
 
+import fileinput
 import logging
 import os
 import re
 import shutil
+import traceback
 from unittest.mock import MagicMock, patch
 from urllib.parse import quote as urlquote
 
@@ -616,14 +618,15 @@ def cli(request, xfs, cli_config):
     runner = CliRunner(**kwargs)
 
     def _cli_wrapper(expected_result, *args, **kwargs):
-        nonlocal runner
-
-        import traceback
-
         from alpenhorn.cli import entry
 
+        nonlocal runner
+
         result = runner.invoke(entry, *args, **kwargs)
 
+        # Clean up fileinput
+        fileinput.close()
+
         # Show traceback if one was created
         if (
             result.exit_code