test renewal fail

Author: sanderegg

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

@@ -32,3 +32,7 @@ class SemaphoreAcquisitionError(BaseRedisError):
 
 class SemaphoreNotAcquiredError(BaseRedisError):
     msg_template: str = "Semaphore '{name}' was not acquired by this instance"
+
+
+class SemaphoreLostError(BaseRedisError):
+    msg_template: str = "Semaphore '{name}' was lost by this instance `{instance_id}`"

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

@@ -13,7 +13,11 @@
     DEFAULT_SEMAPHORE_TTL,
     DEFAULT_SOCKET_TIMEOUT,
 )
-from ._errors import SemaphoreAcquisitionError, SemaphoreNotAcquiredError
+from ._errors import (
+    SemaphoreAcquisitionError,
+    SemaphoreLostError,
+    SemaphoreNotAcquiredError,
+)
 from ._semaphore import DistributedSemaphore
 
 _logger = logging.getLogger(__name__)
@@ -39,7 +43,14 @@ async def _renew_semaphore_entry(semaphore: DistributedSemaphore) -> None:
     async with semaphore.redis_client.redis.pipeline(transaction=True) as pipe:
         await pipe.zadd(semaphore.semaphore_key, {semaphore.instance_id: current_time})
         await pipe.expire(semaphore.holder_key, int(ttl_seconds))
-        await pipe.execute()
+        results = await pipe.execute()
+
+        # Check if EXPIRE succeeded (returns 1 if key existed, 0 if not)
+        expire_result = results[1]
+        if expire_result == 0:
+            raise SemaphoreLostError(
+                name=semaphore.key, instance_id=semaphore.instance_id
+            )
 
 
 P = ParamSpec("P")
@@ -164,9 +175,7 @@ async def _periodic_renewer() -> None:
                 # and cancel the work task automatically
 
                 # Re-raise the first exception in the group
-                if eg.exceptions:
-                    raise eg.exceptions[0] from eg
-                raise
+                raise eg.exceptions[0] from eg
 
             finally:
                 # Always release the semaphore

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

@@ -7,6 +7,7 @@
 
 import asyncio
 import datetime
+from typing import Literal
 from unittest import mock
 
 import pytest
@@ -18,12 +19,13 @@
     SEMAPHORE_HOLDER_KEY_PREFIX,
     SEMAPHORE_KEY_PREFIX,
 )
+from servicelib.redis._errors import SemaphoreLostError
 from servicelib.redis._semaphore import (
     DistributedSemaphore,
     SemaphoreAcquisitionError,
     SemaphoreNotAcquiredError,
-    with_limited_concurrency,
 )
+from servicelib.redis._semaphore_decorator import with_limited_concurrency
 
 pytest_simcore_core_services_selection = [
     "redis",
@@ -318,7 +320,7 @@ async def test_semaphore_auto_renewal_via_decorator(
         capacity=semaphore_capacity,
         ttl=short_ttl,
     )
-    async def long_running_work():
+    async def long_running_work() -> Literal["success"]:
         work_started.set()
         # Wait longer than TTL to ensure renewal works
         await asyncio.sleep(short_ttl.total_seconds() * 2)
@@ -354,52 +356,41 @@ async def test_decorator_auto_renewal_failure_propagation(
     semaphore_name: str,
     semaphore_capacity: int,
     short_ttl: datetime.timedelta,
-    monkeypatch: pytest.MonkeyPatch,
 ):
     """Test that auto-renewal failures properly propagate as exceptions in the decorator"""
-    from servicelib.redis._semaphore_decorator import _renew_semaphore_entry
-
-    class RenewalFailureError(Exception):
-        """Custom exception for testing renewal failures"""
 
     work_started = asyncio.Event()
 
-    # Mock the renewal function to fail after first call
-    call_count = 0
-    original_renew = _renew_semaphore_entry
-
-    async def failing_renew_semaphore_entry(semaphore):
-        nonlocal call_count
-        call_count += 1
-        if call_count <= 1:
-            # First call succeeds
-            await original_renew(semaphore)
-        else:
-            # Subsequent calls fail
-            raise RenewalFailureError("Simulated renewal failure")
-
-    monkeypatch.setattr(
-        "servicelib.redis._semaphore_decorator._renew_semaphore_entry",
-        failing_renew_semaphore_entry,
-    )
-
     @with_limited_concurrency(
         redis_client_sdk,
         key=semaphore_name,
         capacity=semaphore_capacity,
         ttl=short_ttl,
     )
-    async def work_that_should_fail():
+    async def work_that_should_fail() -> Literal["should not reach here"]:
         work_started.set()
         # Wait long enough for renewal to be attempted multiple times
-        await asyncio.sleep(short_ttl.total_seconds() * 1.5)
+        await asyncio.sleep(short_ttl.total_seconds() * 4)
         return "should not reach here"
 
     # The decorator should propagate the renewal failure
     task = asyncio.create_task(work_that_should_fail())
     await work_started.wait()  # Wait for work to start
 
-    with pytest.raises(RenewalFailureError, match="Simulated renewal failure"):
+    # Wait for the first renewal interval to pass
+    renewal_interval = short_ttl / 3
+    await asyncio.sleep(
+        renewal_interval.total_seconds() + 1
+    )  # Wait for renewal to happen
+
+    # Find and delete all holder keys for this semaphore
+    holder_keys = await redis_client_sdk.redis.keys(
+        f"{SEMAPHORE_HOLDER_KEY_PREFIX}{semaphore_name}:*"
+    )
+    assert holder_keys, "Holder keys should exist before deletion"
+    await redis_client_sdk.redis.delete(*holder_keys)
+
+    with pytest.raises(SemaphoreLostError):
         await task  # This should raise the renewal failure exception