Store full feed batches before spider closes (#7730)

Start closing and storing each batch as soon as it reaches the configured item count. Track unfinished close tasks so spider shutdown still waits for all deliveries before emitting the exporter-closed signal.

Add an end-to-end regression test that verifies the first batch is stored while the crawl is still running.

scrapy/extensions/feedexport.py

@@ -13,7 +13,7 @@ import re
 import sys
 import warnings
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Coroutine
+from collections.abc import Callable
 from datetime import datetime, timezone
 from pathlib import Path, PureWindowsPath
 from tempfile import NamedTemporaryFile
@@ -458,7 +458,7 @@ class FeedExporter:
         self.feeds = {}
         self.slots: list[FeedSlot] = []
         self.filters: dict[str, ItemFilter] = {}
-        self._pending_close_coros: list[Coroutine[Any, Any, None]] = []
+        self._pending_close_tasks: list[asyncio.Task[None] | Deferred[None]] = []
 
         if not self.settings["FEEDS"] and not self.settings["FEED_URI"]:
             raise NotConfigured
@@ -522,23 +522,44 @@ class FeedExporter:
             )
 
     async def close_spider(self, spider: Spider) -> None:
-        self._pending_close_coros.extend(
-            self._close_slot(slot, spider) for slot in self.slots
-        )
+        for slot in self.slots:
+            self._schedule_slot_close(slot, spider)
 
-        if self._pending_close_coros:
+        if self._pending_close_tasks:
             if is_asyncio_available():
                 await asyncio.wait(
-                    [asyncio.create_task(coro) for coro in self._pending_close_coros]
+                    cast("list[asyncio.Task[None]]", list(self._pending_close_tasks))
                 )
             else:
                 await DeferredList(
-                    deferred_from_coro(coro) for coro in self._pending_close_coros
+                    cast("list[Deferred[None]]", list(self._pending_close_tasks))
                 )
 
         # Send FEED_EXPORTER_CLOSED signal
         await self.crawler.signals.send_catch_log_async(signals.feed_exporter_closed)
 
+    def _schedule_slot_close(
+        self, slot: FeedSlot, spider: Spider
+    ) -> asyncio.Task[None] | Deferred[None]:
+        """Start closing the slot without waiting for it to finish, keeping
+        track of the pending work so that it can be awaited in
+        :meth:`close_spider` if it hasn't finished by then."""
+        aw: asyncio.Task[None] | Deferred[None]
+        coro = self._close_slot(slot, spider)
+        if is_asyncio_available():
+            aw = asyncio.create_task(coro)
+            self._pending_close_tasks.append(aw)
+            aw.add_done_callback(self._pending_close_tasks.remove)
+        else:
+            aw = deferred_from_coro(coro)
+            self._pending_close_tasks.append(aw)
+            aw.addBoth(self._untrack_pending_close_task, aw)
+        return aw
+
+    def _untrack_pending_close_task(self, result: Any, aw: Deferred[None]) -> Any:
+        self._pending_close_tasks.remove(aw)
+        return result
+
     @staticmethod
     def _get_file(slot_: FeedSlot) -> IO[bytes]:
         assert slot_.file
@@ -635,7 +656,7 @@ class FeedExporter:
                 uri_params = self._get_uri_params(
                     spider, self.feeds[slot.uri_template]["uri_params"], slot
                 )
-                self._pending_close_coros.append(self._close_slot(slot, spider))
+                self._schedule_slot_close(slot, spider)
                 slots.append(
                     self._start_new_batch(
                         batch_id=slot.batch_id + 1,

tests/test_feedexport_batch.py

@@ -210,6 +210,47 @@ class TestBatchDeliveries(TestFeedExportBase):
         header = self.MyItem.fields.keys()
         await self.assertExported(items, header, rows, settings=settings)
 
+    @coroutine_test
+    async def test_batch_delivered_when_full(self):
+        """Full batches must be finalized and delivered as soon as they are
+        full, instead of when the spider closes (#7730)."""
+        dir_path = self._random_temp_filename()
+        batch1_path = Path(dir_path, "1.json")
+        mockserver_url = self.mockserver.url("/")
+        batch1_contents: list[bytes | None] = []
+
+        class TestSpider(scrapy.Spider):
+            name = "testspider"
+            start_urls = [mockserver_url]
+
+            def parse(self, response):
+                yield {"foo": "bar1"}
+                yield {"foo": "bar2"}
+                yield scrapy.Request(
+                    mockserver_url, callback=self.parse2, dont_filter=True
+                )
+
+            def parse2(self, response):
+                # the first batch was full after the second item, so it must
+                # have been delivered by now
+                batch1_contents.append(
+                    batch1_path.read_bytes() if batch1_path.exists() else None
+                )
+                yield {"foo": "bar3"}
+
+        settings = {
+            "FEEDS": {
+                build_url(dir_path / "%(batch_id)d.json"): {"format": "json"},
+            },
+            "FEED_EXPORT_BATCH_ITEM_COUNT": 2,
+        }
+        crawler = get_crawler(TestSpider, settings)
+        await crawler.crawl_async()
+
+        assert batch1_contents, "the second request was not processed"
+        assert batch1_contents[0] is not None, "batch 1 was not stored during the crawl"
+        assert json.loads(batch1_contents[0]) == [{"foo": "bar1"}, {"foo": "bar2"}]
+
     def test_wrong_path(self):
         """If path is without %(batch_time)s and %(batch_id) an exception must be raised"""
         settings = {