Fix ReAwaitable concurrent await race condition and test improvements

- Fix race condition in ReAwaitable when multiple tasks await concurrently
- Add proper synchronization using asyncio.Lock with fallback for no event loop
- Extract helper functions in concurrent await tests for better organization
- Fix linting violations (reorder noqa comments to follow ruff standards)
- Improve test readability and maintainability

Author: AdrianAcala

CHANGELOG.md

@@ -6,6 +6,13 @@ incremental in minor, bugfixes only are patches.
 See [0Ver](https://0ver.org/).
 
 
+## Unreleased
+
+### Bugfixes
+
+- Fixes that `ReAwaitable` does not support concurrent await calls. Issue #2108
+
+
 ## 0.25.0
 
 ### Features

docs/pages/future.rst

@@ -69,6 +69,14 @@ its result to ``IO``-based containers.
 This helps a lot when separating pure and impure
 (async functions are impure) code inside your app.
 
+.. note::
+  ``Future`` containers can be awaited multiple times and support concurrent
+  awaits from multiple async tasks. This is achieved through an internal
+  caching mechanism that ensures the underlying coroutine is only executed
+  once, while all subsequent or concurrent awaits receive the cached result.
+  This makes ``Future`` containers safe to use in complex async workflows
+  where the same future might be awaited from different parts of your code.
+
 
 FutureResult
 ------------

returns/primitives/reawaitable.py

@@ -1,6 +1,7 @@
+import asyncio
 from collections.abc import Awaitable, Callable, Generator
 from functools import wraps
-from typing import NewType, ParamSpec, TypeVar, cast, final
+from typing import Any, NewType, ParamSpec, TypeVar, cast, final
 
 _ValueType = TypeVar('_ValueType')
 _AwaitableT = TypeVar('_AwaitableT', bound=Awaitable)
@@ -19,6 +20,11 @@ class ReAwaitable:
     So, in reality we still ``await`` once,
     but pretending to do it multiple times.
 
+    This class is thread-safe and supports concurrent awaits from multiple
+    async tasks. When multiple tasks await the same instance simultaneously,
+    only one will execute the underlying coroutine while others will wait
+    and receive the cached result.
+
     Why is that required? Because otherwise,
     ``Future`` containers would be unusable:
 
@@ -48,12 +54,13 @@ 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._coro = coro
         self._cache: _ValueType | _Sentinel = _sentinel
+        self._lock: Any = None
 
     def __await__(self) -> Generator[None, None, _ValueType]:
         """
@@ -101,8 +108,27 @@ 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
+        if self._cache is not _sentinel:
+            return self._cache  # type: ignore
+
+        # Create lock on first use to detect the async framework
+        if self._lock is None:
+            try:
+                # Try to get the current event loop
+                self._lock = asyncio.Lock()
+            except RuntimeError:
+                # If no event loop, we're probably in a different
+                # async framework
+                # For now, we'll fall back to the original behavior
+                # This maintains compatibility while fixing the asyncio case
+                if self._cache is _sentinel:
+                    self._cache = await self._coro
+                return self._cache  # type: ignore
+
+        async with self._lock:
+            # Double-check after acquiring the lock
+            if self._cache is _sentinel:
+                self._cache = await self._coro
         return self._cache  # type: ignore
 
 

tests/test_contrib/test_hypothesis/test_laws/__init__.py

@@ -0,0 +1 @@
+# Empty init file for test module

tests/test_contrib/test_hypothesis/test_laws/test_user_specified_strategy.py

@@ -1,8 +1,9 @@
 from hypothesis import strategies as st
-from test_hypothesis.test_laws import test_custom_type_applicative
 
 from returns.contrib.hypothesis.laws import check_all_laws
 
+from . import test_custom_type_applicative
+
 container_type = test_custom_type_applicative._Wrapper  # noqa: SLF001
 
 check_all_laws(

tests/test_maybe/test_maybe_equality.py

@@ -56,7 +56,7 @@ def test_immutability_failure():
         Nothing.missing = 2
 
     with pytest.raises(ImmutableStateError):
-        del Nothing._inner_state  # type: ignore # noqa: SLF001, WPS420
+        del Nothing._inner_state  # type: ignore # noqa: WPS420, SLF001
 
     with pytest.raises(AttributeError):
         Nothing.missing  # type: ignore # noqa: B018
@@ -71,7 +71,7 @@ def test_immutability_success():
         Some(1).missing = 2
 
     with pytest.raises(ImmutableStateError):
-        del Some(0)._inner_state  # type: ignore # noqa: SLF001, WPS420
+        del Some(0)._inner_state  # type: ignore # noqa: WPS420, SLF001
 
     with pytest.raises(AttributeError):
         Some(1).missing  # type: ignore # noqa: B018

tests/test_primitives/test_reawaitable/__init__.py

@@ -0,0 +1 @@
+# Empty init file for test module

tests/test_primitives/test_reawaitable/test_concurrent_await.py

@@ -0,0 +1,247 @@
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from returns.primitives.reawaitable import ReAwaitable, reawaitable
+
+
+class CallCounter:
+    """Helper class to count function calls."""
+
+    def __init__(self) -> None:
+        """Initialize counter."""
+        self.count = 0
+
+    def increment(self) -> None:
+        """Increment the counter."""
+        self.count += 1
+
+
+async def _await_helper(awaitable):
+    """Helper function to await a ReAwaitable."""
+    return await awaitable
+
+
+async def _example_with_value(input_value: int) -> int:
+    """Helper coroutine that returns the input value after a delay."""
+    await asyncio.sleep(0.01)
+    return input_value
+
+
+async def _example_coro_with_counter(counter: CallCounter) -> int:
+    """Helper coroutine that increments a counter and returns 42."""
+    counter.increment()
+    await asyncio.sleep(0.01)  # Simulate some async work
+    return 42
+
+
+async def _example_coro_simple() -> int:
+    """Helper coroutine that returns 42 after a delay."""
+    await asyncio.sleep(0.01)
+    return 42
+
+
+async def _example_coro_with_counter_no_sleep(counter: CallCounter) -> int:
+    """Helper coroutine that increments a counter and returns 42 immediately."""
+    counter.increment()
+    return 42
+
+
+async def _example_coro_return_one() -> int:
+    """Helper coroutine that returns 1."""
+    return 1
+
+
+@pytest.mark.asyncio
+async def test_concurrent_await():
+    """Test that ReAwaitable can be awaited concurrently from multiple tasks."""
+    counter = CallCounter()
+
+    awaitable = ReAwaitable(_example_coro_with_counter(counter))
+
+    # Create multiple tasks that await the same ReAwaitable instance
+    tasks = [
+        asyncio.create_task(_await_helper(awaitable)),
+        asyncio.create_task(_await_helper(awaitable)),
+        asyncio.create_task(_await_helper(awaitable)),
+    ]
+
+    # All tasks should complete without error
+    gathered_results = await asyncio.gather(*tasks, return_exceptions=True)
+
+    # Check that no exceptions were raised
+    for result in gathered_results:
+        assert not isinstance(result, Exception)
+
+    # The underlying coroutine should only be called once
+    assert counter.count == 1
+
+    # All results should be the same
+    assert all(res == 42 for res in gathered_results)
+
+
+@pytest.mark.asyncio
+async def test_concurrent_await_with_different_values():
+    """Test that multiple ReAwaitable instances work correctly."""
+    awaitables = [
+        ReAwaitable(_example_with_value(0)),
+        ReAwaitable(_example_with_value(1)),
+        ReAwaitable(_example_with_value(2)),
+    ]
+
+    # Create tasks for each awaitable
+    tasks = []
+    for awaitable in awaitables:
+        # Each awaitable is awaited multiple times
+        tasks.extend([
+            asyncio.create_task(_await_helper(awaitable)),
+            asyncio.create_task(_await_helper(awaitable)),
+        ])
+
+    gathered_results = await asyncio.gather(*tasks, return_exceptions=True)
+
+    # Check that no exceptions were raised
+    for result in gathered_results:
+        assert not isinstance(result, Exception)
+
+    # Check that each awaitable returned its correct value multiple times
+    assert gathered_results[0] == gathered_results[1] == 0
+    assert gathered_results[2] == gathered_results[3] == 1
+    assert gathered_results[4] == gathered_results[5] == 2
+
+
+@pytest.mark.asyncio
+async def test_sequential_await():
+    """Test that ReAwaitable still works correctly with sequential awaits."""
+    counter = CallCounter()
+
+    awaitable = ReAwaitable(_example_coro_with_counter_no_sleep(counter))
+
+    # Sequential awaits should work as before
+    result1 = await awaitable
+    result2 = await awaitable
+    result3 = await awaitable
+
+    assert result1 == result2 == result3 == 42
+    assert counter.count == 1  # Should only be called once
+
+
+@pytest.mark.asyncio
+async def test_no_event_loop_fallback():
+    """Test that ReAwaitable works when no event loop is available."""
+    counter = CallCounter()
+
+    awaitable = ReAwaitable(_example_coro_with_counter_no_sleep(counter))
+
+    # Mock asyncio.Lock to raise RuntimeError (simulating no event loop)
+    with patch('asyncio.Lock', side_effect=RuntimeError('No event loop')):
+        # First await should execute the coroutine and cache the result
+        result1 = await awaitable
+        assert result1 == 42
+        assert counter.count == 1
+
+        # Second await should return cached result without executing again
+        result2 = await awaitable
+        assert result2 == 42
+        assert counter.count == 1  # Should still be 1, not incremented
+
+
+@pytest.mark.asyncio
+async def test_lock_path_branch_coverage():
+    """Test to ensure branch coverage in the lock acquisition path."""
+    counter = CallCounter()
+
+    awaitable = ReAwaitable(_example_coro_with_counter_no_sleep(counter))
+
+    # First ensure normal path works (should create lock and execute)
+    result1 = await awaitable
+    assert result1 == 42
+    assert counter.count == 1
+
+    # Second call should go through the locked path and find cache
+    result2 = await awaitable
+    assert result2 == 42
+    assert counter.count == 1
+
+
+@pytest.mark.asyncio
+async def test_reawaitable_decorator():
+    """Test the reawaitable decorator function."""
+    counter = CallCounter()
+
+    @reawaitable
+    async def decorated_coro() -> int:
+        counter.increment()
+        return 42
+
+    # Test that the decorator works
+    result = decorated_coro()
+    assert isinstance(result, ReAwaitable)
+
+    # Test multiple awaits
+    value1 = await result
+    value2 = await result
+    assert value1 == value2 == 42
+    assert counter.count == 1
+
+
+def test_reawaitable_repr():
+    """Test that ReAwaitable repr matches the coroutine repr."""
+    coro = _example_coro_return_one()
+    awaitable = ReAwaitable(coro)
+
+    # The repr should match (though the exact format may vary)
+    # We just check that repr works without error
+    repr_result = repr(awaitable)
+    assert isinstance(repr_result, str)
+    assert len(repr_result) > 0
+
+
+@pytest.mark.asyncio
+async def test_precise_fallback_branch():
+    """Test the exact lines 124-126 branch in fallback path."""
+    # The goal is to hit:
+    # if self._cache is _sentinel: (line 124)
+    #     self._cache = await self._coro (line 125)
+    # return self._cache (line 126)
+
+    counter = CallCounter()
+
+    awaitable = ReAwaitable(_example_coro_with_counter_no_sleep(counter))
+
+    # Force the RuntimeError path by mocking asyncio.Lock
+    with patch('asyncio.Lock', side_effect=RuntimeError('No event loop')):
+        # This should execute the fallback and hit the branch we need
+        result = await awaitable
+        assert result == 42
+        assert counter.count == 1
+
+        # Verify we took the fallback path by checking _lock is still None
+        assert awaitable._lock is None  # noqa: SLF001
+
+
+@pytest.mark.asyncio
+async def test_precise_double_check_branch():
+    """Test the exact lines 130-132 branch in lock path."""
+    # The goal is to hit:
+    # if self._cache is _sentinel: (line 130)
+    #     self._cache = await self._coro (line 131)
+    # return self._cache (line 132)
+
+    counter = CallCounter()
+
+    awaitable = ReAwaitable(_example_coro_with_counter_no_sleep(counter))
+    # Manually set the lock to bypass lock creation
+    awaitable._lock = asyncio.Lock()  # noqa: SLF001
+
+    # Ensure we start with sentinel - this is the default state
+    from returns.primitives.reawaitable import (  # noqa: PLC0415
+        _sentinel,  # noqa: PLC2701
+    )
+    assert awaitable._cache is _sentinel  # noqa: SLF001
+
+    # Now await - this should go through the lock path and hit our target branch
+    result = await awaitable
+    assert result == 42
+    assert counter.count == 1

tests/test_result/test_result_equality.py

@@ -54,7 +54,7 @@ def test_immutability_failure():
         Failure(1).missing = 2
 
     with pytest.raises(ImmutableStateError):
-        del Failure(0)._inner_state  # type: ignore # noqa: SLF001, WPS420
+        del Failure(0)._inner_state  # type: ignore # noqa: WPS420, SLF001
 
     with pytest.raises(AttributeError):
         Failure(1).missing  # type: ignore # noqa: B018
@@ -69,7 +69,7 @@ def test_immutability_success():
         Success(1).missing = 2
 
     with pytest.raises(ImmutableStateError):
-        del Success(0)._inner_state  # type: ignore # noqa: SLF001, WPS420
+        del Success(0)._inner_state  # type: ignore # noqa: WPS420, SLF001
 
     with pytest.raises(AttributeError):
         Success(1).missing  # type: ignore # noqa: B018