@@ -394,6 +394,10 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
             logger.warning(f'Marking request {request.unique_key} as handled that is not in progress.')
             return None
 
+        # Update the request's handled_at timestamp.
+        if request.handled_at is None:
+            request.handled_at = datetime.now(timezone.utc)
+
         async with self._get_pipeline() as pipe:
             if self._dedup_strategy == 'default':
                 await await_redis_response(pipe.sadd(self._handled_set_key, request.unique_key))
@@ -402,6 +406,7 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
                 await await_redis_response(pipe.bf().add(self._handled_filter_key, request.unique_key))
 
             await await_redis_response(pipe.hdel(self._in_progress_key, request.unique_key))
+            await await_redis_response(pipe.hset(self._data_key, request.unique_key, request.model_dump_json()))
 
             await self._update_metadata(
                 pipe,
@@ -444,6 +449,7 @@ async def reclaim_request(
                         f'{{"client_id":"{self.client_key}","blocked_until_timestamp":{blocked_until_timestamp}}}',
                     )
                 )
+                await await_redis_response(pipe.hset(self._data_key, request.unique_key, request.model_dump_json()))
                 self._pending_fetch_cache.appendleft(request)
             else:
                 await await_redis_response(pipe.rpush(self._queue_key, request.unique_key))