feat(DEVC-1286): update migration logic

Author: kossman

src/corva/cache_adapter.py

@@ -9,40 +9,33 @@
 import redis
 import semver
 
+from version import VERSION
+
 
 class CacheRepositoryProtocol(Protocol):
     def set(
         self,
         key: str,
         value: str,
         ttl: int,
-    ) -> None:
-        ...
+    ) -> None: ...
 
-    def set_many(self, data: Sequence[Tuple[str, str, int]]) -> None:
-        ...
+    def set_many(self, data: Sequence[Tuple[str, str, int]]) -> None: ...
 
-    def get(self, key: str) -> Optional[str]:
-        ...
+    def get(self, key: str) -> Optional[str]: ...
 
-    def get_many(self, keys: Sequence[str]) -> Dict[str, Optional[str]]:
-        ...
+    def get_many(self, keys: Sequence[str]) -> Dict[str, Optional[str]]: ...
 
-    def get_all(self) -> Dict[str, str]:
-        ...
+    def get_all(self) -> Dict[str, str]: ...
 
-    def delete(self, key: str) -> None:
-        ...
+    def delete(self, key: str) -> None: ...
 
-    def delete_many(self, keys: Sequence[str]) -> None:
-        ...
+    def delete_many(self, keys: Sequence[str]) -> None: ...
 
-    def delete_all(self) -> None:
-        ...
+    def delete_all(self) -> None: ...
 
 
 class RedisRepository:
-
     def __init__(self, hash_name: str, client: redis.Redis):
         self.hash_name = hash_name
         self.client = client
@@ -85,45 +78,49 @@ def delete_all(self) -> None:
 
 class HashMigrator:
     MINIMUM_ALLOWED_REDIS_SERVER = semver.Version(major=7, minor=4, patch=0)
+    NEW_HASH_PREFIX = "/new"
 
     def __init__(self, hash_name: str, client: redis.Redis):
         self.hash_name = hash_name
-        self.zset_name = f'{hash_name}.EXPIREAT'
+        self.zset_name = f"{hash_name}.EXPIREAT"
         self.client = client
 
-    def _proper_redis_server_version(self) -> bool:
+    def check_redis_server_version(self) -> None:
         # Require Redis 7.4+ for per-field TTL commands
-        redis_version_str = self.client.info(section="server").get("redis_version")
-
-        if not redis_version_str:
-            return False
-
+        redis_version_str = self.client.info(section="server")["redis_version"]
         server_version = semver.Version.parse(version=redis_version_str)
 
         if server_version < self.MINIMUM_ALLOWED_REDIS_SERVER:
-            return False
-
-        return True
+            raise RuntimeError(
+                f"Redis server version {server_version} "
+                f"less then {self.MINIMUM_ALLOWED_REDIS_SERVER} -> "
+                f"incompatible with used python SDK version `{VERSION}`")
 
     def run(self) -> bool:
-        """Migrate from old Lua+ZSET per-field TTL to Redis built-in per-field TTL.
+        """Prepare parallel new-style cache while keeping legacy structures.
 
-        Safe to call multiple times. Behavior:
-          - If the legacy ZSET ("<hash>.EXPIREAT") does not exist → return False.
-          - Requires Redis server version >= 7.4.0 (built-in per-field hash TTLs).
+        Behavior (idempotent):
+          - If legacy ZSET ("<hash>.EXPIREAT") does not exist → return False.
+          - Requires Redis server >= 7.4.0 for per-field hash TTL commands.
+          - Creates a new hash key with prefix NEW_HASH_PREFIX + hash_name.
           - For each zset member (field → absolute ms deadline):
-              • Past deadline → HDEL the field.
-              • Future deadline → set per-field TTL via HPEXPIRE (milliseconds).
-          - After processing completes: PERSIST the hash and DELETE the legacy ZSET.
+              • Past deadline → do not copy field to new hash
+              • Future deadline → copy current value from legacy hash to new hash and
+                set per-field TTL via HPEXPIRE (milliseconds) on the new hash.
+          - Legacy hash and legacy ZSET are preserved intact to allow rollback.
 
-        Returns True if migration was attempted on a supported server (i.e., legacy
-        ZSET existed); otherwise False.
+        Returns True if the new-style cache was created during this run
         """
+        self.check_redis_server_version()
+
         # Legacy structure must exist; otherwise nothing to do
         if not self.client.exists(self.zset_name):
             return False
 
-        if not self._proper_redis_server_version():
+        new_hash_name = self.NEW_HASH_PREFIX + self.hash_name
+
+        # If new hash already exists, consider migration already done
+        if self.client.exists(new_hash_name):
             return False
 
         from corva import Logger
@@ -132,27 +129,35 @@ def run(self) -> bool:
         sec, micro = self.client.time()
         now_ms = int(sec) * 1000 + int(micro) // 1000
 
-        # Queue all operations in a single pipeline and execute once
+        # Create pipeline for batched ops on the NEW hash
         pipe = self.client.pipeline()
 
+        # Ensure the new hash key exists
+        # Copy fields from old hash into the new one based on ZSET deadlines
         for field, score in self.client.zscan_iter(self.zset_name):
             # score is the absolute deadline in ms
             deadline_ms = int(float(score))
-            ttl_ms = deadline_ms - now_ms
-            if ttl_ms <= 0:
-                pipe.hdel(self.hash_name, field)
-            else:
-                pipe.execute_command(
-                    "HPEXPIRE", self.hash_name, ttl_ms, "FIELDS", 1, field
-                )
+            remaining_ttl_ms = deadline_ms - now_ms
+            if remaining_ttl_ms <= 0:
+                continue
+
+            value = self.client.hget(self.hash_name, field)
+            if value is None:
+                # No value to copy (may have been removed already)
+                continue
+
+            # Write to the new hash and apply per-field TTL there
+            pipe.hset(new_hash_name, field, value)
+            pipe.execute_command(
+                "HPEXPIRE", new_hash_name, remaining_ttl_ms, "FIELDS", 1, field
+            )
 
         # Execute queued field operations (no-op if nothing queued)
         pipe.execute()
 
-        # Remove key-level TTL and legacy ZSET now that fields are migrated
-        self.client.persist(self.hash_name)
-        self.client.delete(self.zset_name)
-
-        # Migration was attempted because legacy ZSET existed
-        Logger.info(f"Migration success: hash_name = '{self.hash_name}'")
+        # Do NOT modify/persist legacy structures; keep them for rollback
+        Logger.info(
+            f"Migration prepared parallel cache: legacy='{self.hash_name}', "
+            f"new='{new_hash_name}'"
+        )
         return True

src/corva/service/cache_sdk.py

@@ -17,32 +17,24 @@
 
 
 class UserCacheSdkProtocol(Protocol):
-    def set(self, key: str, value: str, ttl: int = ...) -> None:
-        ...
+    def set(self, key: str, value: str, ttl: int = ...) -> None: ...
 
     def set_many(
         self, data: Sequence[Union[Tuple[str, str], Tuple[str, str, int]]]
-    ) -> None:
-        ...
+    ) -> None: ...
 
-    def get(self, key: str) -> Optional[str]:
-        ...
+    def get(self, key: str) -> Optional[str]: ...
 
-    def get_many(self, keys: Sequence[str]) -> Dict[str, Optional[str]]:
-        ...
+    def get_many(self, keys: Sequence[str]) -> Dict[str, Optional[str]]: ...
 
-    def get_all(self) -> Dict[str, str]:
-        ...
+    def get_all(self) -> Dict[str, str]: ...
 
     # TODO: remove asterisk in v2 - it was added for backward compatibility
-    def delete(self, *, key: str) -> None:
-        ...
+    def delete(self, *, key: str) -> None: ...
 
-    def delete_many(self, keys: Sequence[str]) -> None:
-        ...
+    def delete_many(self, keys: Sequence[str]) -> None: ...
 
-    def delete_all(self) -> None:
-        ...
+    def delete_all(self) -> None: ...
 
 
 class UserRedisSdk:
@@ -70,17 +62,15 @@ def __init__(
         elif redis_client is None:
             redis_client = redis.Redis.from_url(url=redis_dsn, decode_responses=True)
 
+        migrator = HashMigrator(hash_name, redis_client)
+        migrator.run()
+        hash_name = HashMigrator.NEW_HASH_PREFIX + hash_name
+
         self.cache_repo = cache_adapter.RedisRepository(
             hash_name=hash_name,
             client=cast(redis.Redis, redis_client),
         )
 
-        try:
-            migrator = HashMigrator(hash_name, redis_client)
-            migrator.run()
-        except Exception:
-            pass
-
     def set(self, key: str, value: str, ttl: int = SIXTY_DAYS) -> None:
         self.cache_repo.set(key=key, value=value, ttl=ttl)
 
@@ -112,9 +102,3 @@ def delete_many(self, keys: Sequence[str]) -> None:
 
     def delete_all(self) -> None:
         self.cache_repo.delete_all()
-
-
-
-
-
-

src/plugin.py

@@ -36,7 +36,6 @@ def pytest_load_initial_conftests(args, early_config, parser):
         'CACHE_URL': 'redis://localhost:6379',
         'APP_KEY': f'{provider}.test-app-name',
         'PROVIDER': provider,
-        'FAKEREDIS_LUA_VERSION': "5.2",
         **os.environ,  # override env values if provided by user
     }
     os.environ.update(env)

tests/conftest.py

@@ -1,16 +1,27 @@
-from typing import cast
+from typing import Iterable, cast
 
 import pytest
 from fakeredis import FakeRedis
 from redis import Redis
 
+from corva import cache_adapter
 from corva.configuration import SETTINGS
 from corva.testing import TestClient
 from corva.validate_app_init import read_manifest
 
+from .utils.patch_fakeredis import info  # noqa
 
-@pytest.fixture(scope='function', autouse=True)
-def clean_real_redis():
+
+@pytest.fixture(scope="function")
+def redis_adapter(redis_client: Redis) -> Iterable[cache_adapter.RedisRepository]:
+    redis_adapter = cache_adapter.RedisRepository(
+        hash_name="test_hash_name", client=redis_client
+    )
+    yield redis_adapter
+
+
+@pytest.fixture(scope="function")
+def redis_client():
     redis_client = Redis.from_url(url=SETTINGS.CACHE_URL, decode_responses=True)
 
     redis_client.flushall()
@@ -20,24 +31,25 @@ def clean_real_redis():
     redis_client.flushall()
 
 
-@pytest.fixture(scope='function', autouse=True)
-def clean_fake_redis():
-    redis_client = cast(FakeRedis, FakeRedis.from_url(url=SETTINGS.CACHE_URL))
-
-    redis_client.flushall()
-
+@pytest.fixture(scope="function", autouse=True)
+def clean_redis_clients():
+    redis_clients = (
+        Redis.from_url(url=SETTINGS.CACHE_URL),
+        cast(FakeRedis, FakeRedis.from_url(url=SETTINGS.CACHE_URL)),
+    )
+    [client.flushall() for client in redis_clients]
     yield
 
-    redis_client.flushall()
+    [client.flushall() for client in redis_clients]
 
 
-@pytest.fixture(scope='function', autouse=True)
+@pytest.fixture(scope="function", autouse=True)
 def clean_read_manifest_lru_cache():
     read_manifest.cache_clear()
     yield
     read_manifest.cache_clear()
 
 
-@pytest.fixture(scope='function')
+@pytest.fixture(scope="function")
 def context():
     return TestClient._context