fix(mirror): refresh a mirrored tag that upstream re-pushed

Fork CI run 41147980556 pgAdmin#0 failed in sys-ctl-hlth-runtime with
"stalwart_webmail: PHP 8.4.26 is below the security floor 8.4.27".
The deploy ran the fork's GHCR mirror of roundcube/roundcubemail:1.7.5-
apache, copied on 10-19 with PHP 8.4.26, while upstream re-pushed the
same tag on 10-27 with PHP 8.4.27. Branch CI runs the sync with
--only-missing, which skipped the tag because it existed ("destination
exists, skipping") without comparing digests, and the full refresh only
runs on the fork's default branch, which still pins 1.7.4.

- utils/docker/registry.py: manifest_digest() reads the
  Docker-Content-Digest of a reference with a HEAD request, which does
  not count against Docker Hub's pull allowance.
- GHCRProvider.is_stale() reports a mirrored tag as stale only when both
  registries answer with a digest and the two differ.
- Under --only-missing an existing tag is re-copied when it is stale; a
  failed re-copy is reported on stderr with the image and keeps the
  existing copy, so one refresh cannot block every deploy.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/unit/python/cli/contributing/mirror/test_providers.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import unittest
+from unittest.mock import patch
 
 from cli.contributing.mirror.model import ImageRef
 from cli.contributing.mirror.providers import GHCRProvider
@@ -60,5 +61,51 @@ class TestGHCRProviderImageBase(unittest.TestCase):
         self.assertEqual(provider.repository, "myrepo")
 
 
+class TestGHCRProviderIsStale(unittest.TestCase):
+    UPSTREAM = "docker.io/roundcube/roundcubemail"
+    MIRROR = "ghcr.io/acme/myrepo/mirror/docker.io/roundcube/roundcubemail"
+
+    def setUp(self) -> None:
+        self.image = ImageRef(
+            role="web-app-stalwart",
+            service="webmail",
+            name="roundcube/roundcubemail",
+            version="1.7.5-apache",
+            source="docker.io/roundcube/roundcubemail:1.7.5-apache",
+            registry="docker.io",
+        )
+        self.provider = GHCRProvider("acme", "myrepo")
+
+    def _is_stale(self, digests: dict) -> bool:
+        def fake(image: str, reference: str) -> str | None:
+            self.assertEqual(reference, "1.7.5-apache")
+            return digests[image]
+
+        with patch(
+            "cli.contributing.mirror.providers.manifest_digest", side_effect=fake
+        ):
+            return self.provider.is_stale(self.image)
+
+    def test_upstream_repushed_tag_makes_the_copy_stale(self) -> None:
+        self.assertTrue(
+            self._is_stale({self.UPSTREAM: "sha256:new", self.MIRROR: "sha256:old"})
+        )
+
+    def test_matching_digests_are_current(self) -> None:
+        self.assertFalse(
+            self._is_stale({self.UPSTREAM: "sha256:same", self.MIRROR: "sha256:same"})
+        )
+
+    def test_indeterminate_upstream_counts_as_current(self) -> None:
+        self.assertFalse(
+            self._is_stale({self.UPSTREAM: None, self.MIRROR: "sha256:old"})
+        )
+
+    def test_indeterminate_mirror_counts_as_current(self) -> None:
+        self.assertFalse(
+            self._is_stale({self.UPSTREAM: "sha256:new", self.MIRROR: None})
+        )
+
+
 if __name__ == "__main__":  # pragma: no cover
     unittest.main()

tests/unit/python/cli/contributing/mirror/test_sync.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import subprocess
 import unittest
 from unittest.mock import patch
 
@@ -8,20 +9,20 @@ from cli.contributing.mirror.model import ImageRef
 
 
 class TestMirrorSync(unittest.TestCase):
-    def test_only_missing_skips_existing_destination(self) -> None:
-        image = ImageRef(
-            role="web-app-nextcloud",
-            service="app",
-            name="nextcloud",
-            version="31.0.0",
-            source="docker.io/library/nextcloud:31.0.0",
-            registry="docker.io",
-        )
+    IMAGE = ImageRef(
+        role="web-app-nextcloud",
+        service="app",
+        name="nextcloud",
+        version="31.0.0",
+        source="docker.io/library/nextcloud:31.0.0",
+        registry="docker.io",
+    )
 
+    def _run_only_missing(self, *, stale: bool, mirror_error=None):
         with (
             patch(
                 "cli.contributing.mirror.sync.__main__.iter_role_images",
-                return_value=[image],
+                return_value=[self.IMAGE],
             ),
             patch(
                 "cli.contributing.mirror.sync.__main__.iter_runner_images",
@@ -35,7 +36,12 @@ class TestMirrorSync(unittest.TestCase):
             patch.object(
                 sync_main.GHCRProvider, "tag_exists", return_value=True
             ) as mock_tag_exists,
-            patch.object(sync_main.GHCRProvider, "mirror") as mock_mirror,
+            patch.object(
+                sync_main.GHCRProvider, "is_stale", return_value=stale
+            ) as mock_is_stale,
+            patch.object(
+                sync_main.GHCRProvider, "mirror", side_effect=mirror_error
+            ) as mock_mirror,
             patch(
                 "sys.argv",
                 [
@@ -49,11 +55,30 @@ class TestMirrorSync(unittest.TestCase):
             ),
         ):
             result = sync_main.main()
+        mock_tag_exists.assert_called_once_with(self.IMAGE)
+        mock_is_stale.assert_called_once_with(self.IMAGE)
+        return result, mock_mirror
+
+    def test_only_missing_skips_current_destination(self) -> None:
+        result, mock_mirror = self._run_only_missing(stale=False)
 
         self.assertEqual(result, 0)
-        mock_tag_exists.assert_called_once_with(image)
         mock_mirror.assert_not_called()
 
+    def test_only_missing_refreshes_a_stale_destination(self) -> None:
+        result, mock_mirror = self._run_only_missing(stale=True)
+
+        self.assertEqual(result, 0)
+        mock_mirror.assert_called_once_with(self.IMAGE)
+
+    def test_failed_refresh_keeps_the_existing_copy_without_failing(self) -> None:
+        error = subprocess.CalledProcessError(1, ["skopeo", "copy"])
+
+        result, mock_mirror = self._run_only_missing(stale=True, mirror_error=error)
+
+        self.assertEqual(result, 0)
+        mock_mirror.assert_called_once_with(self.IMAGE)
+
     def test_runner_images_are_mirrored_with_the_role_images(self) -> None:
         image = ImageRef(
             role="runner",