OCI: retry registry lookups and say why one failed

oci/remote/oci_installation_state.py

@@ -56,11 +56,39 @@ def version_from_environment(repository, environment):
     return found.get(expected)
 
 
+# What a registry answer means, read from skopeo's error output. Only the
+# cause is reported: the output itself can carry credentials.
+REGISTRY_ERRORS = (
+    (('toomanyrequests', 'too many requests', 'rate limit', 'registry 429'),
+     'The registry limited the requests. Try again later, or add an account for this registry.'),
+    (('unauthorized', 'authentication required', 'access to the resource is denied', '403 forbidden',
+      'registry 401', 'registry 403'),
+     'The registry refused access to the image.'),
+    (('manifest unknown', 'name unknown', '404 not found'),
+     'The image or its tag does not exist in the registry.'),
+    (('no such host', 'i/o timeout', 'connection refused', 'connection reset', 'network is unreachable',
+      'tls handshake timeout', 'context deadline exceeded', ': eof'),
+     'The registry could not be reached. Check the connection and try again.'),
+)
+
+
+def registry_error(errors):
+    text = errors.decode(errors='replace').casefold()
+    for patterns, message in REGISTRY_ERRORS:
+        if any(pattern in text for pattern in patterns):
+            return translate(message)
+    return None
+
+
 def command(*args):
-    result = subprocess.run(args, capture_output=True, timeout=120)
+    if args[:2] == ('skopeo', 'inspect'):
+        # A registry that fails for a moment is asked again before giving up.
+        args = args[:2] + ('--retry-times', '3', '--retry-delay', '5s') + args[2:]
+    result = subprocess.run(args, capture_output=True, timeout=300)
     if result.returncode:
         # Tool errors may contain credentials or environment values.
-        raise RuntimeError(f"{args[0]} {translate('failed with exit code')} {result.returncode}")
+        reason = registry_error(result.stderr) if args[0] == 'skopeo' else None
+        raise RuntimeError(reason or f"{args[0]} {translate('failed with exit code')} {result.returncode}")
     return result.stdout
 
 

oci/remote/oci_update_current.py

@@ -67,8 +67,10 @@ def resolve_archive(desired, config, current=None, check=None, required_digest=N
     if required_digest is not None and not re.fullmatch(r'sha256:[a-f0-9]{64}', required_digest):
         raise ValueError(translate('Invalid saved image digest'))
     lookup = repository(reference) + '@' + required_digest if required_digest else reference
-    code = ('import json,sys; from oci_installation_state import resolve_candidate; '
-            'print(json.dumps(resolve_candidate(sys.argv[1],sys.argv[2])))')
+    # The reason of a failure reaches the log as one line, not as a traceback.
+    code = ('import json,sys; from oci_installation_state import resolve_candidate\n'
+            'try: print(json.dumps(resolve_candidate(sys.argv[1],sys.argv[2])))\n'
+            'except (RuntimeError, ValueError, OSError) as error: sys.exit(str(error))')
     candidate = json.loads(run_quiet([sys.executable, '-c', code, lookup, architecture],
                                      translate('Could not query the image registry'), capture=True))
     digest = candidate['manifest_digest']

oci/tests/test_registry_errors.py

@@ -0,0 +1,81 @@
+"""A registry that fails for a moment is asked again, and a failure says why."""
+import ast
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+REMOTE = Path(__file__).resolve().parents[1] / 'remote'
+sys.path.insert(0, str(REMOTE))
+import oci_installation_state as state
+
+RATE = (b'time="2026-10-26T08:35:58Z" level=fatal msg="Error parsing image name \\"docker://gtstef/filebrowser:latest\\": '
+        b'reading manifest latest in docker.io/gtstef/filebrowser: toomanyrequests: You have reached your unauthenticated pull rate limit."')
+OFFLINE = (b'level=fatal msg="Error parsing image name: pinging container registry registry-1.docker.io: '
+           b'Get \\"https://registry-1.docker.io/v2/\\": dial tcp: lookup registry-1.docker.io: no such host"')
+DENIED = b'level=fatal msg="reading manifest latest in ghcr.io/example/private: unauthorized: authentication required"'
+MISSING = b'level=fatal msg="reading manifest nope in docker.io/library/alpine: manifest unknown"'
+FORBIDDEN = (b'level=fatal msg="Error parsing image name \\"docker://ghcr.io/example/private:latest\\": '
+             b'Requesting bearer token: invalid status code from registry 403 (Forbidden)"')
+
+
+def failed(stderr):
+    return subprocess.CompletedProcess([], 1, stdout=b'', stderr=stderr)
+
+
+class RegistryErrorTests(unittest.TestCase):
+    def test_an_image_lookup_is_retried_by_skopeo(self):
+        with patch.object(state.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, b'{}', b'')) as run:
+            state.command('skopeo', 'inspect', '--raw', 'docker://example/app:latest')
+        self.assertEqual(run.call_args.args[0], ('skopeo', 'inspect', '--retry-times', '3', '--retry-delay', '5s',
+                                                 '--raw', 'docker://example/app:latest'))
+
+    def test_other_commands_are_run_as_given(self):
+        with patch.object(state.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, b'', b'')) as run:
+            state.command('pct', 'config', '101')
+        self.assertEqual(run.call_args.args[0], ('pct', 'config', '101'))
+
+    def test_a_failure_names_its_cause_without_the_output_of_skopeo(self):
+        cases = {RATE: 'The registry limited the requests',
+                 OFFLINE: 'The registry could not be reached',
+                 DENIED: 'The registry refused access to the image',
+                 FORBIDDEN: 'The registry refused access to the image',
+                 MISSING: 'does not exist in the registry'}
+        for stderr, expected in cases.items():
+            with self.subTest(expected=expected), patch.object(state.subprocess, 'run', return_value=failed(stderr)), \
+                    patch.object(state, 'translate', side_effect=lambda text: text):
+                with self.assertRaises(RuntimeError) as raised:
+                    state.command('skopeo', 'inspect', '--raw', 'docker://example/app:latest')
+                self.assertIn(expected, str(raised.exception))
+                self.assertNotIn('docker.io', str(raised.exception))
+
+    def test_an_unknown_failure_keeps_the_exit_code_message(self):
+        with patch.object(state.subprocess, 'run', return_value=failed(b'level=fatal msg="something else"')), \
+                patch.object(state, 'translate', side_effect=lambda text: text):
+            with self.assertRaises(RuntimeError) as raised:
+                state.command('skopeo', 'inspect', '--raw', 'docker://example/app:latest')
+        self.assertEqual(str(raised.exception), 'skopeo failed with exit code 1')
+
+    def test_a_digest_that_contains_error_codes_is_not_a_cause(self):
+        stderr = b'level=fatal msg="reading blob sha256:4290404401429ab403: unexpected status"'
+        self.assertIsNone(state.registry_error(stderr))
+
+    def test_the_update_logs_the_reason_as_one_line(self):
+        tree = ast.parse((REMOTE / 'oci_update_current.py').read_text())
+        code = next(ast.literal_eval(node.value) for node in ast.walk(tree)
+                    if isinstance(node, ast.Assign) and getattr(node.targets[0], 'id', None) == 'code')
+        with tempfile.TemporaryDirectory() as folder:
+            Path(folder, 'oci_installation_state.py').write_text(
+                'def resolve_candidate(reference, architecture):\n'
+                '    raise RuntimeError("The registry could not be reached. Check the connection and try again.")\n')
+            result = subprocess.run([sys.executable, '-c', code, 'example/app:latest', 'amd64'],
+                                    capture_output=True, text=True, cwd=folder, env={'PYTHONPATH': folder})
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(result.stderr.strip(), 'The registry could not be reached. Check the connection and try again.')
+        self.assertNotIn('Traceback', result.stderr)
+
+
+if __name__ == '__main__':
+    unittest.main()