refactor(scrapy): simplify httpcache close_spider cleanup and document cache boundary logs

Author: vdusek

src/apify/scrapy/extensions/_httpcache.py

@@ -85,44 +85,18 @@ def close_spider(self, _: Spider, current_time: int | None = None) -> None:
         if self._async_thread is None:
             raise ValueError('Async thread not initialized')
 
+        if current_time is None:
+            current_time = int(time())
+
         logger.info(f'Cleaning up cache items (max {self._expiration_max_items})')
-        # `close` always runs in the `finally`, so neither a cleanup failure below nor an early return can leak
-        # the event-loop thread.
+        # Best-effort: a cleanup failure is logged and swallowed (the sweep only reclaims storage, so failing it
+        # must not turn a normal spider close into an error), and `close` always runs in the `finally`, so
+        # neither the failure nor an early return can leak the event-loop thread.
         try:
             if self._expiration_secs > 0:
-                if current_time is None:
-                    current_time = int(time())
-
-                async def expire_kvs() -> None:
-                    if self._kvs is None:
-                        raise ValueError('Key value store not initialized')
-                    # Best-effort cleanup: at most `_expiration_max_items` keys per close, in no guaranteed order,
-                    # so stale entries may linger. This only reclaims storage; `retrieve_response` already treats
-                    # an expired entry as a cache miss.
-                    processed = 0
-                    async for item in self._kvs.iterate_keys():
-                        if processed >= self._expiration_max_items:
-                            break
-                        processed += 1
-                        value = await self._kvs.get_value(item.key)
-                        try:
-                            gzip_time = read_gzip_time(value)
-                        except Exception as exc:
-                            logger.warning(f'Malformed cache item {item.key}: {exc}')
-                            await self._kvs.delete_value(item.key)
-                        else:
-                            if self._expiration_secs < current_time - gzip_time:
-                                logger.debug(f'Expired cache item {item.key}')
-                                await self._kvs.delete_value(item.key)
-                            else:
-                                logger.debug(f'Valid cache item {item.key}')
-
-                # Best-effort: log and swallow a cleanup failure rather than raise. The sweep only reclaims
-                # storage, so failing it must not turn a normal spider close into an error.
-                try:
-                    self._async_thread.run_coro(expire_kvs())
-                except Exception:
-                    logger.exception('Failed to clean up expired cache items.')
+                self._async_thread.run_coro(self._expire_kvs(current_time))
+        except Exception:
+            logger.exception('Failed to clean up expired cache items.')
         finally:
             logger.debug('Closing cache storage')
             try:
@@ -147,6 +121,9 @@ def retrieve_response(self, _: Spider, request: Request, current_time: int | Non
         try:
             value = self._async_thread.run_coro(self._kvs.get_value(key))
         except Exception:
+            # Not redundant: `run_coro` re-raises but no longer logs, and Scrapy logs this downloader-middleware
+            # Deferred failure only generically ('Error downloading <request>'), or not at all if a request
+            # errback swallows it. So record the cache-specific context here before re-raising.
             logger.exception('Failed to retrieve a response from the cache.')
             raise
 
@@ -198,9 +175,38 @@ def store_response(self, _: Spider, request: Request, response: Response) -> Non
         try:
             self._async_thread.run_coro(self._kvs.set_value(key, value))
         except Exception:
+            # Boundary log, see `retrieve_response`: `run_coro` re-raises without logging and Scrapy logs this
+            # Deferred failure only generically (or not at all behind an errback), so capture the context here.
             logger.exception('Failed to store a response in the cache.')
             raise
 
+    async def _expire_kvs(self, current_time: int) -> None:
+        """Sweep the cache key-value store, deleting expired or unreadable entries.
+
+        Best-effort cleanup: at most `_expiration_max_items` keys per close, in no guaranteed order, so stale
+        entries may linger. This only reclaims storage; `retrieve_response` already treats an expired entry as
+        a cache miss.
+        """
+        if self._kvs is None:
+            raise ValueError('Key value store not initialized')
+        processed = 0
+        async for item in self._kvs.iterate_keys():
+            if processed >= self._expiration_max_items:
+                break
+            processed += 1
+            value = await self._kvs.get_value(item.key)
+            try:
+                gzip_time = read_gzip_time(value)
+            except Exception as exc:
+                logger.warning(f'Malformed cache item {item.key}: {exc}')
+                await self._kvs.delete_value(item.key)
+            else:
+                if self._expiration_secs < current_time - gzip_time:
+                    logger.debug(f'Expired cache item {item.key}')
+                    await self._kvs.delete_value(item.key)
+                else:
+                    logger.debug(f'Valid cache item {item.key}')
+
 
 def to_gzip(data: dict, mtime: int | None = None) -> bytes:
     """Dump a dictionary to a gzip-compressed JSON byte stream.