feat(archive-cache): added retry mechanism, and some code cleanups

rhodecode/apps/repository/views/repo_files.py

@@ -40,7 +40,7 b' from rhodecode.lib import diffs, helpers'
 from rhodecode.lib import audit_logger
 from rhodecode.lib.hash_utils import sha1_safe
 from rhodecode.lib.rc_cache.archive_cache import (
-    get_archival_cache_store, get_archival_config, ArchiveCacheLock, archive_iterator)
+    get_archival_cache_store, get_archival_config, ArchiveCacheGenerationLock, archive_iterator)
 from rhodecode.lib.str_utils import safe_bytes, convert_special_chars
 from rhodecode.lib.view_utils import parse_path_ref
 from rhodecode.lib.exceptions import NonRelativePathError
@@ -444,7 +444,8 b' class RepoFilesView(RepoAppView):'
                                             archive_at_path=at_path, cache_config=d_cache_conf)
                     except ImproperArchiveTypeError:
                         return _('Unknown archive type')
-            except ArchiveCacheLock:
+
+            except ArchiveCacheGenerationLock:
                 retry_after = round(random.uniform(0.3, 3.0), 1)
                 time.sleep(retry_after)
 

rhodecode/lib/rc_cache/archive_cache/__init__.py

@@ -20,7 +20,7 b' from .fanout_cache import get_archival_c'
 from .fanout_cache import get_archival_config
 
 from .utils import archive_iterator
-from .utils import ArchiveCacheLock
+from .utils import ArchiveCacheGenerationLock
 
 
 def includeme(config):

rhodecode/lib/rc_cache/archive_cache/fanout_cache.py

@@ -112,6 +112,11 b' class FileSystemCache:'
         self._index = index
         self._directory = directory
 
+    @property
+    def directory(self):
+        """Cache directory."""
+        return self._directory
+
     def _write_file(self, full_path, iterator, mode, encoding=None):
         full_dir, _ = os.path.split(full_path)
 
@@ -169,8 +174,17 b' class FileSystemCache:'
 
         return key, size, MODE_BINARY, filename, _metadata
 
-    def fetch(self, key) -> tuple[typing.BinaryIO, dict]:
+    def fetch(self, key, retry=False, retry_attempts=10) -> tuple[typing.BinaryIO, dict]:
+
+        if retry:
+            for attempt in range(retry_attempts):
+                if key in self:
+                    break
+                # we dind't find the key, wait 1s, and re-check
+                time.sleep(1)
+
         if key not in self:
+            log.exception('requested {key} not found in {self}', key, self)
             raise KeyError(key)
 
         key_file = self._get_keyfile(key)
@@ -180,7 +194,7 b' class FileSystemCache:'
         filename = metadata['filename']
 
         try:
-            return open(os.path.join(self._directory, filename), 'rb'), metadata
+            return open(os.path.join(self.directory, filename), 'rb'), metadata
         finally:
             # update usage stats, count and accessed
             metadata["access_count"] = metadata.get("access_count", 0) + 1
@@ -202,7 +216,7 b' class FileSystemCache:'
         sub_dir = os.path.join(hex_name[:2], hex_name[2:4])
         name = hex_name[4:] + '.archive_cache'
         filename = os.path.join(sub_dir, name)
-        full_path = os.path.join(self._directory, filename)
+        full_path = os.path.join(self.directory, filename)
         return filename, full_path
 
     def hash(self, key):
@@ -225,6 +239,9 b' class FileSystemCache:'
         key_file = self._get_keyfile(key)
         return os.path.exists(key_file)
 
+    def __repr__(self):
+        return f'FileSystemCache(index={self._index}, dir={self.directory})'
+
 
 class FanoutCache:
     """Cache that shards keys and values."""
@@ -262,6 +279,11 b' class FanoutCache:'
         )
         self._hash = self._shards[0].hash
 
+    @property
+    def directory(self):
+        """Cache directory."""
+        return self._directory
+
     def get_lock(self, lock_key):
         return GenerationLock(lock_key, self._locking_url)
 
@@ -274,11 +296,11 b' class FanoutCache:'
         shard = self._get_shard(key)
         return shard.store(key, value_reader, metadata)
 
-    def fetch(self, key):
+    def fetch(self, key, retry=False, retry_attempts=10):
         """Return file handle corresponding to `key` from cache.
         """
         shard = self._get_shard(key)
-        return shard.fetch(key)
+        return shard.fetch(key, retry=retry, retry_attempts=retry_attempts)
 
     def has_key(self, key):
         """Return `True` if `key` matching item is found in cache.
@@ -326,9 +348,9 b' class FanoutCache:'
         data = []
         cnt = 1
         for shard in self._shards:
-            for key_file in os.listdir(shard._directory):
+            for key_file in os.listdir(shard.directory):
                 if key_file.endswith('.key'):
-                    key_file_path = os.path.join(shard._directory, key_file)
+                    key_file_path = os.path.join(shard.directory, key_file)
                     with open(key_file_path, 'rb') as f:
                         metadata = json.loads(f.read())
 

rhodecode/lib/rc_cache/archive_cache/lock.py

@@ -18,7 +18,7 b''
 
 import redis
 from ..._vendor import redis_lock
-from .utils import ArchiveCacheLock
+from .utils import ArchiveCacheGenerationLock
 
 
 class GenerationLock:
@@ -53,7 +53,7 b' class GenerationLock:'
     def __enter__(self):
         acquired = self.lock.acquire(blocking=False)
         if not acquired:
-            raise ArchiveCacheLock('Failed to create a lock')
+            raise ArchiveCacheGenerationLock('Failed to create a lock')
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.lock.release()

rhodecode/lib/rc_cache/archive_cache/utils.py

@@ -19,7 +19,7 b''
 import os
 
 
-class ArchiveCacheLock(Exception):
+class ArchiveCacheGenerationLock(Exception):
     pass