Add lock to ReAwaitable for concurrent awaits

CHANGELOG.md

@@ -6,6 +6,13 @@ incremental in minor, bugfixes only are patches.
 See [0Ver](https://0ver.org/).
 
 
+## 0.25.1
+
+### Bugfixes
+
+- Adds lock to `ReAwaitable` to safely handle multiple concurrent awaits on the same instance
+
+
 ## 0.25.0
 
 ### Features

returns/primitives/reawaitable.py

@@ -2,6 +2,16 @@
 from functools import wraps
 from typing import NewType, ParamSpec, TypeVar, cast, final
 
+# Try to use anyio.Lock, fall back to asyncio.Lock
+try:
+    import anyio  # noqa: WPS433
+except ImportError:  # pragma: no cover
+    import asyncio  # noqa: WPS433
+
+    Lock = asyncio.Lock
+else:
+    Lock = anyio.Lock
+
 _ValueType = TypeVar('_ValueType')
 _AwaitableT = TypeVar('_AwaitableT', bound=Awaitable)
 _Ps = ParamSpec('_Ps')
@@ -48,10 +58,11 @@ class ReAwaitable:
 
     """
 
-    __slots__ = ('_cache', '_coro')
+    __slots__ = ('_cache', '_coro', '_lock')
 
     def __init__(self, coro: Awaitable[_ValueType]) -> None:
         """We need just an awaitable to work with."""
+        self._lock = Lock()
         self._coro = coro
         self._cache: _ValueType | _Sentinel = _sentinel
 
@@ -101,9 +112,10 @@ def __repr__(self) -> str:
 
     async def _awaitable(self) -> _ValueType:
         """Caches the once awaited value forever."""
-        if self._cache is _sentinel:
-            self._cache = await self._coro
-        return self._cache  # type: ignore
+        async with self._lock:
+            if self._cache is _sentinel:
+                self._cache = await self._coro
+            return self._cache  # type: ignore
 
 
 def reawaitable(

tests/test_primitives/test_reawaitable/test_reawaitable_concurrency.py

@@ -0,0 +1,67 @@
+import anyio
+import pytest
+
+from returns.primitives.reawaitable import ReAwaitable, reawaitable
+
+
+# Fix for issue with multiple awaits on the same ReAwaitable instance:
+# https://github.com/dry-python/returns/issues/2108
+async def sample_coro() -> str:
+    """Sample coroutine that simulates an async operation."""
+    await anyio.sleep(1)
+    return 'done'
+
+
+async def await_helper(awaitable_obj) -> str:
+    """Helper to await objects in tasks."""
+    return await awaitable_obj
+
+
+@pytest.mark.anyio
+async def test_concurrent_awaitable() -> None:
+    """Test that ReAwaitable safely handles concurrent awaits using a lock."""
+    test_target = ReAwaitable(sample_coro())
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(await_helper, test_target)
+        tg.start_soon(await_helper, test_target)
+
+
+@pytest.mark.anyio  # noqa: WPS210
+async def test_reawaitable_decorator() -> None:
+    """Test the reawaitable decorator with concurrent awaits."""
+
+    async def test_coro() -> str:  # noqa: WPS430
+        await anyio.sleep(1)
+        return 'decorated'
+
+    decorated = reawaitable(test_coro)
+    instance = decorated()
+
+    # Test multiple awaits
+    result1 = await instance
+    result2 = await instance
+
+    assert result1 == 'decorated'
+    assert result1 == result2
+
+    # Test concurrent awaits
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(await_helper, instance)
+        tg.start_soon(await_helper, instance)
+
+
+@pytest.mark.anyio
+async def test_reawaitable_repr() -> None:
+    """Test the __repr__ method of ReAwaitable."""
+
+    async def test_func() -> int:  # noqa: WPS430
+        return 1
+
+    coro = test_func()
+    target = ReAwaitable(coro)
+
+    # Test the representation
+    assert repr(target) == repr(coro)
+    # Ensure the coroutine is properly awaited
+    assert await target == 1