Repository: radiocosmology/alpenhorn

Issue:
Daemon should be able to delete suspect/corrupt files
i.e.  we should attempt to delete `has_file='X'/'M', wants_file='N'` copies.

https://github.com/radiocosmology/alpenhorn/blob/56348b109b35ceeaa62b8ca3dd12eaccaaf4e903/alpenhorn/daemon/update.py#L392

* Corollary: daemon shouldn't attempt to check files with `has_file='M', wants_file='N'`, because we're just going to delete them.
* Also need a client option to be able to flag them: `node clean --include-bad`

Discussion:

Patch:
diff --git a/alpenhorn/cli/node/clean.py b/alpenhorn/cli/node/clean.py
index ccf001d..10a7f3b 100644
--- a/alpenhorn/cli/node/clean.py
+++ b/alpenhorn/cli/node/clean.py
@@ -1,5 +1,7 @@
 """alpenhorn node clean"""
 
+from __future__ import annotations
+
 import click
 import datetime
 import peewee as pw
@@ -25,6 +27,7 @@ def _run_query(
     acq,
     archive_ok,
     days: datetime.datetime,
+    has_file: pw.Expression,
     size: int,
     target,
     clean_goal: str,
@@ -51,6 +54,11 @@ def _run_query(
         Replaces the `now` and `cancel` parameters.
     days:
         Converted to datetime
+    has_file:
+        peewee.Expression instance for the ArchiveFileCopy.has_file constraint.
+        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.
     size:
         Converted to bytes
     warn:
@@ -117,9 +125,7 @@ def _run_query(
                 raise click.ClickException("No such acquisition: " + acqname)
 
         # Find all candidate file copies
-        query = ArchiveFileCopy.select().where(
-            ArchiveFileCopy.node == node, ArchiveFileCopy.has_file == "Y"
-        )
+        query = ArchiveFileCopy.select().where(ArchiveFileCopy.node == node, has_file)
 
         # Join to ArchiveFile, if necessary
         if acqs or days or size:
@@ -315,6 +321,11 @@ def _run_query(
     help="Force cleaning (skips confirmation).  Incompatible with --check",
     is_flag=True,
 )
+@click.option(
+    "--include-bad",
+    help="Include suspect and corrupt files in the operation.",
+    is_flag=True,
+)
 @click.option("--now", "-n", help="Force immediate removal.", is_flag=True)
 @click.option(
     "--size",
@@ -331,7 +342,20 @@ def _run_query(
     "files in all GROUPs.",
 )
 @click.pass_context
-def clean(ctx, name, acq, archive_ok, cancel, check, days, force, now, size, target):
+def clean(
+    ctx,
+    name,
+    acq,
+    archive_ok,
+    cancel,
+    check,
+    days,
+    force,
+    include_bad,
+    now,
+    size,
+    target,
+):
     """Remove files from a Storage Node.
 
     There are two ways alpenhorn can schedule a file for cleaning:
@@ -419,11 +443,27 @@ def clean(ctx, name, acq, archive_ok, cancel, check, days, force, now, size, tar
     else:
         clean_goal = "M"
 
+    # Determine the has_file constraint
+    if include_bad:
+        has_file = ArchiveFileCopy.has_file != "N"
+    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, size, target, clean_goal, warn=True
+            False,
+            ctx,
+            name,
+            acq,
+            archive_ok,
+            days,
+            has_file,
+            size,
+            target,
+            clean_goal,
+            warn=True,
         )
 
         # With --check, we're done
@@ -439,5 +479,15 @@ def clean(ctx, name, acq, archive_ok, cancel, check, days, force, now, size, tar
     # Execute the query, if not in --check mode
     if not check:
         _run_query(
-            True, ctx, name, acq, archive_ok, days, size, target, clean_goal, warn=force
+            True,
+            ctx,
+            name,
+            acq,
+            archive_ok,
+            days,
+            has_file,
+            size,
+            target,
+            clean_goal,
+            warn=force,
         )
diff --git a/alpenhorn/daemon/update.py b/alpenhorn/daemon/update.py
index 1823ba2..e768ab2 100644
--- a/alpenhorn/daemon/update.py
+++ b/alpenhorn/daemon/update.py
@@ -389,7 +389,7 @@ class UpdateableNode(updateable_base):
             .where(
                 dfclause,
                 ArchiveFileCopy.node == self.db,
-                ArchiveFileCopy.has_file == "Y",
+                ArchiveFileCopy.has_file != "N",
             )
             .order_by(ArchiveFileCopy.id)
         ):
@@ -451,7 +451,9 @@ class UpdateableNode(updateable_base):
 
             # Check the integrity of any questionable files (has_file=M)
             for copy in ArchiveFileCopy.select().where(
-                ArchiveFileCopy.node == self.db, ArchiveFileCopy.has_file == "M"
+                ArchiveFileCopy.node == self.db,
+                ArchiveFileCopy.has_file == "M",
+                ArchiveFileCopy.wants_file != "N",
             ):
                 log.info(
                     f'Checking copy "{copy.file.acq.name}/{copy.file.name}" '


Test patch:
diff --git a/tests/cli/node/test_clean.py b/tests/cli/node/test_clean.py
index e4511ae..b7b1a95 100644
--- a/tests/cli/node/test_clean.py
+++ b/tests/cli/node/test_clean.py
@@ -97,16 +97,22 @@ def test_run(clidb, cli):
     group = StorageGroup.create(name="Group")
     node = StorageNode.create(name="NODE", group=group, storage_type="F")
     acq = ArchiveAcq.create(name="Acq")
-    fileY = ArchiveFile.create(name="FileY", acq=acq, size_b=2345)
-    ArchiveFileCopy.create(node=node, file=fileY, has_file="Y", wants_file="Y")
-    fileN = ArchiveFile.create(name="FileN", acq=acq, size_b=3456)
-    ArchiveFileCopy.create(node=node, file=fileN, has_file="Y", wants_file="N")
+    fileYY = ArchiveFile.create(name="FileYY", acq=acq, size_b=2345)
+    ArchiveFileCopy.create(node=node, file=fileYY, has_file="Y", wants_file="Y")
+    fileYN = ArchiveFile.create(name="FileYN", acq=acq, size_b=3456)
+    ArchiveFileCopy.create(node=node, file=fileYN, has_file="Y", wants_file="N")
+    fileMY = ArchiveFile.create(name="FileMY", acq=acq, size_b=3456)
+    ArchiveFileCopy.create(node=node, file=fileMY, has_file="M", wants_file="Y")
+    fileXY = ArchiveFile.create(name="FileXY", acq=acq, size_b=3456)
+    ArchiveFileCopy.create(node=node, file=fileXY, has_file="X", wants_file="Y")
 
     cli(0, ["node", "clean", "NODE"], input="Y\n")
 
-    # Clean should have run, but only on the 'Y' file
-    assert ArchiveFileCopy.get(node=node, file=fileY).wants_file == "M"
-    assert ArchiveFileCopy.get(node=node, file=fileN).wants_file == "N"
+    # Clean should have run, but only on the 'YY' file
+    assert ArchiveFileCopy.get(node=node, file=fileYY).wants_file == "M"
+    assert ArchiveFileCopy.get(node=node, file=fileYN).wants_file == "N"
+    assert ArchiveFileCopy.get(node=node, file=fileMY).wants_file == "Y"
+    assert ArchiveFileCopy.get(node=node, file=fileXY).wants_file == "Y"
 
 
 def test_now(clidb, cli):
@@ -463,7 +469,7 @@ def test_size_part(clidb, cli):
     assert ArchiveFileCopy.get(node=node, file=file4).wants_file == "Y"
 
 
-def test_size_part(clidb, cli):
+def test_size_done(clidb, cli):
     """Test clean with --size already satisfied"""
 
     group = StorageGroup.create(name="Group1")
@@ -487,3 +493,26 @@ def test_size_part(clidb, cli):
     assert ArchiveFileCopy.get(node=node, file=file2).wants_file == "N"
     assert ArchiveFileCopy.get(node=node, file=file3).wants_file == "Y"
     assert ArchiveFileCopy.get(node=node, file=file4).wants_file == "Y"
+
+
+def test_include_bad(clidb, cli):
+    """Test clean with --include-bad"""
+
+    group = StorageGroup.create(name="Group1")
+    node = StorageNode.create(name="NODE", group=group, storage_type="F")
+
+    acq = ArchiveAcq.create(name="Acq")
+
+    fileY = ArchiveFile.create(name="FileY", acq=acq)
+    ArchiveFileCopy.create(node=node, file=fileY, has_file="Y", wants_file="N")
+    fileM = ArchiveFile.create(name="FileM", acq=acq)
+    ArchiveFileCopy.create(node=node, file=fileM, has_file="M", wants_file="N")
+    fileX = ArchiveFile.create(name="FileX", acq=acq)
+    ArchiveFileCopy.create(node=node, file=fileX, has_file="X", wants_file="N")
+
+    cli(0, ["node", "clean", "NODE", "--force", "--include-bad"])
+
+    # All the files are cleaned
+    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"
diff --git a/tests/daemon/test_update_node.py b/tests/daemon/test_update_node.py
index be0cdc2..636ddd1 100644
--- a/tests/daemon/test_update_node.py
+++ b/tests/daemon/test_update_node.py
@@ -376,6 +376,24 @@ def test_update_delete_transfer_pending(
     mock_delete.assert_not_called()
 
 
+def test_update_delete_bad_file(
+    unode, simplegroup, simpleacq, archivefile, archivefilecopy
+):
+    """update_delete() should attempt to delete bad files."""
+
+    fileM = archivefile(name="fileM", acq=simpleacq)
+    copyM = archivefilecopy(file=fileM, node=unode.db, has_file="M", wants_file="N")
+    fileX = archivefile(name="fileX", acq=simpleacq)
+    copyX = archivefilecopy(file=fileX, node=unode.db, has_file="X", wants_file="N")
+
+    mock_delete = MagicMock()
+    with patch.object(unode.io, "delete", mock_delete):
+        unode.update_delete()
+
+    # Both files were deleted
+    mock_delete.assert_called_once_with([copyM, copyX])
+
+
 def test_update_node_run(
     unode,
     queue,
@@ -387,9 +405,17 @@ def test_update_node_run(
 ):
     """Test running UpdateableNode.update_node."""
 
-    # Make something to check
-    badfile = archivefile(name="check_me", acq=simpleacq)
-    copy = archivefilecopy(node=unode.db, file=badfile, has_file="M")
+    # Make some things to check
+    badfileY = archivefile(name="check_me", acq=simpleacq)
+    badcopyY = archivefilecopy(
+        node=unode.db, file=badfileY, has_file="M", wants_file="Y"
+    )
+
+    # This one should be skipped (because it's about to be deleted)
+    badfileN = archivefile(name="dont_check_me", acq=simpleacq)
+    badcopyN = archivefilecopy(
+        node=unode.db, file=badfileN, has_file="M", wants_file="N"
+    )
 
     # And something to ready for a pull
     goodfile = archivefile(name="ready_me", acq=simpleacq)
@@ -432,9 +458,10 @@ def test_update_node_run(
 
     # Check I/O calls
     calls = list(mock.mock_calls)
-    assert len(calls) == 4
+    assert len(calls) == 5
     assert call.bytes_avail(fast=False) in calls
-    assert call.check(copy) in calls
+    assert call.check(badcopyY) in calls
+    assert call.delete([badcopyN]) in calls
     assert call.ready_pull(afcr_good) in calls
 
     # Check remote.pull_ready calls