Include the test from #7650 and fix an off-by-one issue caught by it

scrapy/extensions/feedexport.py

@@ -538,7 +538,7 @@ class FeedExporter:
                 saved_id = feed_batch_ids.get(slot.uri_template, 0)
                 if saved_id == 0:
                     continue
-                batch_id = saved_id + 1
+                batch_id = saved_id
                 uri_params = self._get_uri_params(
                     spider, self.feeds[slot.uri_template]["uri_params"]
                 )

tests/test_feedexport.py

@@ -1451,8 +1451,8 @@ class TestFeedExporterBatchIdState:
             exporter.crawler.spider = spider
             await exporter._on_state_loaded(spider.state)
             batch_ids = {slot.uri_template: slot.batch_id for slot in exporter.slots}
-            assert batch_ids[uri_a] == 4
-            assert batch_ids[uri_b] == 8
+            assert batch_ids[uri_a] == 3
+            assert batch_ids[uri_b] == 7
 
     @coroutine_test
     async def test_no_jobdir_no_error(self):
@@ -1481,8 +1481,8 @@ class TestFeedExporterBatchIdState:
         Uses the full Scrapy machinery (real crawl_async()) so that any internal
         change to signal ordering or extension wiring is caught.  SpiderState
         fires spider_state_loaded / spider_state_saving; FeedExporter uses those
-        to resume from the right batch ID, so the second run's files (feed-4.jl,
-        feed-5.jl) never collide with the first run's files (feed-1.jl, feed-2.jl).
+        to resume from the right batch ID, so the second run's files (feed-3.jl,
+        feed-4.jl) never collide with the first run's files (feed-1.jl, feed-2.jl).
         """
         with tempfile.TemporaryDirectory() as tmpdir:
             jobdir = Path(tmpdir) / "jobdir"
@@ -1514,8 +1514,8 @@ class TestFeedExporterBatchIdState:
             assert Path(f"{tmpdir}/feed-1.jl").read_bytes() == content_run1[1]
             assert Path(f"{tmpdir}/feed-2.jl").read_bytes() == content_run1[2]
             # Second run created new files with higher batch IDs
+            assert Path(f"{tmpdir}/feed-3.jl").exists()
             assert Path(f"{tmpdir}/feed-4.jl").exists()
-            assert Path(f"{tmpdir}/feed-5.jl").exists()
 
     @coroutine_test
     async def test_feed_slots_initialized_fires_after_state_loaded(self):
@@ -1539,7 +1539,7 @@ class TestFeedExporterBatchIdState:
             await exporter._on_state_loaded(spider.state)
 
             assert len(received) == 1
-            assert received[0][0].batch_id == 6
+            assert received[0][0].batch_id == 5
 
     @coroutine_test
     async def test_feed_slots_initialized_fires_from_engine_started_without_state(self):

tests/test_feedexport_batch.py

@@ -364,6 +364,61 @@ class TestBatchDeliveries(TestFeedExportBase):
         data = await self.exported_data(items, settings)
         assert len(items) == len(data["json"])
 
+    @coroutine_test
+    async def test_jobdir_batch_id_continues_after_restart(self):
+        """Regression test for #5153.
+
+        When JOBDIR is set and the feed URI references ``%(batch_id)``, the
+        batch_id counter must persist across restarts so that re-running the
+        crawl with the same JOBDIR does not overwrite previously-written
+        batch files.
+        """
+        items = [
+            self.MyItem({"foo": "bar1", "egg": "spam1"}),
+            self.MyItem({"foo": "bar2", "egg": "spam2"}),
+        ]
+        feed_dir = self._random_temp_filename()
+        jobdir = self._random_temp_filename()
+        uri_template = feed_dir / "%(batch_id)d.jl"
+
+        def make_settings():
+            return {
+                "FEEDS": {
+                    uri_template: {"format": "jl"},
+                },
+                "FEED_EXPORT_BATCH_ITEM_COUNT": 1,
+                "JOBDIR": str(jobdir),
+            }
+
+        # First run: should produce 1.jl and 2.jl (one per item).
+        await self.exported_data(items, make_settings())
+        first_run_files = sorted(p.name for p in feed_dir.iterdir())
+        assert first_run_files == ["1.jl", "2.jl"]
+        # Mark each file so that we can detect overwrites.
+        sentinel = b"SENTINEL-FROM-FIRST-RUN\n"
+        for name in first_run_files:
+            with (feed_dir / name).open("ab") as f:
+                f.write(sentinel)
+        first_run_contents = {
+            name: (feed_dir / name).read_bytes() for name in first_run_files
+        }
+
+        # Second run with the same JOBDIR: must NOT overwrite the prior files
+        # and must start the batch_id counter at 3.
+        more_items = [
+            self.MyItem({"foo": "bar3", "egg": "spam3"}),
+            self.MyItem({"foo": "bar4", "egg": "spam4"}),
+        ]
+        await self.exported_data(more_items, make_settings())
+
+        for name, prior in first_run_contents.items():
+            assert (feed_dir / name).read_bytes() == prior, (
+                f"{name} was overwritten by the second run"
+            )
+
+        all_files = sorted(p.name for p in feed_dir.iterdir())
+        assert all_files == ["1.jl", "2.jl", "3.jl", "4.jl"]
+
     @inline_callbacks_test
     def test_stats_batch_file_success(self):
         settings = {