Repository: radiocosmology/alpenhorn

Issue:
Can't re-check file copies marked "has_file=M"
I think I ran into a loophole where file copies are marked "M" and there is no way to restore them short of hand-editing the database.

This is what happened:

1. transport disk at UBC, all files on it are good, and marked as "has_file=Y" in the database
2. copy request from the TD to cedar
3. on cedar, alpenhornd starts pulling the files, but I run out of quota space, and bbcp returns an error as it tries to write them.
4. alpenhorn detects error during copying, and marks the file copy on the TD as "has_file=M".
5. there are no ways to reset that to "Y" using client utilities, and I have to write SQL in order to do so and resume the copying after fixing my quota issues.

Commands I tried:
- `verify`: only checks files which are "has_file=Y" to mark them "N" or "X" if they fail
- `scan`: detects the file copy is already registered and does not make any checks or changes to the status

Discussion:
@ketiltrout: This happens all the time with our UBC-to-cedar copying, too.  At least with alpenhorn 1, the solution is to run an alpenhorn daemon on the machine at UBC with the TDs (`jingle` in our case).  The _only_ thing that particular daemon _ever_ does is wait around for the `cedar` daemon to mark TD files as `has_file='M'` and then runs a verify on them to either mark them `X` or `Y`.

Other that that (i.e. the vast, vast majority of the time), it just spins there doing nothing.

That said, a manual `verify` should also be checking `has_file="M"` entries.@ketiltrout: Really, verify should check all file copies that don't have `has_file='N'`, i.e. all supposedly extant, files whether they're ok (`Y`), uncertain (`M`) or known to be bad already (`X`), and then reporting differences from the database.@jrs65: Yeah, I agree the condition should probably `has_file!="N"` for `verify`@ketiltrout: Then it's complementary to `scan`, which is meant to find files which aren't expected to be on that node (i.e. it only updates absent entries and entries with `has_file=N`).@ketiltrout: The official solution to this problem now is what I mentioned above: run the daemon to update the node.  It will find `has_file='M'` files and check them.

See also #230 : we should have an on-demand mode for the daemon, but that's beyond the scope of this issue.

Solution patch:
diff --git a/alpenhorn/cli/acq/__init__.py b/alpenhorn/cli/acq/__init__.py
index 6701e76ce..9e1748420 100644
--- a/alpenhorn/cli/acq/__init__.py
+++ b/alpenhorn/cli/acq/__init__.py
@@ -5,6 +5,7 @@
 import click
 import peewee as pw
 
+from .create import create
 from .files import files
 from .list import list_
 from .show import show
@@ -15,6 +16,7 @@ def cli():
     """Manage Acquisitions."""
 
 
+cli.add_command(create, "create")
 cli.add_command(files, "files")
 cli.add_command(list_, "list")
 cli.add_command(show, "show")
diff --git a/alpenhorn/cli/acq/create.py b/alpenhorn/cli/acq/create.py
new file mode 100644
index 000000000..7e49b7777
--- /dev/null
+++ b/alpenhorn/cli/acq/create.py
@@ -0,0 +1,28 @@
+"""alpenhorn acq create command"""
+
+import click
+import peewee as pw
+
+from ...db import database_proxy, ArchiveAcq
+from ..options import cli_option
+from ..cli import echo
+
+
+@click.command()
+@click.argument("name", metavar="NAME")
+def create(name):
+    """Create a new Acquisitiop.
+
+    The Acquisition will be called NAME, which must not be the name of
+    another existing Acquisition.
+    """
+
+    with database_proxy.atomic():
+        try:
+            ArchiveAcq.get(name=name)
+            raise click.ClickException(f'Acquisition "{name}" already exists.')
+        except pw.DoesNotExist:
+            pass
+
+        ArchiveAcq.create(name=name)
+        echo(f'Created acquisition "{name}".')
diff --git a/alpenhorn/cli/cli.py b/alpenhorn/cli/cli.py
index fd2f6b782..67c2fc3e6 100644
--- a/alpenhorn/cli/cli.py
+++ b/alpenhorn/cli/cli.py
@@ -5,6 +5,7 @@
 import click
 import datetime
 import peewee as pw
+from typing import TYPE_CHECKING
 
 from ..common.logger import echo as echo
 from ..db import (
@@ -17,6 +18,72 @@
     StorageTransferAction,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+del TYPE_CHECKING
+
+
+def check_then_update(
+    do_check: bool, do_update: bool, func: Callable, ctx, *, args: list = []
+) -> None:
+    """Boilerplate for the check-confirm-update pattern.
+
+    Calls `func` up to twice, and maybe asks for confirmation in between.
+
+    Flow is like this:
+
+    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?")
+    5. If user declines confirmation, program exits
+    6. If `do_update` is True, calls `func` in "update" mode.
+
+    When calling `func` the following arguments are passed, in order:
+
+    1. `update`: a bool which is False in "check" mode (step 2) and True in
+        "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
+        `func` is called, and False the second time (if called twice).
+
+    Anything returned by `func` is ignored.
+
+    Parameters
+    ----------
+    do_check
+        True if we should run the check phase
+    do_update
+        True if we should run the update phase
+    func:
+        The function to call
+    ctx:
+        The click context
+    args:
+        A list of positional arguments to pass `func`
+    """
+
+    if do_check:
+        func(False, ctx, *args, first_time=True)
+
+        # If we're not doing the update, we're done
+        if not do_update:
+            ctx.exit()
+
+        # Ask for confirmation
+        echo()
+        if not click.confirm("Continue?"):
+            echo("\nCancelled.")
+            ctx.exit()
+        echo()
+
+    # If, both do_check and do_update are False, I guess we do nothing...
+    if do_update:
+        # `first_time` is `not do_check` here because if do_check is True, this is the
+        # second call, and if do_check is False, this is the first call
+        func(True, ctx, *args, first_time=not do_check)
+
 
 def update_or_remove(field: str, new: str | None, old: str | None) -> dict:
     """Helper for metadata updates.
diff --git a/alpenhorn/cli/group/autosync.py b/alpenhorn/cli/group/autosync.py
index ee457a229..dc97d5450 100644
--- a/alpenhorn/cli/group/autosync.py
+++ b/alpenhorn/cli/group/autosync.py
@@ -1,4 +1,4 @@
-"""alpenhorn group auto-sync command"""
+"""alpenhorn group autosync command"""
 
 import click
 import peewee as pw
@@ -13,17 +13,17 @@
 @click.option(
     "--remove",
     is_flag=True,
-    help="Remove (instead of add) NODE as an auto-sync source.",
+    help="Remove (instead of add) NODE as an autosync source.",
 )
 @click.pass_context
 def autosync(ctx, group_name, node_name, remove):
-    """Manage auto-sync sources for this group.
+    """Manage autosync sources for this group.
 
     This allows you to add (the default) or remove (using --remove)
-    the StorageNode named NODE as a an auto-sync souce for the Storage
+    the StorageNode named NODE as a an autosync souce for the Storage
     Group named GROUP.
 
-    If NODE is added as an auto-sync source for GROUP, then, whenever
+    If NODE is added as an autosync source for GROUP, then, whenever
     a file is added to NODE, it will be automatically synced into the
     Group GROUP, so long as the file isn't already present in the Group.
     """
@@ -39,10 +39,10 @@ def autosync(ctx, group_name, node_name, remove):
         except pw.DoesNotExist:
             raise click.ClickException(f"no such node: {node_name}")
 
-        # Sanity check: can't auto-sync within a group
+        # Sanity check: can't autosync within a group
         if group == node.group and not remove:
             raise click.ClickException(
-                "can't enable auto-sync: "
+                "can't enable autosync: "
                 f'Node "{node_name}" is in group "{group_name}"'
             )
 
diff --git a/alpenhorn/cli/group/show.py b/alpenhorn/cli/group/show.py
index 0e2307fca..ddeba4ca4 100644
--- a/alpenhorn/cli/group/show.py
+++ b/alpenhorn/cli/group/show.py
@@ -5,9 +5,17 @@
 import peewee as pw
 from tabulate import tabulate
 
-from ...db import StorageGroup, StorageNode, StorageTransferAction
+from ...common.util import pretty_bytes
+from ...db import (
+    StorageGroup,
+    StorageNode,
+    StorageTransferAction,
+    ArchiveFileCopyRequest,
+    ArchiveFile,
+)
 from ..cli import echo
 from ..node.stats import get_stats
+from ..options import cli_option
 
 
 @click.command()
@@ -17,10 +25,11 @@
     is_flag=True,
     help="Show post-transfer auto-actions affecting this group.",
 )
-@click.option("all_", "--all", "-a", is_flag=True, help="Show all additional data.")
+@cli_option("all_", help="Show all additional data.")
 @click.option("--node-details", is_flag=True, help="Show details of listed nodes.")
 @click.option("--node-stats", is_flag=True, help="Show usage stats of listed nodes.")
-def show(group_name, actions, all_, node_details, node_stats):
+@click.option("--transfers", is_flag=True, help="Show pending inbound transfers.")
+def show(group_name, actions, all_, node_details, node_stats, transfers):
     """Show details of a storage group.
 
     Shows details of the storage group named GROUP.
@@ -29,6 +38,7 @@ def show(group_name, actions, all_, node_details, node_stats):
     if all_:
         node_details = True
         node_stats = True
+        transfers = True
         actions = True
 
     try:
@@ -120,7 +130,35 @@ def show(group_name, actions, all_, node_details, node_stats):
     else:
         echo("  none")
 
-    # List transfer actions, if requested
+    # List transfers, if requested
+    if transfers:
+        echo("\nPending inbound transfers:\n")
+
+        query = (
+            ArchiveFileCopyRequest.select(
+                StorageNode.name,
+                pw.fn.COUNT(ArchiveFileCopyRequest.id).alias("count"),
+                pw.fn.Sum(ArchiveFile.size_b).alias("size"),
+            )
+            .join(ArchiveFile)
+            .switch(ArchiveFileCopyRequest)
+            .join(StorageNode)
+            .where(
+                ArchiveFileCopyRequest.group_to == group,
+                ArchiveFileCopyRequest.completed == 0,
+                ArchiveFileCopyRequest.cancelled == 0,
+            )
+            .group_by(ArchiveFileCopyRequest.node_from_id)
+            .order_by(StorageNode.name)
+        )
+
+        data = []
+        for node in query.tuples():
+            data.append((node[0], node[1], pretty_bytes(node[2])))
+
+        echo(tabulate(data, headers=["Source Node", "Request Count", "Total Size"]))
+
+    # List auto-actions, if requested
     if actions:
         echo("\nAuto-actions:\n")
 
diff --git a/alpenhorn/cli/node/__init__.py b/alpenhorn/cli/node/__init__.py
index 73668c702..10f3926a5 100644
--- a/alpenhorn/cli/node/__init__.py
+++ b/alpenhorn/cli/node/__init__.py
@@ -22,6 +22,7 @@
 from .rename import rename
 from .show import show
 from .stats import stats
+from .verify import verify
 
 RE_LOCK_FILE = re.compile(r"^\..*\.lock$")
 
@@ -41,6 +42,7 @@ def cli():
 cli.add_command(rename, "rename")
 cli.add_command(show, "show")
 cli.add_command(stats, "stats")
+cli.add_command(verify, "verify")
 
 
 @cli.command()
@@ -306,152 +308,3 @@ def scan(node_name, verbose, acq, register_new, dry):
         for fn in sorted(not_acqs_roots):
             print(fn)
         print()
-
-
-@cli.command()
-@click.argument("node_name", metavar="NODE")
-@click.option("--md5", help="perform full check against md5sum", is_flag=True)
-@click.option(
-    "--fixdb", help="fix up the database to be consistent with reality", is_flag=True
-)
-@click.option(
-    "--acq",
-    metavar="ACQ",
-    multiple=True,
-    help="Limit verification to specified acquisitions. Use repeated --acq flags to specify multiple acquisitions.",
-)
-def verify(node_name, md5, fixdb, acq):
-    """Verify the archive on NODE against the database.
-
-    If there are no issues with the archive returns with exit status of zero,
-    non-zero if there are issues. Specifically:
-
-    `0`
-        No problems.
-    `1`
-        Corrupt files found.
-    `2`
-        Files missing from archive.
-    `3`
-        Both corrupt and missing files.
-    """
-
-    import os
-
-    config_connect()
-
-    try:
-        this_node = StorageNode.get(name=node_name)
-    except pw.DoesNotExist:
-        click.echo('Storage node "{}" does not exist.'.format(node_name))
-        exit(1)
-
-    if not this_node.active:
-        click.echo('Node "{}" is not active.'.format(node_name))
-        exit(1)
-    if not util.alpenhorn_node_check(this_node):
-        click.echo(
-            'Node "{}" does not match ALPENHORN_NODE: {}'.format(
-                node_name, this_node.root
-            )
-        )
-        exit(1)
-
-    # Use a complicated query with a tuples construct to fetch everything we
-    # need in a single query. This massively speeds up the whole process versus
-    # fetching all the FileCopy's then querying for Files and Acqs.
-    lfiles = (
-        ArchiveFile.select(
-            ArchiveFile.name,
-            ArchiveAcq.name,
-            ArchiveFile.size_b,
-            ArchiveFile.md5sum,
-            ArchiveFileCopy.id,
-        )
-        .join(ArchiveAcq)
-        .switch(ArchiveFile)
-        .join(ArchiveFileCopy)
-        .where(ArchiveFileCopy.node == this_node, ArchiveFileCopy.has_file == "Y")
-    )
-
-    if acq:
-        lfiles = lfiles.where(ArchiveAcq.name << acq)
-
-    missing_files = []
-    corrupt_files = []
-
-    missing_ids = []
-    corrupt_ids = []
-
-    nfiles = 0
-
-    with click.progressbar(lfiles.tuples(), label="Scanning files") as lfiles_iter:
-        for filename, acqname, filesize, md5sum, fc_id in lfiles_iter:
-            nfiles += 1
-
-            filepath = this_node.root + "/" + acqname + "/" + filename
-
-            # Check if file is plain missing
-            if not os.path.exists(filepath):
-                missing_files.append(filepath)
-                missing_ids.append(fc_id)
-                continue
-
-            if md5:
-                file_md5 = util.md5sum_file(filepath)
-                corrupt = file_md5 != md5sum
-            else:
-                corrupt = os.path.getsize(filepath) != filesize
-
-            if corrupt:
-                corrupt_files.append(filepath)
-                corrupt_ids.append(fc_id)
-                continue
-
-    if len(missing_files) > 0:
-        click.echo()
-        click.echo("=== Missing files ===")
-        for fname in missing_files:
-            click.echo(fname)
-
-    if len(corrupt_files) > 0:
-        print()
-        click.echo("=== Corrupt files ===")
-        for fname in corrupt_files:
-            click.echo(fname)
-
-    click.echo()
-    click.echo("=== Summary ===")
-    click.echo("  %i total files" % nfiles)
-    click.echo("  %i missing files" % len(missing_files))
-    click.echo("  %i corrupt files" % len(corrupt_files))
-    click.echo()
-
-    # Fix up the database by marking files as missing, and marking
-    # corrupt files for verification by alpenhornd.
-    if fixdb:
-        # TODO: ensure write access to the database
-        # # We need to write to the database.
-        # di.connect_database(read_write=True)
-
-        if (len(missing_files) > 0) and click.confirm("Fix missing files"):
-            missing_count = (
-                ArchiveFileCopy.update(has_file="N")
-                .where(ArchiveFileCopy.id << missing_ids)
-                .execute()
-            )
-            click.echo("  %i marked as missing" % missing_count)
-
-        if (len(corrupt_files) > 0) and click.confirm("Fix corrupt files"):
-            corrupt_count = (
-                ArchiveFileCopy.update(has_file="M")
-                .where(ArchiveFileCopy.id << corrupt_ids)
-                .execute()
-            )
-            click.echo("  %i corrupt files marked for verification" % corrupt_count)
-    else:
-        # Set the exit status
-        status = 1 if corrupt_files else 0
-        status += 2 if missing_files else 0
-
-        exit(status)
diff --git a/alpenhorn/cli/node/autoclean.py b/alpenhorn/cli/node/autoclean.py
index 16ed09fd5..88e3fc7e0 100644
--- a/alpenhorn/cli/node/autoclean.py
+++ b/alpenhorn/cli/node/autoclean.py
@@ -1,4 +1,4 @@
-"""alpenhorn node auto-clean command"""
+"""alpenhorn node autoclean command"""
 
 import click
 import peewee as pw
@@ -13,17 +13,17 @@
 @click.option(
     "--remove",
     is_flag=True,
-    help="Remove (instead of add) GROUP as an auto-clean trigger.",
+    help="Remove (instead of add) GROUP as an autoclean trigger.",
 )
 @click.pass_context
 def autoclean(ctx, group_name, node_name, remove):
-    """Manage auto-clean triggers for this node.
+    """Manage autoclean triggers for this node.
 
     This allows you to add (the default) or remove (using --remove)
-    the StorageGroup named GROUP as a an auto-clean trigger for the
+    the StorageGroup named GROUP as a an autoclean trigger for the
     Storage Node named NODE.
 
-    If GROUP is added as an auto-clean trigger for NODE, then, whenever
+    If GROUP is added as an autoclean trigger for NODE, then, whenever
     a file is added to GROUP, it will be automatically released for
     deletion on NODE.
     """
@@ -39,10 +39,10 @@ def autoclean(ctx, group_name, node_name, remove):
         except pw.DoesNotExist:
             raise click.ClickException(f"no such group: {group_name}")
 
-        # Sanity check: can't auto-clean within a group
+        # Sanity check: can't autoclean within a group
         if group == node.group and not remove:
             raise click.ClickException(
-                "can't enable auto-clean: "
+                "can't enable autoclean: "
                 f'Node "{node_name}" is in group "{group_name}"'
             )
 
diff --git a/alpenhorn/cli/node/clean.py b/alpenhorn/cli/node/clean.py
index 10a7f3b00..616a00bfe 100644
--- a/alpenhorn/cli/node/clean.py
+++ b/alpenhorn/cli/node/clean.py
@@ -16,8 +16,8 @@
     utcnow,
 )
 from ...common.util import pretty_bytes
-from ..options import not_both
-from ..cli import echo
+from ..options import cli_option, not_both
+from ..cli import check_then_update, echo
 
 
 def _run_query(
@@ -31,10 +31,13 @@ def _run_query(
     size: int,
     target,
     clean_goal: str,
-    warn: bool = True,
+    first_time: bool = True,
 ) -> None:
     """Actually do the clean query, either in check mode or update mode.
 
+    This function is called via `alpenhorn.cli.cli.check_then_update`,
+    which sets some of the parameters.
+
     We separate this from check() because we typically have to do the
     whole thing twice, once at the start to get the results to present
     to the user, and then again to actually do the update.
@@ -61,7 +64,7 @@ def _run_query(
         clause in the query.
     size:
         Converted to bytes
-    warn:
+    first_time:
         True the first time this function is called (and we should print
         warnings).
     """
@@ -77,7 +80,7 @@ def _run_query(
         # --archive-ok was given
         if node.archive:
             if archive_ok:
-                if warn:
+                if first_time:
                     echo(f'DANGER: "{name}" is an archive node. Forcing clean.')
             else:
                 raise click.ClickException(f'Cannot clean archive node "{name}".')
@@ -275,7 +278,6 @@ def _run_query(
                 echo(
                     f'  {results["N"]["count"]} released {files} ({pretty_bytes(results["N"]["size"])})'
                 )
-        echo("")
 
         # Now update, if needed
         if update:
@@ -290,13 +292,7 @@ def _run_query(
 
 @click.command()
 @click.argument("name", metavar="NODE")
-@click.option(
-    "--acq",
-    metavar="ACQ",
-    default=None,
-    multiple=True,
-    help="Only clean files in acquisition ACQ.  May be specified multiple times.",
-)
+@cli_option("acq")
 @click.option(
     "--archive-ok", help="Run the clean, even if NODE is an archive node.", is_flag=True
 )
@@ -317,7 +313,6 @@ def _run_query(
 )
 @click.option(
     "--force",
-    "-f",
     help="Force cleaning (skips confirmation).  Incompatible with --check",
     is_flag=True,
 )
@@ -449,45 +444,11 @@ def clean(
     else:
         has_file = ArchiveFileCopy.has_file == "Y"
 
-    # If we're forcing, skip the check phase
-    if not force:
-        # Doesn't return if there's nothing to clean
-        _run_query(
-            False,
-            ctx,
-            name,
-            acq,
-            archive_ok,
-            days,
-            has_file,
-            size,
-            target,
-            clean_goal,
-            warn=True,
-        )
-
-        # With --check, we're done
-        if check:
-            ctx.exit()
-
-        # Ask for confirmation
-        if not click.confirm("Continue?"):
-            echo("\nCancelled.")
-            ctx.exit()
-        echo()
-
-    # Execute the query, if not in --check mode
-    if not check:
-        _run_query(
-            True,
-            ctx,
-            name,
-            acq,
-            archive_ok,
-            days,
-            has_file,
-            size,
-            target,
-            clean_goal,
-            warn=force,
-        )
+    # 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],
+    )
diff --git a/alpenhorn/cli/node/show.py b/alpenhorn/cli/node/show.py
index 1686af323..38b208875 100644
--- a/alpenhorn/cli/node/show.py
+++ b/alpenhorn/cli/node/show.py
@@ -6,8 +6,15 @@
 from tabulate import tabulate
 
 from ...common.util import pretty_bytes
-from ...db import StorageGroup, StorageNode, StorageTransferAction
+from ...db import (
+    StorageGroup,
+    StorageNode,
+    StorageTransferAction,
+    ArchiveFileCopyRequest,
+    ArchiveFile,
+)
 from ..cli import echo
+from ..options import cli_option
 from .stats import get_stats
 
 
@@ -18,9 +25,12 @@
     is_flag=True,
     help="Show post-transfer auto-actions affecting this group.",
 )
-@click.option("all_", "--all", "-a", is_flag=True, help="Show all additional data.")
+@cli_option("all_", help="Show all additional data.")
 @click.option("--stats", is_flag=True, help="Show usage stats of the node.")
-def show(name, actions, all_, stats):
+@click.option(
+    "--transfers", is_flag=True, help="Show pending transfers out from the node."
+)
+def show(name, actions, all_, stats, transfers):
     """Show details of a Storage Node.
 
     Shows details of the Storage Node named NODE.
@@ -29,6 +39,7 @@ def show(name, actions, all_, stats):
     if all_:
         actions = True
         stats = True
+        transfers = True
 
     try:
         node = StorageNode.get(name=name)
@@ -108,7 +119,35 @@ def show(name, actions, all_, stats):
         echo("     Total Size: " + stats["size"])
         echo("          Usage: " + stats["percent"].lstrip() + "%")
 
-    # List transfer actions, if requested
+    # List transfers, if requested
+    if transfers:
+        echo("\nPending outbound transfers:\n")
+
+        query = (
+            ArchiveFileCopyRequest.select(
+                StorageGroup.name,
+                pw.fn.COUNT(ArchiveFileCopyRequest.id).alias("count"),
+                pw.fn.Sum(ArchiveFile.size_b).alias("size"),
+            )
+            .join(ArchiveFile)
+            .switch(ArchiveFileCopyRequest)
+            .join(StorageGroup)
+            .where(
+                ArchiveFileCopyRequest.node_from == node,
+                ArchiveFileCopyRequest.completed == 0,
+                ArchiveFileCopyRequest.cancelled == 0,
+            )
+            .group_by(ArchiveFileCopyRequest.group_to_id)
+            .order_by(StorageGroup.name)
+        )
+
+        data = []
+        for group in query.tuples():
+            data.append((group[0], group[1], pretty_bytes(group[2])))
+
+        echo(tabulate(data, headers=["Dest. Group", "Request Count", "Total Size"]))
+
+    # List auto-actions, if requested
     if actions:
         echo("\nAuto-actions:\n")
 
diff --git a/alpenhorn/cli/node/verify.py b/alpenhorn/cli/node/verify.py
new file mode 100644
index 000000000..0ac5a4e8c
--- /dev/null
+++ b/alpenhorn/cli/node/verify.py
@@ -0,0 +1,292 @@
+"""alpenhorn node verify command"""
+
+from __future__ import annotations
+
+import click
+import peewee as pw
+
+from ...common.util import pretty_bytes
+from ...db import StorageNode, ArchiveAcq, ArchiveFile, ArchiveFileCopy, database_proxy
+from ..options import cli_option, not_both
+from ..cli import check_then_update, echo
+
+
+def _run_query(
+    update: bool,
+    ctx,
+    name: str,
+    acq: list,
+    file_selection: pw.Expression,
+    verify_goal: str,
+    first_time: bool,
+):
+    """Run the veroify 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.
+
+    Parameters:
+    -----------
+    update:
+        True if in "update" mode.  False in "count" mode.
+    ctx:
+        click context object
+    name:
+        node name
+    acq:
+        list of --acq arguments, if any
+    file_selection:
+        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.
+    verify_goal:
+        The value we want to set has_file to.  This is 'M' except in
+        --cancel mode.
+    """
+
+    with database_proxy.atomic():
+        # Check name
+        try:
+            node = StorageNode.get(name=name)
+        except pw.DoesNotExist:
+            raise click.ClickException("no such node: " + name)
+
+        # Resolve acqs
+        acqs = []
+        for acqname in acq:
+            try:
+                acqs.append(ArchiveAcq.get(name=acqname))
+            except pw.DoesNotExist:
+                raise click.ClickException("No such acquisition: " + acqname)
+
+        # Find all candidate file copies
+        query = ArchiveFileCopy.select(ArchiveFileCopy.id).where(
+            ArchiveFileCopy.node == node, file_selection
+        )
+
+        # Limit to acquisitions, if any
+        if acqs:
+            query = query.join(ArchiveFile).where(ArchiveFile.acq << acqs)
+
+        # Perform the query
+        copies = list(query.execute())
+
+        # Did we match?
+        if not copies:
+            echo("No matching files found.")
+            ctx.exit()
+
+        # Tot everything up.
+        total = (
+            ArchiveFileCopy.select(
+                pw.fn.COUNT(ArchiveFileCopy.id).alias("count"),
+                pw.fn.Sum(ArchiveFile.size_b).alias("size"),
+            )
+            .join(ArchiveFile)
+            .where(ArchiveFileCopy.id << copies)
+            .execute()
+        )[0]
+
+        # Grammar
+        files = "file" if total.count == 1 else "files"
+        if verify_goal != "M":  # i.e. cancelling
+            if update:
+                verb = "Forcing"
+            else:
+                verb = "Would force"
+            what = "status"
+
+            if verify_goal == "Y":
+                goal = " to healthy"
+            elif verify_goal == "N":
+                goal = " to missing"
+            else:
+                goal = " to corrupt"
+        else:
+            if update:
+                verb = "Requesting"
+            else:
+                verb = "Would request"
+            what = "verification"
+            goal = ""
+
+        size = pretty_bytes(total.size) if total.size is not None else "size unknown"
+        echo(f"{verb} {what} of {total.count} {files} ({size}){goal}.")
+
+        if update:
+            # Do the update
+            count = (
+                ArchiveFileCopy.update(has_file=verify_goal)
+                .where(ArchiveFileCopy.id << copies)
+                .execute()
+            )
+            files = "file" if count == 1 else "files"
+            echo(f"Updated {count} {files}.")
+
+
+@click.command()
+@click.argument("name", metavar="NODE")
+@cli_option("acq")
+@cli_option(
+    "all_",
+    help="Ignored if used with --cancel.  Otherwise, verify all files "
+    '(equivalent to "--corrupt --healthy --missing").',
+)
+@click.option(
+    "--cancel",
+    is_flag=True,
+    help="Cancel existing verifcation requests by explicitly setting their status",
+)
+@click.option(
+    "--count",
+    is_flag=True,
+    help="Don't actually perform the operation, just print how many files "
+    "would be affected",
+)
+@click.option(
+    "--corrupt",
+    help="With --cancel, set file status to corrupt.  Without --cancel, "
+    "verify known corrupt files",
+    is_flag=True,
+)
+@click.option(
+    "--force",
+    help="Force update (skips confirmation).  Incompatible with --count",
+    is_flag=True,
+)
+@click.option(
+    "--healthy",
+    help="With --cancel, set file status to healthy.  Without --cancel, "
+    "verify known good files",
+    is_flag=True,
+)
+@click.option(
+    "--missing",
+    help="With --cancel, set file status to missing.  Without --cancel, "
+    "verify known missing files",
+    is_flag=True,
+)
+@click.pass_context
+def verify(ctx, name, acq, all_, cancel, count, corrupt, force, healthy, missing):
+    """Verify files on a Storage Node.
+
+    Use this command to request verification of files on NODE, or to
+    explicitly set the status of suspect files (thus, cancelling verificaiton).
+
+    \b
+    VERIFYING FILES
+    ---------------
+
+    Without the "--cancel" flag, this command requests that the daemon re-verify
+    the integrity of files on NODE.  This is done by changing the status of the
+    files being re-verified to "suspect".
+
+    By default, only files known to be corrupt or missing are verified,
+    but the types of files to re-verify can be explicitly chosen with the
+    --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.
+
+    NOTE: Be careful when you request re-verification.  After this command
+    sets an affected file to "suspect", that file will not be available for
+    other operations (e.g. transfer) until the daemon has re-verified the
+    it.  Re-verification, therefore, is best done during times of low
+    activity on the NODE.
+
+    \b
+    CANCELLING VERIFICATION
+    -----------------------
+
+    With the "--cancel" flag, this command instead, changes the status
+    of matching suspect files to the new status specified by one of the
+    flags "--corrupt", "--missing", or "--healthy".  (If none are given,
+    "--corrupt" is the default.)
+
+    Which files are updated may be further restricted with the "--acq"
+    option.
+
+    NOTE: If you cancel verification of a file which the daemon happens to
+    be in the process of re-verifying, the result of the daemon's check
+    will override whatever status you set manually.
+    """
+    # usage checks
+    not_both(count, "count", force, "force")
+
+    # Cancel and non-cancel modes are fairly different about what most
+    # of the flags mean
+    if cancel:
+        # All is explicitly ignored in cancel mode: we effectively cancel
+        # all matching verification requests
+        not_both(corrupt, "corrupt", healthy, "healthy")
+        not_both(corrupt, "corrupt", missing, "missing")
+        not_both(healthy, "healthy", missing, "missing")
+
+        # We match against suspect files in this case
+        file_selection = (ArchiveFileCopy.has_file == "M") & (
+            ArchiveFileCopy.wants_file != "N"
+        )
+
+        # Our goal is determined by flag.  This is the
+        # value we'll be setting has_file to in the update
+        if healthy:
+            verify_goal = "Y"
+        elif missing:
+            verify_goal = "N"
+        else:
+            # --corrupt is the default, if no flag is given
+            verify_goal = "X"
+    else:
+        # Verify mode; normalise corrupt/missing/healthy flags
+        if all_:
+            corrupt = True
+            missing = True
+            healthy = True
+        elif not corrupt and not healthy and not missing:
+            corrupt = True
+            missing = True
+
+        # In verify mode, we're always going to set has_file to 'M'
+        verify_goal = "M"
+
+        # Build up a peewee.Expression for the types of files we want to find
+        if corrupt:
+            file_selection = (ArchiveFileCopy.has_file == "X") & (
+                ArchiveFileCopy.wants_file != "N"
+            )
+        else:
+            file_selection = None
+
+        if healthy:
+            healthy_expr = (ArchiveFileCopy.has_file == "Y") & (
+                ArchiveFileCopy.wants_file != "N"
+            )
+            if file_selection is None:
+                file_selection = healthy_expr
+            else:
+                file_selection = file_selection | healthy_expr
+
+        if missing:
+            missing_expr = (ArchiveFileCopy.has_file == "N") & (
+                ArchiveFileCopy.wants_file == "Y"
+            )
+            if file_selection is None:
+                file_selection = missing_expr
+            else:
+                file_selection = file_selection | missing_expr
+
+        # Sanity check
+        if file_selection is None:
+            raise RuntimeError("Something went wrong!  Selection expression is empty.")
+
+    # Do a check-then-update with the `_run_query` function.
+    check_then_update(
+        not force,
+        not count,
+        _run_query,
+        ctx,
+        args=[name, acq, file_selection, verify_goal],
+    )
diff --git a/alpenhorn/cli/options.py b/alpenhorn/cli/options.py
index 097f7ee97..73f944d5b 100644
--- a/alpenhorn/cli/options.py
+++ b/alpenhorn/cli/options.py
@@ -19,13 +19,26 @@ def cli_option(option: str, **extra_kwargs):
     """
 
     # Set args for the click.option decorator
-    if option == "address":
+    if option == "acq":
+        args = ("--acq",)
+        kwargs = {
+            "metavar": "ACQ",
+            "default": None,
+            "multiple": True,
+            "help": "May be specified multiple times.  "
+            "Limits operation to files in acquisition(s) named ACQ.",
+        }
+    elif option == "address":
         args = ("--address",)
         kwargs = {
             "metavar": "ADDR",
             "help": "Domain name or IP address to use for remote access to the "
             "node.",
         }
+    elif option == "all_":
+        # Has no help; must be provided when used
+        args = ("all_", "--all", "-a")
+        kwargs = {"is_flag": True}
     elif option == "archive":
         args = ("--archive",)
         kwargs = {
@@ -136,6 +149,12 @@ def cli_option(option: str, **extra_kwargs):
     # Update kwargs, if given
     kwargs.update(extra_kwargs)
 
+    # Default help string for common options missing them
+    if "help" not in kwargs:
+        kwargs["help"] = (
+            "SOMEONE FORGOT TO WRITE SOME HELP FOR THIS OPTION.  GOOD LUCK, I GUESS!"
+        )
+
     def _decorator(func):
         nonlocal args, kwargs
 
diff --git a/alpenhorn/common/util.py b/alpenhorn/common/util.py
index 75757eff0..575f6eaff 100644
--- a/alpenhorn/common/util.py
+++ b/alpenhorn/common/util.py
@@ -177,7 +177,7 @@ def _md5_chunk(f, md5, block_size, blocks_per_chunk):
                         _md5_chunk, f, md5, block_size, blocks_per_chunk
                     )
             except TimeoutError:
-                log.warning("Timeout trying to MD5 {filename}.")
+                log.warning(f"Timeout trying to MD5 {filename}.")
                 return None
 
     if hr:


Test patch:
diff --git a/tests/cli/acq/test_create.py b/tests/cli/acq/test_create.py
new file mode 100644
index 000000000..3794d594d
--- /dev/null
+++ b/tests/cli/acq/test_create.py
@@ -0,0 +1,26 @@
+"""Test CLI: alpenhorn acq create"""
+
+import json
+import pytest
+from alpenhorn.db import ArchiveAcq
+
+
+def test_create(clidb, cli):
+    """Test creating an acq."""
+
+    cli(0, ["acq", "create", "TEST"])
+
+    # Check created acq
+    ArchiveAcq.get(name="TEST")
+
+
+def test_create_existing(clidb, cli):
+    """Test creating an acq that already exists."""
+
+    # Add the acq
+    ArchiveAcq.create(name="TEST")
+
+    cli(1, ["acq", "create", "TEST"])
+
+    # Check that no extra acq was created
+    assert ArchiveAcq.select().count() == 1
diff --git a/tests/cli/group/test_show.py b/tests/cli/group/test_show.py
index 65bd60b65..6146816ec 100644
--- a/tests/cli/group/test_show.py
+++ b/tests/cli/group/test_show.py
@@ -8,6 +8,7 @@
     ArchiveAcq,
     ArchiveFile,
     ArchiveFileCopy,
+    ArchiveFileCopyRequest,
 )
 
 
@@ -222,6 +223,57 @@ def test_show_actions(clidb, cli, assert_row_present):
     assert "Node5" not in result.output
 
 
+def test_show_transfers(clidb, cli, assert_row_present):
+    """Test show --transfers."""
+
+    # Make a StorageGroup with some nodes in it.
+    group = StorageGroup.create(name="SGroup", io_class="IOClass")
+    node1 = StorageNode.create(name="Node1", group=group, active=True, host="over_here")
+    node2 = StorageNode.create(
+        name="Node2",
+        group=group,
+        active=False,
+        host="over_there",
+        io_class="NodeClass",
+        max_total_gb=1,
+    )
+
+    StorageTransferAction.create(
+        node_from=node1, group_to=group, autosync=1, autoclean=1
+    )
+
+    acq = ArchiveAcq.create(name="acq")
+    file = ArchiveFile.create(name="File1", acq=acq, size_b=1234)
+    ArchiveFileCopy.create(file=file, node=node1, has_file="Y", wants_file="Y")
+    ArchiveFileCopy.create(file=file, node=node2, has_file="X", wants_file="Y")
+
+    # Transfers
+    ArchiveFileCopyRequest.create(
+        node_from=node1, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node1, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node1, group_to=group, file=file, completed=0, cancelled=0
+    )
+
+    ArchiveFileCopyRequest.create(
+        node_from=node2, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node2, group_to=group, file=file, completed=0, cancelled=1
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node2, group_to=group, file=file, completed=1, cancelled=0
+    )
+
+    result = cli(0, ["group", "show", "SGroup", "--transfers"])
+
+    assert_row_present(result.output, "Node1", "3", "3.615 kiB")
+    assert_row_present(result.output, "Node2", "1", "1.205 kiB")
+
+
 def test_show_all(clidb, cli, assert_row_present):
     """Test show --all."""
 
@@ -255,6 +307,27 @@ def test_show_all(clidb, cli, assert_row_present):
     ArchiveFileCopy.create(file=file, node=node1, has_file="Y", wants_file="Y")
     ArchiveFileCopy.create(file=file, node=node2, has_file="Y", wants_file="Y")
 
+    # Transfers
+    ArchiveFileCopyRequest.create(
+        node_from=node1, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node1, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node1, group_to=group, file=file, completed=0, cancelled=0
+    )
+
+    ArchiveFileCopyRequest.create(
+        node_from=node2, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node2, group_to=group, file=file, completed=0, cancelled=1
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node2, group_to=group, file=file, completed=1, cancelled=0
+    )
+
     result = cli(0, ["group", "show", "SGroup", "--all"])
 
     assert_row_present(
@@ -264,5 +337,8 @@ def test_show_all(clidb, cli, assert_row_present):
         result.output, "Node2", "over_there", "No", "NodeClass", 2, "5.665 kiB", "0.00"
     )
 
+    assert_row_present(result.output, "Node1", "3", "10.12 kiB")
+    assert_row_present(result.output, "Node2", "1", "3.375 kiB")
+
     assert_row_present(result.output, "Node1", "Auto-clean", "File added to this group")
     assert_row_present(result.output, "Node1", "Auto-sync", "File added to that node")
diff --git a/tests/cli/node/test_clean.py b/tests/cli/node/test_clean.py
index b7b1a9552..dda752a5f 100644
--- a/tests/cli/node/test_clean.py
+++ b/tests/cli/node/test_clean.py
@@ -37,7 +37,7 @@ def test_cancel_size(clidb, cli):
     cli(2, ["node", "clean", "NODE", "--cancel", "--size=3"])
 
 
-def test_cancel_size(clidb, cli):
+def test_check_force(clidb, cli):
     """Test --check --force fails."""
 
     group = StorageGroup.create(name="Group")
@@ -66,31 +66,6 @@ def test_bad_size(clidb, cli):
     cli(2, ["node", "clean", "NODE", "--size=-1"])
 
 
-def test_bad_size(clidb, cli):
-    """Test non-positive --size."""
-
-    group = StorageGroup.create(name="Group")
-    node = StorageNode.create(name="NODE", group=group, storage_type="F")
-
-    cli(2, ["node", "clean", "NODE", "--size=0"])
-    cli(2, ["node", "clean", "NODE", "--size=-1"])
-
-
-def test_cancel(clidb, cli):
-    """Test cancelling a clean."""
-
-    group = StorageGroup.create(name="Group")
-    node = StorageNode.create(name="NODE", group=group, storage_type="F")
-    acq = ArchiveAcq.create(name="Acq")
-    file = ArchiveFile.create(name="File", acq=acq, size_b=1234)
-    ArchiveFileCopy.create(node=node, file=file, has_file="Y", wants_file="Y")
-
-    cli(0, ["node", "clean", "NODE"], input="N\n")
-
-    # Clean should not have run
-    assert ArchiveFileCopy.get(node=node, file=file).wants_file == "Y"
-
-
 def test_run(clidb, cli):
     """Test a simple clean."""
 
diff --git a/tests/cli/node/test_show.py b/tests/cli/node/test_show.py
index 2234abf31..cf68399d8 100644
--- a/tests/cli/node/test_show.py
+++ b/tests/cli/node/test_show.py
@@ -8,6 +8,7 @@
     ArchiveAcq,
     ArchiveFile,
     ArchiveFileCopy,
+    ArchiveFileCopyRequest,
     utcnow,
 )
 
@@ -189,10 +190,48 @@ def test_show_actions(clidb, cli, assert_row_present):
     assert "Node6" not in result.output
 
 
+def test_show_transfers(clidb, cli, assert_row_present):
+    """Test show --transfers."""
+
+    group1 = StorageGroup.create(name="Group1")
+    group2 = StorageGroup.create(name="Group2")
+    node = StorageNode.create(name="Node", group=group1)
+
+    acq = ArchiveAcq.create(name="acq")
+    file = ArchiveFile.create(name="File1", acq=acq, size_b=1234)
+
+    # Transfers
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group1, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group1, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group1, file=file, completed=0, cancelled=0
+    )
+
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group2, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group2, file=file, completed=0, cancelled=1
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group2, file=file, completed=1, cancelled=0
+    )
+
+    result = cli(0, ["node", "show", "Node", "--transfers"])
+
+    assert_row_present(result.output, "Group1", "3", "3.615 kiB")
+    assert_row_present(result.output, "Group2", "1", "1.205 kiB")
+
+
 def test_show_all(clidb, cli, assert_row_present):
     """Test show --all."""
 
     group = StorageGroup.create(name="Group")
+    group2 = StorageGroup.create(name="Group2")
     node = StorageNode.create(
         name="Node",
         group=group,
@@ -213,6 +252,27 @@ def test_show_all(clidb, cli, assert_row_present):
     file = ArchiveFile.create(name="File3", acq=acq, size_b=3456)
     ArchiveFileCopy.create(file=file, node=node, has_file="Y", wants_file="Y")
 
+    # Transfers
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group, file=file, completed=0, cancelled=0
+    )
+
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group2, file=file, completed=0, cancelled=0
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group2, file=file, completed=0, cancelled=1
+    )
+    ArchiveFileCopyRequest.create(
+        node_from=node, group_to=group2, file=file, completed=1, cancelled=0
+    )
+
     result = cli(0, ["node", "show", "Node", "--all"])
 
     assert "Total Files: 2" in result.output
@@ -223,5 +283,8 @@ def test_show_all(clidb, cli, assert_row_present):
     # 4.58 out of 8 == 57.25 percent
     assert "57.25%" in result.output
 
+    assert_row_present(result.output, "Group", "3", "10.12 kiB")
+    assert_row_present(result.output, "Group2", "1", "3.375 kiB")
+
     assert_row_present(result.output, "Group", "Auto-clean", "File added to that group")
     assert_row_present(result.output, "Group", "Auto-sync", "File added to this node")
diff --git a/tests/cli/node/test_verify.py b/tests/cli/node/test_verify.py
new file mode 100644
index 000000000..8699c624c
--- /dev/null
+++ b/tests/cli/node/test_verify.py
@@ -0,0 +1,383 @@
+"""Test CLI: alpenhorn node verify"""
+
+import pytest
+from datetime import timedelta
+
+from alpenhorn.db import (
+    StorageGroup,
+    StorageNode,
+    ArchiveAcq,
+    ArchiveFile,
+    ArchiveFileCopy,
+    utcnow,
+)
+
+
+@pytest.fixture
+def file_gamut(clidb):
+    """Fixture to create the has_file/wants_file gamut of files.
+
+    Yields node and a list of 3-tuples:
+        (has_file, wants_file, ArchiveFileCopy)
+    """
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="NODE", group=group, storage_type="F")
+    acq = ArchiveAcq.create(name="Acq")
+    files = []
+    for has in ["Y", "M", "X", "N"]:
+        for wants in ["Y", "M", "N"]:
+            file = ArchiveFile.create(name="File" + has + wants, acq=acq, size_b=1234)
+            ArchiveFileCopy.create(node=node, file=file, has_file=has, wants_file=wants)
+            files.append((has, wants, file))
+
+    return node, files
+
+
+def test_no_node(clidb, cli):
+    """Test clean with no node."""
+
+    cli(1, ["node", "verify", "TEST"])
+
+
+def test_check_force(clidb, cli):
+    """Test --check --force fails."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="NODE", group=group, storage_type="F")
+
+    cli(2, ["node", "verify", "NODE", "--check", "--force"])
+
+
+def test_cancel_multistatus(clidb, cli):
+    """Test --cancel with multiple statuses chosen."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="NODE", group=group, storage_type="F")
+
+    cli(2, ["node", "verify", "NODE", "--cancel", "--corrupt", "--healthy"])
+    cli(2, ["node", "verify", "NODE", "--cancel", "--corrupt", "--missing"])
+    cli(2, ["node", "verify", "NODE", "--cancel", "--healthy", "--missing"])
+
+
+def test_run(clidb, cli, file_gamut):
+    """Test a simple verify."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE"], input="Y\n")
+
+    # Verify should have run on 'XY', 'XM', 'NY'
+    for item in files:
+        if item[0] == "X" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "N" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    # i.e. we didn't update the existing has_file == 'M' copies
+    assert "Updated 3 files" in result.output
+
+
+def test_verify_corrupt(clidb, cli, file_gamut):
+    """Test verify --corupt."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--corrupt"], input="Y\n")
+
+    # Verify should have run on 'XY', 'XM'
+    for item in files:
+        if item[0] == "X" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 2 files" in result.output
+
+
+def test_verify_missing(clidb, cli, file_gamut):
+    """Test verify --missing."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--missing"], input="Y\n")
+
+    # Verify should have run on 'NY' only
+    for item in files:
+        if item[0] == "N" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 1 file" in result.output
+
+
+def test_verify_healthy(clidb, cli, file_gamut):
+    """Test verify --healthy."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--healthy"], input="Y\n")
+
+    # Verify should have run on 'YY', 'YM'
+    for item in files:
+        if item[0] == "Y" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "Y" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 2 files" in result.output
+
+
+def test_corrupt_missing(clidb, cli, file_gamut):
+    """Test explicit verify --corrupt --missing."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--corrupt", "--missing"], input="Y\n")
+
+    # Verify should have run on 'XY', 'XM', 'NY'
+    for item in files:
+        if item[0] == "X" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "N" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    # i.e. we didn't update the existing has_file == 'M' copies
+    assert "Updated 3 files" in result.output
+
+
+def test_corrupt_healtyh(clidb, cli, file_gamut):
+    """Test verify --corrupt --healthy."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--corrupt", "--healthy"], input="Y\n")
+
+    # Verify should have run on 'YY', 'YM', 'XY', 'XM'
+    for item in files:
+        if item[0] == "Y" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "Y" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 4 files" in result.output
+
+
+def test_healthy_missing(clidb, cli, file_gamut):
+    """Test verify --healthy --missing."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--healthy", "--missing"], input="Y\n")
+
+    # Verify should have run on 'YY', 'YM', 'NY'
+    for item in files:
+        if item[0] == "Y" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "Y" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "N" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 3 files" in result.output
+
+
+def test_hmc(clidb, cli, file_gamut):
+    """Test verify --healthy --missing --corrupt."""
+
+    node, files = file_gamut
+
+    result = cli(
+        0,
+        ["node", "verify", "NODE", "--healthy", "--missing", "--corrupt"],
+        input="Y\n",
+    )
+
+    # Verify should have run on 'YY', 'YM', 'XY', 'XM', 'NY'
+    for item in files:
+        if item[0] == "Y" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "Y" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "N" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 5 files" in result.output
+
+
+def test_all(clidb, cli, file_gamut):
+    """Test verify --all.
+
+    Should be the same as the previous --healthy --missing --corrupt test.
+    """
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--all"], input="Y\n")
+
+    # Verify should have run on 'YY', 'YM', 'XY', 'XM', 'NY'
+    for item in files:
+        if item[0] == "Y" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "Y" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "X" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        elif item[0] == "N" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "M"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 5 files" in result.output
+
+
+def test_cancel(clidb, cli, file_gamut):
+    """Test verify --cancel."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--cancel"], input="Y\n")
+
+    # should have run on 'MY', 'MM', and set has_file to 'X'
+    for item in files:
+        if item[0] == "M" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "X"
+        elif item[0] == "M" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "X"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 2 files" in result.output
+
+
+def test_cancel_corrupt(clidb, cli, file_gamut):
+    """Test verify --cancel --corrupt."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--cancel", "--corrupt"], input="Y\n")
+
+    # should have run on 'MY', 'MM', and set has_file to 'X'
+    for item in files:
+        if item[0] == "M" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "X"
+        elif item[0] == "M" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "X"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 2 files" in result.output
+
+
+def test_cancel_missing(clidb, cli, file_gamut):
+    """Test verify --cancel --missing."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--cancel", "--missing"], input="Y\n")
+
+    # should have run on 'MY', 'MM', and set has_file to 'N'
+    for item in files:
+        if item[0] == "M" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "N"
+        elif item[0] == "M" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "N"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 2 files" in result.output
+
+
+def test_cancel_missing(clidb, cli, file_gamut):
+    """Test verify --cancel --healthy."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--cancel", "--healthy"], input="Y\n")
+
+    # should have run on 'MY', 'MM', and set has_file to 'N'
+    for item in files:
+        if item[0] == "M" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "Y"
+        elif item[0] == "M" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "Y"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 2 files" in result.output
+
+
+def test_cancel_all(clidb, cli, file_gamut):
+    """Test --cancel --all (--all should be ignored)."""
+
+    node, files = file_gamut
+
+    result = cli(0, ["node", "verify", "NODE", "--cancel", "--all"], input="Y\n")
+
+    # should have run on 'MY', 'MM', and set has_file to 'X'
+    for item in files:
+        if item[0] == "M" and item[1] == "Y":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "X"
+        elif item[0] == "M" and item[1] == "M":
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == "X"
+        else:
+            assert ArchiveFileCopy.get(node=node, file=item[2]).has_file == item[0]
+
+    assert "Updated 2 files" in result.output
+
+
+def test_verify_bad_acq(clidb, cli, file_gamut):
+    """Test verify with bad --acq."""
+
+    result = cli(1, ["node", "verify", "NODE", "--acq=BAD"])
+
+
+def test_verify_acq(clidb, cli):
+    """Test verify with some --acq constraints."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="NODE", group=group, storage_type="F")
+    acq = ArchiveAcq.create(name="Acq1")
+    file1 = ArchiveFile.create(name="File", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file1, has_file="X", wants_file="Y")
+    acq = ArchiveAcq.create(name="Acq2")
+    file2 = ArchiveFile.create(name="File", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file2, has_file="X", wants_file="Y")
+    acq = ArchiveAcq.create(name="Acq3")
+    file3 = ArchiveFile.create(name="File", acq=acq)
+    ArchiveFileCopy.create(node=node, file=file3, has_file="X", wants_file="Y")
+
+    result = cli(0, ["node", "verify", "NODE", "--acq=Acq1", "--acq=Acq2"], input="Y\n")
+
+    # only 1 and 2 should be updated
+    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"