Remove the job directory of a download slot once it drains (#7955)

scrapy/pqueues.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 import hashlib
 import logging
+from contextlib import suppress
+from pathlib import Path
 from typing import TYPE_CHECKING, Protocol, cast
 
 from scrapy.utils.misc import build_from_crawler
@@ -409,6 +411,11 @@ class DownloaderAwarePriorityQueue:
         request = queue.pop()
         if len(queue) == 0:
             del self.pqueues[slot]
+            if self.key:
+                # Reclaim the slot directory; rmdir leaves it alone if the
+                # downstream queues did not remove all their files.
+                with suppress(OSError):
+                    Path(self.key, _path_safe(slot)).rmdir()
         return request
 
     def push(self, request: Request) -> None:

tests/test_pqueues.py

@@ -6,7 +6,7 @@ import queuelib
 
 from scrapy.core.downloader import Downloader
 from scrapy.http.request import Request
-from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue
+from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue, _path_safe
 from scrapy.spiders import Spider
 from scrapy.squeues import FifoMemoryQueue, PickleFifoDiskQueue
 from scrapy.utils.misc import build_from_crawler, load_object
@@ -258,6 +258,29 @@ class TestDownloaderAwarePriorityQueue:
         assert "other-slot" not in self.queue
 
 
+def test_slot_directory_removed_when_slot_drains(tmp_path):
+    crawler = get_crawler(Spider)
+    crawler.spider = crawler._create_spider("foo")
+    crawler.engine = Mock(downloader=MockDownloader())
+    queue = DownloaderAwarePriorityQueue.from_crawler(
+        crawler=crawler,
+        downstream_queue_cls=PickleFifoDiskQueue,
+        key=str(tmp_path),
+    )
+    request = Request("https://example.org/1")
+    slot_dir = tmp_path / _path_safe("example.org")
+
+    queue.push(request)
+    assert slot_dir.is_dir()
+
+    assert queue.pop().url == request.url
+    assert not slot_dir.exists()
+
+    queue.push(request)
+    assert slot_dir.is_dir()
+    queue.close()
+
+
 @pytest.mark.parametrize(
     ("input_", "output"),
     [