ensure we have auto expiry on keys

Author: sanderegg

packages/service-library/src/servicelib/redis/_semaphore.py

@@ -155,6 +155,18 @@ def holders_set(self) -> str:
         """Redis key for the holders SET."""
         return f"{SEMAPHORE_KEY_PREFIX}{self.key}:holders_set"
 
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def holders_set_ttl(self) -> datetime.timedelta:
+        """TTL for the holders SET"""
+        return self.ttl * 5
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def tokens_set_ttl(self) -> datetime.timedelta:
+        """TTL for the tokens SET"""
+        return self.ttl * 5
+
     @computed_field  # type: ignore[prop-decorator]
     @property
     def holder_key(self) -> str:
@@ -181,12 +193,11 @@ def validate_timeout(
 
     async def _ensure_semaphore_initialized(self) -> None:
         """Initializes the semaphore in Redis if not already done."""
-        ttl_seconds = self.ttl.total_seconds()
         cls = type(self)
         assert cls.register_semaphore is not None  # nosec
         await cls.register_semaphore(  # pylint: disable=not-callable
             keys=[self.tokens_key, self.holders_set],
-            args=[self.capacity, ttl_seconds],
+            args=[self.capacity, self.holders_set_ttl.total_seconds()],
             client=self.redis_client.redis,
         )
 
@@ -259,8 +270,6 @@ async def acquire(self) -> bool:
             )
             return True
 
-        ttl_seconds = self.ttl.total_seconds()
-
         if self.blocking is False:
             self._token = await self._non_blocking_acquire()
             if not self._token:
@@ -277,7 +286,8 @@ async def acquire(self) -> bool:
             args=[
                 self._token,
                 self.instance_id,
-                ttl_seconds,
+                self.ttl.total_seconds(),
+                self.holders_set_ttl.total_seconds(),
             ],
             client=self.redis_client.redis,
         )
@@ -361,8 +371,13 @@ async def reacquire(self) -> None:
         cls = type(self)
         assert cls.renew_script is not None  # nosec
         result = await cls.renew_script(  # pylint: disable=not-callable
-            keys=[self.holders_set, self.holder_key],
-            args=[self.instance_id, ttl_seconds],
+            keys=[self.holders_set, self.holder_key, self.tokens_key],
+            args=[
+                self.instance_id,
+                ttl_seconds,
+                self.holders_set_ttl.total_seconds(),
+                self.tokens_set_ttl.total_seconds(),
+            ],
             client=self.redis_client.redis,
         )
 

packages/service-library/src/servicelib/redis/lua/acquire_fair_semaphore_v2.lua

@@ -5,6 +5,7 @@
 -- ARGV[1]: token (the token received from BRPOP)
 -- ARGV[2]: instance_id (the instance trying to acquire the semaphore)
 -- ARGV[3]: ttl_seconds (for the holder_key)
+-- ARGV[4]: holders_set_ttl_seconds (to set expiry on holders set)
 --
 -- Returns: {exit_code, status, token, current_count}
 -- exit_code: 0 if acquired
@@ -16,6 +17,7 @@ local holder_key = KEYS[2]
 local token = ARGV[1]
 local instance_id = ARGV[2]
 local ttl_seconds = tonumber(ARGV[3])
+local holders_set_ttl_seconds = tonumber(ARGV[4])
 
 
 
@@ -24,7 +26,7 @@ redis.call('SADD', holders_key, instance_id)
 redis.call('SETEX', holder_key, ttl_seconds, token)
 
 -- Step 2: Set expiry on holders set to prevent infinite growth
-redis.call('EXPIRE', holders_key, ttl_seconds * 10)
+redis.call('EXPIRE', holders_key, holders_set_ttl_seconds)
 
 local current_count = redis.call('SCARD', holders_key)
 

packages/service-library/src/servicelib/redis/lua/register_semaphore_holder.lua

@@ -23,7 +23,7 @@ if tokens_exist == 0 and holders_exist == 0 then
         redis.call('LPUSH', tokens_key, 'token_' .. i)
     end
     -- Set expiry on tokens list to prevent infinite growth
-    -- redis.call('EXPIRE', tokens_key, ttl_seconds)
+    redis.call('EXPIRE', tokens_key, ttl_seconds)
 end
 
 return 0

packages/service-library/src/servicelib/redis/lua/renew_fair_semaphore_v2.lua

@@ -1,18 +1,24 @@
 -- Renew semaphore holder TTL (simplified for token pool design)
 -- KEYS[1]: holders_key (SET of current holders)
 -- KEYS[2]: holder_key (individual holder TTL key for this instance)
+-- KEYS[3]: tokens_key (LIST of available tokens)
 -- ARGV[1]: instance_id
 -- ARGV[2]: ttl_seconds
+-- ARGV[3]: holders_ttl_seconds (to renew holders set)
+-- ARGV[4]: tokens_ttl_seconds (to renew tokens list)
 --
 -- Returns: {exit_code, status, current_count}
 -- exit_code: 0 if renewed, 255 if failed
 -- status: 'renewed', 'not_held', or 'expired'
 
 local holders_key = KEYS[1]
 local holder_key = KEYS[2]
+local tokens_key = KEYS[3]
 
 local instance_id = ARGV[1]
 local ttl_seconds = tonumber(ARGV[2])
+local holders_ttl_seconds = tonumber(ARGV[3])
+local tokens_ttl_seconds = tonumber(ARGV[4])
 
 -- Step 1: Check if this instance is currently a holder
 local is_holder = redis.call('SISMEMBER', holders_key, instance_id)
@@ -32,4 +38,8 @@ end
 local token = redis.call('GET', holder_key)
 redis.call('SETEX', holder_key, ttl_seconds, token)
 
+-- Step 4: Renew the holders set and tokens list TTLs to prevent infinite growth
+redis.call('EXPIRE', holders_key, holders_ttl_seconds)
+redis.call('EXPIRE', tokens_key, tokens_ttl_seconds)
+
 return {0, 'renewed', redis.call('SCARD', holders_key)}

packages/service-library/tests/redis/test_semaphore.py

@@ -38,7 +38,7 @@
 def with_short_default_semaphore_ttl(
     mocker: MockerFixture,
 ) -> datetime.timedelta:
-    short_ttl = datetime.timedelta(seconds=2)
+    short_ttl = datetime.timedelta(seconds=5)
     mocker.patch(
         "servicelib.redis._semaphore.DEFAULT_SEMAPHORE_TTL",
         short_ttl,