Repository: radiocosmology/alpenhorn

Issue:
Alpenhorn cancels AFCR for source with `has_file='M'`
Should probably just skip it until after the check.

Discussion:

Solution patch:
diff --git a/alpenhorn/db/storage.py b/alpenhorn/db/storage.py
index 2fc243aa0..022783196 100644
--- a/alpenhorn/db/storage.py
+++ b/alpenhorn/db/storage.py
@@ -240,8 +240,8 @@ def named_copy_tracked(self, acqname: str, filename: str) -> bool:
         # Check has_file
         return copy.has_file != "N"
 
-    def filecopy_present(self, file: ArchiveFile) -> bool:
-        """Is a copy of ArchiveFile `file` present on this node?
+    def filecopy_state(self, file: ArchiveFile) -> str:
+        """What is the state of `file` on this node?
 
         Parameters
         ----------
@@ -250,22 +250,25 @@ def filecopy_present(self, file: ArchiveFile) -> bool:
 
         Returns
         -------
-        filecopy_present : bool
-            True if there is an ArchiveFileCopy of `file`r
-            with `has_file=='Y'` on this node.  False otherwise.
+        filecopy_state : str
+            One of:
+            - 'Y' file copy exists
+            - 'X' file copy is corrupt
+            - 'M' file copy needs to be checked
+            - 'N' file copy does not exist
         """
         from .archive import ArchiveFileCopy
 
         try:
-            ArchiveFileCopy.get(
+            copy = ArchiveFileCopy.get(
                 ArchiveFileCopy.file == file,
                 ArchiveFileCopy.node == self,
-                ArchiveFileCopy.has_file == "Y",
             )
+            return copy.has_file
         except pw.DoesNotExist:
-            return False
+            pass
 
-        return True
+        return "N"
 
     def get_total_gb(self) -> float:
         """Sum the size in GiB of all files on this node.
diff --git a/alpenhorn/io/_default_asyncs.py b/alpenhorn/io/_default_asyncs.py
index 16b926f50..344471e47 100644
--- a/alpenhorn/io/_default_asyncs.py
+++ b/alpenhorn/io/_default_asyncs.py
@@ -8,6 +8,7 @@
 import shutil
 import logging
 import pathlib
+import peewee as pw
 
 from . import ioutil
 from ..db import ArchiveFileCopy, ArchiveFileCopyRequest, utcnow
@@ -335,21 +336,8 @@ def group_search_async(
 
         # ready == False is the safe option here: copy will be readied
         # during the subsequent check if needed.
-        count = (
-            ArchiveFileCopy.update(
-                has_file="M",
-                wants_file="Y",
-                ready=False,
-                last_update=utcnow(),
-            )
-            .where(
-                ArchiveFileCopy.file == req.file,
-                ArchiveFileCopy.node == node.db,
-            )
-            .execute()
-        )
-        if count == 0:
-            # Create new copy
+        try:
+            # Try to create a new copy
             ArchiveFileCopy.create(
                 file=req.file,
                 node=node.db,
@@ -358,6 +346,17 @@ def group_search_async(
                 ready=False,
                 size_b=node.io.filesize(req.file.path, actual=True),
             )
+        except pw.IntegrityError:
+            # Copy already exists, just update the existing
+            ArchiveFileCopy.update(
+                has_file="M",
+                wants_file="Y",
+                ready=False,
+                last_update=utcnow(),
+            ).where(
+                ArchiveFileCopy.file == req.file,
+                ArchiveFileCopy.node == node.db,
+            ).execute()
         return
 
     # Otherwise, escallate to groupio.pull_force to actually perform the pull
diff --git a/alpenhorn/server/update.py b/alpenhorn/server/update.py
index 34380d527..27df4d081 100644
--- a/alpenhorn/server/update.py
+++ b/alpenhorn/server/update.py
@@ -469,14 +469,20 @@ def update(self) -> None:
                 ArchiveFileCopyRequest.cancelled == 0,
                 ArchiveFileCopyRequest.node_from == self.db,
             ):
-                if self.db.filecopy_present(req.file):
+                state = self.db.filecopy_state(req.file)
+                if state == "Y":
                     self._io_happened = True
                     self.io.ready_pull(req)
                 else:
+                    reasons = {
+                        "N": "not present",
+                        "M": "needs check",
+                        "X": "corrupt",
+                    }
                     log.info(
                         "Ignoring ready request for "
                         f"{req.file.acq.name}/{req.file.name} "
-                        f"on node {self.name}: not present."
+                        f"on node {self.name}: {reasons[state]}."
                     )
 
             self._updated = True
@@ -629,8 +635,10 @@ def update_pull(self, req: ArchiveFileCopyRequest) -> None:
             )
             return
 
-        # If the source file doesn't exist, cancel the request.
-        if not req.node_from.filecopy_present(req.file):
+        # If the source file doesn't exist, cancel the request.  If the
+        # source is suspect, skip the request.
+        state = req.node_from.filecopy_state(req.file)
+        if state == "N" or state == "X":
             log.warning(
                 f"Cancelling request for {req.file.acq.name}/{req.file.name}:"
                 f" not available on node {req.node_from.name}. "
@@ -640,6 +648,12 @@ def update_pull(self, req: ArchiveFileCopyRequest) -> None:
                 ArchiveFileCopyRequest.id == req.id
             ).execute()
             return
+        elif state == "M":
+            log.info(
+                f"Skipping request for {req.file.acq.name}/{req.file.name}:"
+                f" source needs check on node {req.node_from.name}."
+            )
+            return
 
         # If the source file is not ready, skip the request.
         node_from = RemoteNode(req.node_from)


Test patch:
diff --git a/tests/db/test_storage.py b/tests/db/test_storage.py
index 70b754f64..5c7582863 100644
--- a/tests/db/test_storage.py
+++ b/tests/db/test_storage.py
@@ -282,19 +282,27 @@ def test_namedcopytracked(simplegroup, storagenode, simplefile, archivefilecopy)
     assert node.named_copy_tracked(acqname, filename) is False
 
 
-def test_copypresent(simplegroup, storagenode, simplefile, archivefilecopy):
-    """Test StorageNode.filecopy_present()."""
+def test_node_copystate(simplegroup, storagenode, simplefile, archivefilecopy):
+    """Test StorageNode.filecopy_state()."""
 
     node = storagenode(name="present", group=simplegroup)
     archivefilecopy(file=simplefile, node=node, has_file="Y")
-    assert node.filecopy_present(simplefile) is True
+    assert node.filecopy_state(simplefile) == "Y"
+
+    node = storagenode(name="suspect", group=simplegroup)
+    archivefilecopy(file=simplefile, node=node, has_file="M")
+    assert node.filecopy_state(simplefile) == "M"
 
     node = storagenode(name="corrupt", group=simplegroup)
     archivefilecopy(file=simplefile, node=node, has_file="X")
-    assert node.filecopy_present(simplefile) is False
+    assert node.filecopy_state(simplefile) == "X"
+
+    node = storagenode(name="removed", group=simplegroup)
+    archivefilecopy(file=simplefile, node=node, has_file="N")
+    assert node.filecopy_state(simplefile) == "N"
 
     node = storagenode(name="missing", group=simplegroup)
-    assert node.filecopy_present(simplefile) is False
+    assert node.filecopy_state(simplefile) == "N"
 
 
 def test_allfiles(simplenode, simpleacq, archivefile, archivefilecopy):