Repository: radiocosmology/alpenhorn

Issue:
Command to cancel copy requests for missing files
It would be nice to have a way to tell alpenhorn to cancel all copy requests for files which are missing on the source node, to resolve the inevitable
```
Skipping request for FOO since it is not available on node "BAR". [file_id=BAZ]
```
messages that we get on cedar due to files missing on transport disks.  I think usually these files are missing due to the unmount-in-the-middle-of-the-main-loop race condition in alpenhorn1.

Discussion:

Patch:
diff --git a/alpenhorn/cli/file/__init__.py b/alpenhorn/cli/file/__init__.py
index 03d9827..b30a118 100644
--- a/alpenhorn/cli/file/__init__.py
+++ b/alpenhorn/cli/file/__init__.py
@@ -6,6 +6,7 @@ import peewee as pw
 from .create import create
 from .import_ import import_
 from .show import show
+from .sync import sync
 
 
 @click.group(context_settings={"help_option_names": ["-h", "--help"]})
@@ -16,3 +17,4 @@ def cli():
 cli.add_command(create, "create")
 cli.add_command(import_, "import")
 cli.add_command(show, "show")
+cli.add_command(sync, "sync")
diff --git a/alpenhorn/cli/file/sync.py b/alpenhorn/cli/file/sync.py
new file mode 100644
index 0000000..0e69823
--- /dev/null
+++ b/alpenhorn/cli/file/sync.py
@@ -0,0 +1,124 @@
+"""alpenhorn file sync command."""
+
+import click
+import peewee as pw
+
+from ...db import ArchiveFileCopyRequest, database_proxy
+from ..cli import echo
+from ..options import cli_option, file_from_path, resolve_group, resolve_node
+
+
+@click.command()
+@click.argument("path", metavar="FILE")
+@click.option("--cancel", is_flag=True, help="Cancel existing transfer requests.")
+@click.option(
+    "--force",
+    is_flag=True,
+    help="Create the transfer request even if the source Node does "
+    "not have a copy of the file.  Ignored if used with --cancel.",
+)
+@cli_option(
+    "from_", help="Source Node for the transfer request.  Required if not cancelling."
+)
+@cli_option(
+    "to",
+    help="Destination Group for the transfer request.  Required if not cancelling.",
+)
+@click.pass_context
+def sync(ctx, path, cancel, force, from_, to):
+    """Create or cancel a File transfer.
+
+    This command allows you to either create a new request to transfer
+    a file from a Node to a Group, or else lets you cancel existing
+    pending transfer requests for FILE.  The FILE should be specified as
+    "<acq_name>/<filename>".
+
+    \b
+    New Requests
+    ------------
+
+    Wihtout "--cancel", this command creates a request for the daemon to
+    transfer the file named FILE from the Node SORUCE_NODE into the
+    destination Group DEST_GROUP.  In this case, both "--from" and "--to"
+    are required.  The File must not already be present in the destination
+    Group.
+
+    By default, the command will only create the request if the file exists
+    on the source.  To skip this check, and make the request anyways use the
+    "--force" flag.  In this case, the daemon may be unable to complete such
+    a request, and may cancel it, if it can't find the source file when it
+    tries to handle the request.
+
+    \b
+    Cancelling Requests
+    -------------------
+
+    To cancel pending transfer requests for FILE use the "--cancel" flag.
+    Which requests are cancelled may be limited with the "--from" and "--to"
+    flags.  By default, all exisingt pending requests for FILE are cancelled.
+    """
+
+    # Usage checks: must use --cancel or else both --from AND --to
+    if not cancel:
+        if not from_:
+            raise click.UsageError("missing --from")
+        if not to:
+            raise click.UsageError("missing --to")
+
+    if cancel:
+        # Cancel mode.  Find matching requests and cancel them.
+        with database_proxy.atomic():
+            file_ = file_from_path(path)
+            query = ArchiveFileCopyRequest.update(cancelled=True).where(
+                ArchiveFileCopyRequest.file == file_,
+                ArchiveFileCopyRequest.cancelled == 0,
+                ArchiveFileCopyRequest.completed == 0,
+            )
+
+            if from_:
+                query = query.where(
+                    ArchiveFileCopyRequest.node_from == resolve_node(from_)
+                )
+            if to:
+                query = query.where(
+                    ArchiveFileCopyRequest.group_to == resolve_group(to)
+                )
+
+            count = query.execute()
+
+        if count == 1:
+            echo("Cancelled 1 request.")
+        elif count:
+            echo(f"Cancelled {count} requests.")
+        else:
+            echo("No requests to cancel.")
+        ctx.exit()
+
+    # Create mode.
+    with database_proxy.atomic():
+        file_ = file_from_path(path)
+        src = resolve_node(from_)
+        dest = resolve_group(to)
+
+        # Is the file already in the dest?
+        state, node = dest.state_on_node(file_)
+        if state == "Y":
+            echo(
+                f'Not done: {path} already present on Node "{node.name}" '
+                f'in destination Group "{to}".'
+            )
+            ctx.exit()
+
+        # Does the source file exist?
+        if src.filecopy_state(file_) != "Y":
+            if force:
+                echo("File missing from source: forcing request.")
+            else:
+                echo("Not done: File missing from source.")
+                ctx.exit()
+
+        # Create request
+        ArchiveFileCopyRequest.create(
+            file=file_, node_from=src, group_to=dest, completed=0, cancelled=0
+        )
+        echo("Request submitted.")
diff --git a/alpenhorn/cli/options.py b/alpenhorn/cli/options.py
index f51261f..e5ea15b 100644
--- a/alpenhorn/cli/options.py
+++ b/alpenhorn/cli/options.py
@@ -72,6 +72,9 @@ 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 == "from_":
+        args = ("from_", "--from")
+        kwargs = {"metavar": "SOURCE_NODE", "help": "Source Node for the transfer."}
     elif option == "group":
         args = ("--group",)
         kwargs = {
@@ -148,6 +151,12 @@ def cli_option(option: str, **extra_kwargs):
             "help": "May be specified multiple times.  Restrict operation to "
             "files which exist in all specified target groups GROUP.",
         }
+    elif option == "to":
+        args = ("--to",)
+        kwargs = {
+            "metavar": "DEST_GROUP",
+            "help": "Destination Group for the transfer.",
+        }
     elif option == "transport":
         args = ("--transport",)
         kwargs = {


Test patch:
diff --git a/tests/cli/file/test_sync.py b/tests/cli/file/test_sync.py
new file mode 100644
index 0000000..4ddc1b6
--- /dev/null
+++ b/tests/cli/file/test_sync.py
@@ -0,0 +1,295 @@
+"""Test CLI: alpenhorn file sync"""
+
+from alpenhorn.db import (
+    ArchiveAcq,
+    ArchiveFile,
+    ArchiveFileCopy,
+    ArchiveFileCopyRequest,
+    StorageGroup,
+    StorageNode,
+    utcnow,
+)
+
+
+def test_bad_file(clidb, cli):
+    """Test a bad file."""
+
+    dest = StorageGroup.create(name="Dest")
+    StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    cli(1, ["file", "sync", "MIS/SING", "--from=Src", "--to=Dest"])
+
+
+def test_no_source(clidb, cli):
+    """Test a missing source node."""
+
+    dest = StorageGroup.create(name="Dest")
+    StorageNode.create(name="DestNode", group=dest)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(2, ["file", "sync", "Acq/File", "--to=Dest"])
+
+
+def test_bad_source(clidb, cli):
+    """Test a bad source node."""
+
+    dest = StorageGroup.create(name="Dest")
+    StorageNode.create(name="DestNode", group=dest)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--from=MISSING", "--to=Dest"])
+
+
+def test_no_dest(clidb, cli):
+    """Test missing dest group."""
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(2, ["file", "sync", "Acq/File", "--from=Src"])
+
+
+def test_bad_dest(clidb, cli):
+    """Test a bad dest group."""
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--from=Src", "--to=MISSING"])
+
+
+def test_no_src(clidb, cli):
+    """Test file missing from src."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest"])
+
+    # No request
+    assert ArchiveFileCopyRequest.select().count() == 0
+
+
+def test_sync(clidb, cli):
+    """Test a good sync."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+    ArchiveFileCopy.create(file=file_, node=src, has_file="Y", wants_file="Y")
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest", "--force"])
+
+    # Request created
+    afcr = ArchiveFileCopyRequest.get(id=1)
+    assert afcr.file == file_
+    assert afcr.node_from == src
+    assert afcr.group_to == dest
+    assert afcr.completed == 0
+    assert afcr.cancelled == 0
+
+
+def test_force(clidb, cli):
+    """Test force with no source file."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest", "--force"])
+
+    # Request created
+    afcr = ArchiveFileCopyRequest.get(id=1)
+    assert afcr.file == file_
+    assert afcr.node_from == src
+    assert afcr.group_to == dest
+    assert afcr.completed == 0
+    assert afcr.cancelled == 0
+
+
+def test_dest_has(clidb, cli):
+    """Test file already on dest."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+    ArchiveFileCopy.create(file=file_, node=src, has_file="Y", wants_file="Y")
+    ArchiveFileCopy.create(file=file_, node=node, has_file="Y", wants_file="Y")
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest"])
+
+    # No request
+    assert ArchiveFileCopyRequest.select().count() == 0
+
+
+def test_cancel_bad_file(clidb, cli):
+    """Test a bad file with --cancel."""
+
+    cli(1, ["file", "sync", "MIS/SING", "--cancel"])
+
+
+def test_cancel_all(clidb, cli):
+    """Test --cancel without limit."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file1 = ArchiveFile.create(name="File1", acq=acq)
+    file2 = ArchiveFile.create(name="File2", acq=acq)
+
+    req0 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=0, completed=0
+    )
+    req1 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=0, completed=1
+    )
+    req2 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=1, completed=0
+    )
+    req3 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=1, completed=1
+    )
+    req4 = ArchiveFileCopyRequest.create(
+        file=file2, node_from=node, group_to=group, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File1", "--cancel"])
+
+    # req0 newly cancelled.   req2, 3 previously cancelled.  Others unchanged.
+    assert ArchiveFileCopyRequest.get(id=req0.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req1.id).cancelled
+    assert ArchiveFileCopyRequest.get(id=req2.id).cancelled
+    assert ArchiveFileCopyRequest.get(id=req3.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req4.id).cancelled
+
+
+def test_cancel_bad_node(clidb, cli):
+    """Test --cancel with a bad --from."""
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--cancel", "--from=Missing"])
+
+
+def test_cancel_node(clidb, cli):
+    """Test --cancel with --from."""
+
+    group = StorageGroup.create(name="Group")
+    node1 = StorageNode.create(name="Node1", group=group)
+    node2 = StorageNode.create(name="Node2", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node1, group_to=group, cancelled=0, completed=0
+    )
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node2, group_to=group, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File", "--cancel", "--from=Node1"])
+
+    assert ArchiveFileCopyRequest.get(node_from=node1).cancelled
+    assert not ArchiveFileCopyRequest.get(node_from=node2).cancelled
+
+
+def test_cancel_bad_group(clidb, cli):
+    """Test --cancel with a bad --to."""
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--cancel", "--to=Missing"])
+
+
+def test_cancel_group(clidb, cli):
+    """Test --cancel with --to."""
+
+    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="File", acq=acq)
+
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node, group_to=group1, cancelled=0, completed=0
+    )
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node, group_to=group2, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File", "--cancel", "--to=Group1"])
+
+    assert ArchiveFileCopyRequest.get(group_to=group1).cancelled
+    assert not ArchiveFileCopyRequest.get(group_to=group2).cancelled
+
+
+def test_cancel_full(clidb, cli):
+    """Test --cancel with both --from and --to."""
+
+    group1 = StorageGroup.create(name="Group1")
+    node1 = StorageNode.create(name="Node1", group=group1)
+
+    group2 = StorageGroup.create(name="Group2")
+    node2 = StorageNode.create(name="Node2", group=group2)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    req11 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node1, group_to=group1, cancelled=0, completed=0
+    )
+    req12 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node1, group_to=group2, cancelled=0, completed=0
+    )
+    req21 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node2, group_to=group1, cancelled=0, completed=0
+    )
+    req22 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node2, group_to=group2, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File", "--cancel", "--from=Node1", "--to=Group2"])
+
+    assert not ArchiveFileCopyRequest.get(id=req11.id).cancelled
+    assert ArchiveFileCopyRequest.get(id=req12.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req21.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req22.id).cancelled