diff --git a/docs/topics/signals.rst b/docs/topics/signals.rst
index 5407141db..405b131ed 100644
--- a/docs/topics/signals.rst
+++ b/docs/topics/signals.rst
@@ -200,6 +200,23 @@ request_scheduled
     :param spider: the spider that yielded the request
     :type spider: :class:`~scrapy.spider.Spider` object
 
+request_dropped
+-----------------
+
+.. signal:: request_dropped
+.. function:: request_dropped(request, spider)
+
+    Sent when a :class:`~scrapy.http.Request`, scheduled by the engine to be
+    downloaded later, is rejected by the scheduler.
+
+    The signal does not support returning deferreds from their handlers.
+
+    :param request: the request that reached the scheduler
+    :type request: :class:`~scrapy.http.Request` object
+
+    :param spider: the spider that yielded the request
+    :type spider: :class:`~scrapy.spider.Spider` object
+
 response_received
 -----------------
 
diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py
index 4ef1d0fc6..b009898a3 100644
--- a/scrapy/core/engine.py
+++ b/scrapy/core/engine.py
@@ -173,7 +173,9 @@ class ExecutionEngine(object):
     def schedule(self, request, spider):
         self.signals.send_catch_log(signal=signals.request_scheduled,
                 request=request, spider=spider)
-        return self.slot.scheduler.enqueue_request(request)
+        if not self.slot.scheduler.enqueue_request(request):
+            self.signals.send_catch_log(signal=signals.request_dropped,
+                                        request=request, spider=spider)
 
     def download(self, request, spider):
         slot = self.slot
diff --git a/scrapy/core/scheduler.py b/scrapy/core/scheduler.py
index ba2ca5a03..232bc6a40 100644
--- a/scrapy/core/scheduler.py
+++ b/scrapy/core/scheduler.py
@@ -47,7 +47,7 @@ class Scheduler(object):
     def enqueue_request(self, request):
         if not request.dont_filter and self.df.request_seen(request):
             self.df.log(request, self.spider)
-            return
+            return False
         dqok = self._dqpush(request)
         if dqok:
             self.stats.inc_value('scheduler/enqueued/disk', spider=self.spider)
@@ -55,6 +55,7 @@ class Scheduler(object):
             self._mqpush(request)
             self.stats.inc_value('scheduler/enqueued/memory', spider=self.spider)
         self.stats.inc_value('scheduler/enqueued', spider=self.spider)
+        return True
 
     def next_request(self):
         request = self.mqs.pop()
diff --git a/scrapy/signals.py b/scrapy/signals.py
index 11bbae945..de0886fb6 100644
--- a/scrapy/signals.py
+++ b/scrapy/signals.py
@@ -12,6 +12,7 @@ spider_idle = object()
 spider_closed = object()
 spider_error = object()
 request_scheduled = object()
+request_dropped = object()
 response_received = object()
 response_downloaded = object()
 item_scraped = object()
diff --git a/tests/test_engine.py b/tests/test_engine.py
index 67fb8ae79..6ed700caa 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -79,6 +79,7 @@ class CrawlerRun(object):
         self.spider = None
         self.respplug = []
         self.reqplug = []
+        self.reqdropped = []
         self.itemresp = []
         self.signals_catched = {}
 
@@ -95,6 +96,7 @@ class CrawlerRun(object):
         self.crawler = get_crawler(TestSpider)
         self.crawler.signals.connect(self.item_scraped, signals.item_scraped)
         self.crawler.signals.connect(self.request_scheduled, signals.request_scheduled)
+        self.crawler.signals.connect(self.request_dropped, signals.request_dropped)
         self.crawler.signals.connect(self.response_downloaded, signals.response_downloaded)
         self.crawler.crawl(start_urls=start_urls)
         self.spider = self.crawler.spider
@@ -123,6 +125,9 @@ class CrawlerRun(object):
     def request_scheduled(self, request, spider):
         self.reqplug.append((request, spider))
 
+    def request_dropped(self, request, spider):
+        self.reqdropped.append((request, spider))
+
     def response_downloaded(self, response, spider):
         self.respplug.append((response, spider))
 
@@ -161,6 +166,11 @@ class EngineTest(unittest.TestCase):
         urls_requested = set([rq[0].url for rq in self.run.reqplug])
         urls_expected = set([self.run.geturl(p) for p in paths_expected])
         assert urls_expected <= urls_requested
+        scheduled_requests_count = len(self.run.reqplug)
+        dropped_requests_count = len(self.run.reqdropped)
+        responses_count = len(self.run.respplug)
+        self.assertEqual(scheduled_requests_count,
+                         dropped_requests_count + responses_count)
 
     def _assert_downloaded_responses(self):
         # response tests