Handle a request with idempotency support. 1. If cached (success only) -> return cached response 2. If in-flight -> wait for result 3. Otherwise -> execute and cache if successful
(
self,
request_id: str,
execute_fn: Callable[[], Awaitable[dict[str, Any]]],
)
| 77 | logger.debug(f"Cleaned up {len(expired_keys)} expired cache entries") |
| 78 | |
| 79 | async def handle_request( |
| 80 | self, |
| 81 | request_id: str, |
| 82 | execute_fn: Callable[[], Awaitable[dict[str, Any]]], |
| 83 | ) -> dict[str, Any]: |
| 84 | """ |
| 85 | Handle a request with idempotency support. |
| 86 | |
| 87 | 1. If cached (success only) -> return cached response |
| 88 | 2. If in-flight -> wait for result |
| 89 | 3. Otherwise -> execute and cache if successful |
| 90 | """ |
| 91 | # Check cache / in-flight status under a single lock acquisition |
| 92 | event: asyncio.Event | None = None |
| 93 | async with self._lock: |
| 94 | if request_id in self._cache: |
| 95 | entry = self._cache[request_id] |
| 96 | if not entry.is_expired(self._ttl_seconds): |
| 97 | logger.debug(f"Cache hit for request {request_id}") |
| 98 | return entry.response |
| 99 | |
| 100 | # Check if already in-flight |
| 101 | event = self._pending.get(request_id) |
| 102 | if event is not None: |
| 103 | logger.debug(f"Request {request_id} is in-flight, waiting...") |
| 104 | else: |
| 105 | # Mark as in-flight while still holding the lock |
| 106 | self._pending[request_id] = asyncio.Event() |
| 107 | |
| 108 | # If in-flight, wait outside the lock |
| 109 | if event is not None: |
| 110 | await event.wait() |
| 111 | async with self._lock: |
| 112 | if request_id in self._pending_results: |
| 113 | return self._pending_results.pop(request_id) |
| 114 | # Fallback: check cache |
| 115 | if request_id in self._cache: |
| 116 | return self._cache[request_id].response |
| 117 | raise RuntimeError(f"Request {request_id} completed but result not found") |
| 118 | |
| 119 | try: |
| 120 | # Execute the request |
| 121 | response = await execute_fn() |
| 122 | |
| 123 | # Only cache successful responses |
| 124 | async with self._lock: |
| 125 | if response.get("success", False): |
| 126 | self._cache[request_id] = CacheEntry(response=response) |
| 127 | logger.debug(f"Cached successful response for {request_id}") |
| 128 | else: |
| 129 | logger.debug( |
| 130 | f"Not caching error response for {request_id}: " |
| 131 | f"{response.get('error', {}).get('code', 'unknown')}" |
| 132 | ) |
| 133 | |
| 134 | # Store result for waiting requests |
| 135 | self._pending_results[request_id] = response |
| 136 |