implementing fair queuing

Author: sanderegg

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

@@ -14,7 +14,6 @@
     field_validator,
 )
 from redis.commands.core import AsyncScript
-from servicelib.redis._utils import handle_redis_returns_union_types
 
 from ._client import RedisClientSDK
 from ._constants import (
@@ -37,6 +36,7 @@
     SCRIPT_BAD_EXIT_CODE,
     SCRIPT_OK_EXIT_CODE,
 )
+from ._utils import handle_redis_returns_union_types
 
 _logger = logging.getLogger(__name__)
 
@@ -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_KEY_PREFIX}{self.key}:holders:{self.instance_id}"
+        return f"{SEMAPHORE_HOLDER_KEY_PREFIX}{self.key}:{self.instance_id}"
 
     @computed_field
     @property
@@ -200,11 +200,11 @@ async def acquire(self) -> bool:
         # Execute the Lua scripts atomically
         cls = type(self)
         assert cls.register_semaphore is not None  # nosec
-        await cls.register_semaphore(
+        await cls.register_semaphore(  # pylint: disable=not-callable
             keys=[self.tokens_key, self.holders_key],
             args=[self.capacity, ttl_seconds],
             client=self.redis_client.redis,
-        )  # pylint: disable=not-callable
+        )
 
         try:
             # this is blocking pop with timeout
@@ -219,9 +219,11 @@ async def acquire(self) -> bool:
                 self.key,
                 self.instance_id,
             )
-            raise SemaphoreAcquisitionError(
-                name=self.key, capacity=self.capacity
-            ) from e
+            if self.blocking:
+                raise SemaphoreAcquisitionError(
+                    name=self.key, capacity=self.capacity
+                ) from e
+            return False
 
         assert len(tokens_key_token) == 2  # nosec
         assert tokens_key_token[0] == self.tokens_key  # nosec

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

@@ -290,43 +290,6 @@ async def _raising_context():
     assert await captured_semaphore.get_current_count() == 0
 
 
-async def test_semaphore_ttl_cleanup(
-    redis_client_sdk: RedisClientSDK,
-    semaphore_name: str,
-    semaphore_capacity: int,
-    short_ttl: datetime.timedelta,
-):
-    # Create semaphore with explicit short TTL
-    semaphore = DistributedSemaphore(
-        redis_client=redis_client_sdk,
-        key=semaphore_name,
-        capacity=semaphore_capacity,
-        ttl=short_ttl,
-    )
-
-    # Manually add an expired entry
-    expired_instance_id = "expired-instance"
-    current_time = asyncio.get_event_loop().time()
-    # Make sure it's definitely expired by using the short TTL
-    expired_time = current_time - short_ttl.total_seconds() - 1
-
-    await redis_client_sdk.redis.zadd(
-        semaphore.semaphore_key, {expired_instance_id: expired_time}
-    )
-
-    # Verify the entry was added
-    initial_count = await redis_client_sdk.redis.zcard(semaphore.semaphore_key)
-    assert initial_count == 1
-
-    # Current count should clean up expired entries
-    count = await semaphore.get_current_count()
-    assert count == 0
-
-    # Verify expired entry was removed
-    remaining = await redis_client_sdk.redis.zcard(semaphore.semaphore_key)
-    assert remaining == 0
-
-
 async def test_multiple_semaphores_different_keys(
     redis_client_sdk: RedisClientSDK,
     faker: Faker,