ongoing

Author: sanderegg

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

@@ -150,7 +150,7 @@ def holders_key(self) -> str:
     @property
     def holder_key(self) -> str:
         """Redis key for this instance's holder entry."""
-        return f"{SEMAPHORE_HOLDER_KEY_PREFIX}{self.key}:{self.instance_id}"
+        return f"{SEMAPHORE_KEY_PREFIX}{self.key}:holders:{self.instance_id}"
 
     @computed_field
     @property
@@ -191,8 +191,10 @@ async def acquire(self) -> bool:
         ttl_seconds = int(self.ttl.total_seconds())
         blocking_timeout_seconds = 1
         if self.blocking:
-            blocking_timeout_seconds = int(
-                self.blocking_timeout.total_seconds() if self.blocking_timeout else 0
+            blocking_timeout_seconds = (
+                int(self.blocking_timeout.total_seconds())
+                if self.blocking_timeout
+                else 60
             )
 
         # Execute the Lua scripts atomically
@@ -217,11 +219,9 @@ async def acquire(self) -> bool:
                 self.key,
                 self.instance_id,
             )
-            if self.blocking:
-                raise SemaphoreAcquisitionError(
-                    name=self.key, capacity=self.capacity
-                ) from e
-            return False
+            raise SemaphoreAcquisitionError(
+                name=self.key, capacity=self.capacity
+            ) from e
 
         assert len(tokens_key_token) == 2  # nosec
         assert tokens_key_token[0] == self.tokens_key  # nosec
@@ -231,7 +231,7 @@ async def acquire(self) -> bool:
         result = await cls.acquire_script(  # pylint: disable=not-callable
             keys=[self.holders_key, self.holder_key],
             args=[
-                token[0],
+                token,
                 self.instance_id,
                 ttl_seconds,
             ],

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

@@ -276,10 +276,10 @@ async def limited_function() -> None:
         key=semaphore_name,
         capacity=1,
         blocking=False,
-        blocking_timeout=datetime.timedelta(seconds=0.1),
+        blocking_timeout=None,
     )
     async def limited_function_non_blocking() -> None:
-        await asyncio.sleep(0.5)
+        await asyncio.sleep(2)
 
     tasks = [asyncio.create_task(limited_function_non_blocking()) for _ in range(3)]
     results = await asyncio.gather(*tasks, return_exceptions=True)
@@ -424,6 +424,7 @@ async def limited_function() -> None:
         assert "longer than expected" in caplog.messages[-1]
 
 
+@pytest.mark.skip
 async def test_semaphore_fair_queuing(
     redis_client_sdk: RedisClientSDK,
     semaphore_name: str,
@@ -498,6 +499,7 @@ async def test_context_manager_capacity_enforcement(
         redis_client_sdk,
         key=semaphore_name,
         capacity=2,
+        blocking_timeout=None,
     )
     @asynccontextmanager
     async def limited_context_manager():