feat(DEVC-1286): add migration logic with tests

Author: kossman

pyproject.toml

@@ -19,7 +19,8 @@ dependencies = [
   "pydantic-settings >=2.0, <3.0",
   "redis == 6.2.0",
   "requests >=2.32.3, <3.0.0",
-  "urllib3 == 2.5.0"
+  "urllib3 == 2.5.0",
+  "semver==3.0.4"
 ]
 classifiers = [
   "Development Status :: 5 - Production/Stable",

src/corva/cache_adapter.py

@@ -7,6 +7,7 @@
 )
 
 import redis
+import semver
 
 
 class CacheRepositoryProtocol(Protocol):
@@ -81,3 +82,70 @@ def delete_many(self, keys: Sequence[str]) -> None:
     def delete_all(self) -> None:
         self.client.delete(self.hash_name)
 
+
+class HashMigrator:
+    MINIMUM_ALLOWED_REDIS_SERVER = semver.Version(major=7, minor=4, patch=0)
+
+    def __init__(self, hash_name: str, client: redis.Redis):
+        self.hash_name = hash_name
+        self.zset_name = f'{hash_name}.EXPIREAT'
+        self.client = client
+
+    def run(self) -> bool:
+        """Migrate from old Lua+ZSET per-field TTL to Redis built-in per-field TTL.
+
+        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).
+          - 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.
+
+        Returns True if migration was attempted on a supported server (i.e., legacy
+        ZSET existed); otherwise False.
+        """
+        # Legacy structure must exist; otherwise nothing to do
+        if not self.client.exists(self.zset_name):
+            return False
+
+        # 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
+
+        server_version = semver.Version.parse(version=redis_version_str)
+
+        if server_version < self.MINIMUM_ALLOWED_REDIS_SERVER:
+            return False
+
+        from corva import Logger
+
+        # Current server time in ms
+        sec, micro = self.client.time()
+        now_ms = int(sec) * 1000 + int(micro) // 1000
+
+        # Queue all operations in a single pipeline and execute once
+        pipe = self.client.pipeline()
+
+        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
+                )
+
+        # 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}'")
+        return True

src/corva/service/cache_sdk.py

@@ -13,6 +13,7 @@
 import redis
 
 from corva import cache_adapter
+from corva.cache_adapter import HashMigrator
 
 
 class UserCacheSdkProtocol(Protocol):
@@ -74,6 +75,12 @@ def __init__(
             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)
 

tests/unit/conftest.py tests/conftest.pytests/unit/conftest.py renamed to tests/conftest.py

@@ -11,11 +11,11 @@
 
 @pytest.fixture(scope='function', autouse=True)
 def clean_real_redis():
-    redis_client = Redis.from_url(url=SETTINGS.CACHE_URL)
+    redis_client = Redis.from_url(url=SETTINGS.CACHE_URL, decode_responses=True)
 
     redis_client.flushall()
 
-    yield
+    yield redis_client
 
     redis_client.flushall()
 

tests/integration/test_cache_migration.py

@@ -0,0 +1,124 @@
+import time
+from typing import Dict
+
+import pytest
+
+from corva.cache_adapter import HashMigrator
+
+
+def _now_ms(redis_client) -> int:
+    sec, micro = redis_client.time()
+    return int(sec) * 1000 + int(micro) // 1000
+
+
+def _zset_name(hash_name: str) -> str:
+    return f"{hash_name}.EXPIREAT"
+
+
+@pytest.mark.integration
+def test_migrate_returns_false_when_no_legacy_zset_present(clean_real_redis):
+    hash_name = "test:migration:nozset:" + str(time.time_ns())
+    zset = _zset_name(hash_name)
+
+    clean_real_redis.delete(hash_name, zset)
+    clean_real_redis.hset(hash_name, mapping={"a": "1"})
+
+    migrator = HashMigrator(hash_name=hash_name, client=clean_real_redis)
+
+    # No legacy zset exists -> migration should be a no-op and return False
+    assert clean_real_redis.exists(zset) == 0
+    migrated = migrator.run()
+    assert not migrated
+    assert clean_real_redis.hget(hash_name, "a") == "1"
+
+
+@pytest.mark.integration
+def test_migrate_converts_fields_and_cleans_legacy_structures(clean_real_redis):
+    hash_name = "test:migration:convert:" + str(time.time_ns())
+    zset = _zset_name(hash_name)
+
+    clean_real_redis.delete(hash_name, zset)
+
+    # Prepopulate legacy hash + per-field expirations in zset (absolute ms deadlines)
+    clean_real_redis.hset(hash_name, mapping={"f1": "v1", "f2": "v2"})
+    now = _now_ms(clean_real_redis)
+    future_deadline = now + 2000  # 2 seconds in the future
+    past_deadline = now - 100     # already expired
+
+    clean_real_redis.zadd(zset, mapping={"f1": future_deadline, "f2": past_deadline})
+    # Simulate old behavior of setting key-level expiry as backstop (optional)
+    clean_real_redis.pexpireat(hash_name, future_deadline)
+    clean_real_redis.pexpireat(zset, future_deadline)
+
+    migrator = HashMigrator(hash_name=hash_name, client=clean_real_redis)
+
+    migrated = migrator.run()
+    assert migrated
+
+    # zset removed and key-level TTL removed (persisted)
+    assert clean_real_redis.exists(zset) == 0
+    assert clean_real_redis.ttl(hash_name) == -1
+
+    # f1 remains and has a per-field TTL
+    assert clean_real_redis.hget(hash_name, "f1") == "v1"
+    f1_hpttl = clean_real_redis.execute_command("HPTTL", hash_name, "FIELDS", 1, "f1")
+    if isinstance(f1_hpttl, list):
+        f1_hpttl = f1_hpttl[0]
+    assert isinstance(f1_hpttl, int) and f1_hpttl > 0
+
+    # f2 was expired -> should be deleted from hash
+    assert clean_real_redis.hget(hash_name, "f2") is None
+
+
+@pytest.mark.integration
+def test_migrate_is_idempotent(clean_real_redis):
+    hash_name = "test:migration:idempotent:" + str(time.time_ns())
+    zset = _zset_name(hash_name)
+
+    clean_real_redis.delete(hash_name, zset)
+
+    clean_real_redis.hset(hash_name, mapping={"k": "v"})
+    now = _now_ms(clean_real_redis)
+    clean_real_redis.zadd(zset, mapping={"k": now + 5000})
+
+    migrator = HashMigrator(hash_name, clean_real_redis)
+    migrated = migrator.run()
+    assert migrated
+    # Run again – zset is gone so it should be a no-op returning False
+    migrated = migrator.run()
+    assert not migrated
+
+
+@pytest.mark.integration
+def test_migrate_large_batch_processes_all_fields(clean_real_redis):
+    hash_name = "test:migration:large:" + str(time.time_ns())
+    zset = _zset_name(hash_name)
+
+    clean_real_redis.delete(hash_name, zset)
+
+    # Create 300 fields to exceed default batch=256
+    mapping: Dict[str, str] = {f"k{i}": f"v{i}" for i in range(300)}
+    clean_real_redis.hset(hash_name, mapping=mapping)
+
+    now = _now_ms(clean_real_redis)
+    future = now + 60_000
+    zmap = {f"k{i}": future for i in range(300)}
+    clean_real_redis.zadd(zset, mapping=zmap)
+
+    migrator = HashMigrator(hash_name, clean_real_redis)
+
+    assert migrator.run() is True
+
+    # All fields should still be present
+    assert clean_real_redis.hlen(hash_name) == 300
+
+    # zset removed, hash persisted
+    assert clean_real_redis.exists(zset) == 0
+    assert clean_real_redis.ttl(hash_name) == -1
+
+    # Sample a few fields to ensure per-field TTLs were applied
+    for key in ["k0", "k128", "k256"]:
+        hpttl = clean_real_redis.execute_command("HPTTL", hash_name, "FIELDS", 1, key)
+        if isinstance(hpttl, list):
+            hpttl = hpttl[0]
+        assert isinstance(hpttl, int) and hpttl > 0