Implement a simple scheduler_empty signal

docs/topics/signals.rst

@@ -131,6 +131,17 @@ engine_stopped
 
     This signal supports returning deferreds from its handlers.
 
+scheduler_empty
+~~~~~~~~~~~~~~~
+
+.. signal:: scheduler_empty
+.. function:: scheduler_empty()
+
+    Sent whenever the engine asks for a pending request from the
+    :ref:`scheduler <topics-scheduler>` (i.e. calls its
+    :meth:`~scrapy.core.scheduler.BaseScheduler.next_request` method) and the
+    scheduler returns none.
+
 Item signals
 ------------
 

scrapy/core/engine.py

@@ -229,6 +229,7 @@ class ExecutionEngine:
 
         request = self._slot.scheduler.next_request()
         if request is None:
+            self.signals.send_catch_log(signals.scheduler_empty)
             return None
 
         d: Deferred[Response | Request] = self._download(request)

scrapy/signals.py

@@ -7,6 +7,7 @@ signals here without documenting them there.
 
 engine_started = object()
 engine_stopped = object()
+scheduler_empty = object()
 spider_opened = object()
 spider_idle = object()
 spider_closed = object()

tests/test_signals.py

@@ -1,8 +1,9 @@
 import pytest
 from twisted.internet import defer
-from twisted.trial import unittest
+from twisted.trial.unittest import TestCase
 
 from scrapy import Request, Spider, signals
+from scrapy.utils.defer import deferred_f_from_coro_f, maybe_deferred_to_future
 from scrapy.utils.test import get_crawler, get_from_asyncio_queue
 from tests.mockserver import MockServer
 
@@ -20,7 +21,21 @@ class ItemSpider(Spider):
         return {"index": response.meta["index"]}
 
 
-class TestAsyncSignal(unittest.TestCase):
+class MainTestCase(TestCase):
+    @deferred_f_from_coro_f
+    async def test_scheduler_empty(self):
+        crawler = get_crawler()
+        calls = []
+
+        def track_call():
+            calls.append(object())
+
+        crawler.signals.connect(track_call, signals.scheduler_empty)
+        await maybe_deferred_to_future(crawler.crawl())
+        assert len(calls) >= 1
+
+
+class MockServerTestCase(TestCase):
     @classmethod
     def setUpClass(cls):
         cls.mockserver = MockServer()