Wait for the processing of start items

scrapy/core/engine.py

@@ -291,9 +291,7 @@ class ExecutionEngine:
                 self.crawl(item_or_request)
             else:
                 assert self._slot is not None
-                _schedule_coro(
-                    self.scraper.start_itemproc_async(item_or_request, response=None)
-                )
+                self.scraper._start_itemproc_nowait(item_or_request)
                 self._slot.nextcall.schedule()
 
     async def _start_request_processing(self) -> None:

scrapy/core/scraper.py

@@ -65,7 +65,7 @@ class Slot:
         self.queue: deque[QueueTuple] = deque()
         self.active: set[Request] = set()
         self.active_size: int = 0
-        self.itemproc_size: int = 0  # just for scrapy.utils.engine.get_engine_status()
+        self.itemproc_size: int = 0
         self.closing: Deferred[Spider] | None = None
 
     def add_response_request(
@@ -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
@@ -477,6 +477,23 @@ class Scraper:
         )
         return deferred_from_coro(self.start_itemproc_async(item, response=response))
 
+    def _start_itemproc_nowait(self, item: Any) -> None:
+        """Send *item* to the item pipelines without waiting for the outcome.
+
+        The scraper slot counts the item as being processed straight away, so
+        that the spider is not considered idle before the item pipelines have
+        had a chance to run.
+        """
+        assert self.slot is not None  # typing
+        self.slot.itemproc_size += 1
+        _schedule_coro(self._start_itemproc_nowait_async(item))
+
+    async def _start_itemproc_nowait_async(self, item: Any) -> None:
+        assert self.slot is not None  # typing
+        # start_itemproc_async() takes over the count of this item.
+        self.slot.itemproc_size -= 1
+        await self.start_itemproc_async(item, response=None)
+
     async def start_itemproc_async(
         self, item: Any, *, response: Response | Failure | None
     ) -> None:

tests/test_spider_start.py

@@ -23,7 +23,10 @@ ITEM_B = {"id": "b"}
 
 class TestMain:
     async def _test_spider(
-        self, spider: type[Spider], expected_items: list[Any] | None = None
+        self,
+        spider: type[Spider],
+        expected_items: list[Any] | None = None,
+        settings: dict[str, Any] | None = None,
     ) -> None:
         actual_items = []
         expected_items = [] if expected_items is None else expected_items
@@ -31,7 +34,7 @@ class TestMain:
         def track_item(item, response, spider):
             actual_items.append(item)
 
-        crawler = get_crawler(spider)
+        crawler = get_crawler(spider, settings)
         crawler.signals.connect(track_item, signals.item_scraped)
         await crawler.crawl_async()
         assert crawler.stats
@@ -98,3 +101,24 @@ class TestMain:
             yield ITEM_A
 
         await self._test_start(start, [ITEM_A])
+
+    @pytest.mark.requires_reactor  # needs a reactor for twisted_sleep()
+    @coroutine_test
+    async def test_slow_pipeline(self):
+        class SlowPipeline:
+            async def process_item(self, item):
+                await maybe_deferred_to_future(twisted_sleep(SLEEP_SECONDS))
+                return item
+
+        class TestSpider(Spider):
+            name = "test"
+
+            async def start(self):
+                yield ITEM_A
+                yield ITEM_B
+
+        await self._test_spider(
+            TestSpider,
+            [ITEM_A, ITEM_B],
+            {"ITEM_PIPELINES": {SlowPipeline: 0}},
+        )