feat(DEVC-1286): init changes

- upgrade redis version for test server
- bump bunch of related third party dependencies:
  - fakeredis[lua] -> removed prefix LUA
  - redis == 6.2.0 latest supported with Python3.9+
  - types-redis~=4.6.0
- removed:
  - all LUA bindings from RedisRepository
  - use_lua_52
  - vacuum(...) and dependant codes
  - InternalCacheSdkProtocol
  - InternalRedisSdk
  - FakeInternalCacheSdk

Author: kossman

Makefile

@@ -112,7 +112,7 @@ up-cache:
 	-d \
 	--name python-sdk-redis \
 	-p 6379:6379 \
-	redis:6.0.9  # apps use 6.0.9 or 6.2.3
+	redis:7.4.3		# The same as on PROD, at least at the 15.09.2025 moment
 
 # down-cache: Stop Redis.
 .PHONY: down-cache

pyproject.toml

@@ -14,12 +14,12 @@ authors = [
 keywords = ["corva", "sdk"]
 requires-python = ">=3.9,<4.0"
 dependencies = [
-  "fakeredis[lua] >=2.26.2, <2.30.0",
+  "fakeredis >=2.30.0, <2.32.0",
   "pydantic >= 2.0, <3.0",
   "pydantic-settings >=2.0, <3.0",
-  "redis >=5.2.1, <6.0.0",
+  "redis == 6.2.0",
   "requests >=2.32.3, <3.0.0",
-  "urllib3==2.5.0"
+  "urllib3 == 2.5.0"
 ]
 classifiers = [
   "Development Status :: 5 - Production/Stable",
@@ -40,7 +40,7 @@ dev = [
   "ruff==0.12.11",
   "mypy>=1.10,<2",
   "types-freezegun~=1.1.9",
-  "types-redis~=4.2.4",
+  "types-redis~=4.6.0",
   "types-requests~=2.27.27",
   "coverage==7.6.1",
   "freezegun==1.5.1",

src/corva/cache_adapter.py

@@ -1,4 +1,3 @@
-import itertools
 from datetime import timedelta
 from typing import (
     Dict,
@@ -8,7 +7,6 @@
     Sequence,
     Tuple,
     Union,
-    cast,
     overload,
 )
 
@@ -47,206 +45,45 @@ def delete_all(self) -> None:
 
 
 class RedisRepository:
-    # Deletes M expired keys from the hash.
-    #
-    # Complexity: O(log(N) + M) + O(M) + O(M * log(N)) where N is the size of hash
-    # and M is the number of elements deleted. For constant M (e.g., 3) the complexity
-    # is O(log(N)).
-    #
-    # KEYS:
-    #      hash_name.
-    #      zset_name.
-    #
-    # ARGV:
-    #      delete_count.
-    #
-    # Returns: nil.
-    LUA_VACUUM_SCRIPT = """
-    local hash_name = KEYS[1]
-    local zset_name = KEYS[2]
-    local delete_count = tonumber(ARGV[1])
-    local time = redis.call('TIME')
-    local pnow = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
-
-    local keys_to_delete = redis.call(
-            'ZRANGEBYSCORE', zset_name, '-inf', pnow, 'LIMIT', 0, delete_count
-    )
-
-    if not next(keys_to_delete) then
-        return
-    end
-
-    redis.call('HDEL', hash_name, unpack(keys_to_delete))
-    redis.call('ZREM', zset_name, unpack(keys_to_delete))
-    """
-
-    # Gets either all or only requested keys and values from hash.
-    #
-    # Gets all keys if no keys specified in ARGV.
-    #
-    # Complexity: O(N) where N is the number of requested keys.
-    #
-    # KEYS:
-    #     hash_name.
-    #     zset_name.
-    #
-    # Optional ARGV:
-    #     key1.
-    #     key2.
-    #     ...
-    #
-    # Returns: list of names and values for non-expired keys.
-    LUA_GET_SCRIPT = """
-    local hash_name = KEYS[1]
-    local zset_name = KEYS[2]
-    local time = redis.call('TIME')
-    local pnow = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
-
-    local keys = ARGV
-    if next(ARGV) == nil then
-        keys = redis.call('HKEYS', hash_name)
-    end
-
-    if not next(keys) then
-        return {}
-    end
-
-    local hash = redis.call('HMGET', hash_name, unpack(keys))
-
-    local result = {}
-
-    for i, key in ipairs(keys) do
-        local pexpireat = redis.call('ZSCORE', zset_name, key)
-
-        if not pexpireat or pnow < tonumber(pexpireat) then
-            table.insert(result, key)
-            table.insert(result, hash[i])
-        end
-    end
-
-    return result
-    """
-
-    # Inserts list of keys-value-expiration tuples into the hash.
-    #
-    # Complexity: O(N * log(M)), where N is the number of inserted elements and
-    # M is the hash size.
-    #
-    # 1. If hash does not exist, it will automatically create one.
-    # 2. If the field already exists, its value and ttl will be overwritten.
-    # 3. Hash and zset ttl is always set to the biggest field's ttl.
-    # 4. When the field expires it may be deleted by:
-    #    - Manually invoking `vacuum` script.
-    #    - Redis automatically deleting expired hash (see note #3).
-    #
-    # KEYS:
-    #     hash_name.
-    #     zset_name.
-    #
-    # ARGV:
-    #     key.
-    #     value.
-    #     ttl.
-    #     ...
-    #
-    # Returns: nil.
-    LUA_SET_SCRIPT = """
-    local hash_name = KEYS[1]
-    local zset_name = KEYS[2]
-    local time = redis.call('TIME')
-
-    for i, _ in ipairs(ARGV) do
-        if i % 3 == 1 then
-            local key = ARGV[i]
-            local value = ARGV[i + 1]
-            local ttl = ARGV[i + 2]
-            local pexpireat = (
-                (tonumber(time[1]) + ttl) * 1000 + math.floor(tonumber(time[2]) / 1000)
-            )
-
-            redis.call('HSET', hash_name, key, value)
-            redis.call('ZADD', zset_name, pexpireat, key)
-
-        end
-    end
-
-    local max_pexpireat = tonumber(redis.call(
-            'ZREVRANGEBYSCORE', zset_name, '+inf', '-inf', 'WITHSCORES', 'LIMIT', 0, 1
-    )[2])
-
-    redis.call('PEXPIREAT', hash_name, max_pexpireat)
-    redis.call('PEXPIREAT', zset_name, max_pexpireat)
-    """
-
-    # Deletes all data from hash and zset.
-    #
-    # Complexity: O(N) where N is the hash size.
-    #
-    # KEYS:
-    #     hash_name.
-    #     zset_name.
-    #
-    # Returns: nil.
-    LUA_DELETE_ALL_SCRIPT = """
-    local hash_name = KEYS[1]
-    local zset_name = KEYS[2]
-
-    redis.call('DEL', hash_name, zset_name)
-    """
 
-    def __init__(self, hash_name: str, client: redis.Redis, use_lua_52: bool = False):
+    def __init__(self, hash_name: str, client: redis.Redis):
         self.hash_name = hash_name
-        self.zset_name = f'{hash_name}.EXPIREAT'
         self.client = client
-        self.lua_set_many = self.client.register_script(self.LUA_SET_SCRIPT)
-
-        lua_get_script = self.LUA_GET_SCRIPT
-        if use_lua_52:
-            # Hack for tests to work with fakeredis, as it uses Lua version > 5.1.
-            # In Lua 5.1, unpack was a global, but in 5.2 it's been moved to
-            # table.unpack.
-            lua_get_script = self.LUA_GET_SCRIPT.replace('unpack', 'table.unpack')
-        self.lua_get = self.client.register_script(lua_get_script)
-
-        self.lua_vacuum = self.client.register_script(self.LUA_VACUUM_SCRIPT)
-        self.lua_delete_all = self.client.register_script(self.LUA_DELETE_ALL_SCRIPT)
 
     def set(self, key: str, value: str, ttl: int) -> None:
         self.set_many(data=[(key, value, ttl)])
 
     def set_many(self, data: Sequence[Tuple[str, str, int]]) -> None:
-        self.lua_set_many(
-            keys=[self.hash_name, self.zset_name],
-            args=list(itertools.chain.from_iterable(data)),
-        )
+        pipe = self.client.pipeline()
+        for key, value, ttl in data:
+            pipe.hset(self.hash_name, key, value)
+            pipe.execute_command("HEXPIRE", self.hash_name, ttl, "FIELDS", 1, key)
+        pipe.execute()
 
     def get(self, key: str) -> Optional[str]:
-        return self.get_many(keys=[key]).get(key)
+        val = self.client.hget(self.hash_name, key)
+        return None if val is None else str(val)
 
     def get_many(self, keys: Sequence[str]) -> Dict[str, Optional[str]]:
-        data = self.lua_get(keys=[self.hash_name, self.zset_name], args=list(keys))
-
-        data = dict(zip(data[::2], data[1::2]))
-
-        for missing_key in set(keys) - set(data):
-            data[missing_key] = None
-
-        return data
+        if not keys:
+            return {}
+        values = self.client.hmget(self.hash_name, keys)
+        # redis-py returns a list of values where non-existent/expired are None
+        return {k: (None if v is None else str(v)) for k, v in zip(keys, values)}
 
     def get_all(self) -> Dict[str, str]:
-        return cast(Dict[str, str], self.get_many(keys=[]))
+        raw = self.client.hgetall(self.hash_name)
+        return dict(raw)
 
     def delete(self, key: str) -> None:
-        self.delete_many(keys=[key])
+        self.client.hdel(self.hash_name, key)
 
     def delete_many(self, keys: Sequence[str]) -> None:
-        self.set_many(data=[(key, '', -1) for key in keys])
+        if keys:
+            self.client.hdel(self.hash_name, *keys)
 
     def delete_all(self) -> None:
-        self.lua_delete_all(keys=[self.hash_name, self.zset_name])
-
-    def vacuum(self, delete_count: int) -> None:
-        self.lua_vacuum(keys=[self.hash_name, self.zset_name], args=[delete_count])
+        self.client.delete(self.hash_name)
 
 
 class DeprecatedRedisAdapter:

src/corva/handlers.py

@@ -36,7 +36,7 @@
 from corva.models.task import RawTaskEvent, TaskEvent, TaskStatus
 from corva.service import service
 from corva.service.api_sdk import CachingApiSdk, CorvaApiSdk
-from corva.service.cache_sdk import FakeInternalCacheSdk, InternalRedisSdk, UserRedisSdk
+from corva.service.cache_sdk import UserRedisSdk
 from corva.validate_app_init import validate_app_type_context
 
 StreamEventT = TypeVar("StreamEventT", bound=StreamEvent)
@@ -187,9 +187,6 @@ def wrapper(
         user_cache_sdk = UserRedisSdk(
             hash_name=hash_name, redis_dsn=SETTINGS.CACHE_URL, redis_client=redis_client
         )
-        internal_cache_sdk = InternalRedisSdk(
-            hash_name=hash_name, redis_client=redis_client
-        )
 
         records = event.filter_records(
             old_max_record_value=event.get_cached_max_record_value(cache=user_cache_sdk)
@@ -222,7 +219,6 @@ def wrapper(
                     api_sdk=CorvaApiSdk(api_adapter=api),
                     ttl=SETTINGS.SECRETS_CACHE_TTL,
                 ),
-                cache_sdk=internal_cache_sdk,
                 app=functools.partial(
                     cast(Callable[[StreamEvent, Api, UserRedisSdk], Any], func),
                     app_event,
@@ -295,9 +291,6 @@ def wrapper(
         user_cache_sdk = UserRedisSdk(
             hash_name=hash_name, redis_dsn=SETTINGS.CACHE_URL, redis_client=redis_client
         )
-        internal_cache_sdk = InternalRedisSdk(
-            hash_name=hash_name, redis_client=redis_client
-        )
 
         if isinstance(event, RawScheduledDataTimeEvent) and event.merge_metadata:
             event = event.rebuild_with_modified_times(
@@ -327,7 +320,6 @@ def wrapper(
                         api_sdk=CorvaApiSdk(api_adapter=api),
                         ttl=SETTINGS.SECRETS_CACHE_TTL,
                     ),
-                    cache_sdk=internal_cache_sdk,
                     app=functools.partial(
                         cast(Callable[[ScheduledEvent, Api, UserRedisSdk], Any], func),
                         cast(ScheduledEvent, app_event),
@@ -380,7 +372,7 @@ def wrapper(
         api_key: str,
         aws_request_id: str,
         logging_ctx: LoggingContext,
-        redis_client: redis.Redis,
+        redis_client: Optional[redis.Redis] = None, # noqa, for safe reasons
     ) -> Any:
         status = TaskStatus.fail
         data: Dict[str, Union[dict, str]] = {"payload": {}}
@@ -418,7 +410,6 @@ def wrapper(
                         api_sdk=CorvaApiSdk(api_adapter=api),
                         ttl=SETTINGS.SECRETS_CACHE_TTL,
                     ),
-                    cache_sdk=FakeInternalCacheSdk(),
                     app=functools.partial(
                         cast(Callable[[TaskEvent, Api], Any], func), app_event, api
                     ),
@@ -505,14 +496,6 @@ def wrapper(
             app_key=SETTINGS.APP_KEY,
             app_connection_id=event.data.rerun_app_connection_id,
         )
-        internal_cache_hash_name = get_cache_key(
-            provider=SETTINGS.PROVIDER,
-            asset_id=event.data.rerun_asset_id,
-            app_stream_id=event.data.rerun_app_stream_id,
-            app_key=SETTINGS.APP_KEY,
-            app_connection_id=event.data.app_connection_id,
-        )
-
         asset_cache = UserRedisSdk(
             hash_name=asset_cache_hash_name,
             redis_dsn=SETTINGS.CACHE_URL,
@@ -547,9 +530,6 @@ def wrapper(
                     api_sdk=CorvaApiSdk(api_adapter=api),
                     ttl=SETTINGS.SECRETS_CACHE_TTL,
                 ),
-                cache_sdk=InternalRedisSdk(
-                    hash_name=internal_cache_hash_name, redis_client=redis_client
-                ),
                 app=functools.partial(
                     cast(
                         Callable[