wait for start item pipelines before closing

scrapy/core/scraper.py

@@ -93,7 +93,7 @@ class Slot:
             self.active_size -= self.MIN_RESPONSE_SIZE
 
     def is_idle(self) -> bool:
-        return not (self.queue or self.active)
+        return not (self.queue or self.active or self.itemproc_size)
 
     def needs_backout(self) -> bool:
         return self.active_size > self.max_active_size
@@ -540,3 +540,4 @@ class Scraper:
             )
         finally:
             self.slot.itemproc_size -= 1
+            self._check_if_closing()

tests/test_core_scraper.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import asyncio
+from typing import TYPE_CHECKING, Any, ClassVar
 
+from scrapy import Spider
+from scrapy.core.scraper import Scraper
 from scrapy.utils.test import get_crawler
 from tests.spiders import SimpleSpider
 from tests.utils.decorators import coroutine_test
@@ -25,3 +28,56 @@ async def test_scraper_exception(
     )
     await crawler.crawl_async(url=mockserver.url("/"))
     assert "Scraper bug processing" in caplog.text
+
+
+@coroutine_test
+async def test_close_spider_async_waits_for_item_processing() -> None:
+    process_started = asyncio.Event()
+    finish_processing = asyncio.Event()
+
+    class BlockingPipeline:
+        events: ClassVar[list[str]] = []
+
+        async def process_item(self, item: Any) -> Any:
+            self.events.append("process_item_start")
+            process_started.set()
+            await finish_processing.wait()
+            self.events.append("process_item_end")
+            return item
+
+        async def close_spider(self) -> None:
+            self.events.append("close_spider")
+
+    class TestSpider(Spider):
+        name = "test"
+
+    crawler = get_crawler(
+        TestSpider,
+        {
+            "ITEM_PIPELINES": {
+                BlockingPipeline: 0,
+            },
+        },
+    )
+    crawler.spider = crawler._create_spider()
+    scraper = Scraper(crawler)
+    await scraper.open_spider_async()
+
+    itemproc_task = asyncio.create_task(scraper.start_itemproc_async({}, response=None))
+    await process_started.wait()
+
+    close_task = asyncio.create_task(scraper.close_spider_async())
+    try:
+        await asyncio.sleep(0)
+        assert not close_task.done()
+        assert BlockingPipeline.events == ["process_item_start"]
+    finally:
+        finish_processing.set()
+        await itemproc_task
+        await close_task
+
+    assert BlockingPipeline.events == [
+        "process_item_start",
+        "process_item_end",
+        "close_spider",
+    ]