diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py
index fa708a2e1..a3b35b3ac 100644
--- a/scrapy/extensions/feedexport.py
+++ b/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,
diff --git a/tests/test_feedexport_batch.py b/tests/test_feedexport_batch.py
index 469985288..8ba09a090 100644
--- a/tests/test_feedexport_batch.py
+++ b/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 = {