diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py
index 0df9ad2b2..6c6d24d5c 100644
--- a/scrapy/core/engine.py
+++ b/scrapy/core/engine.py
@@ -7,12 +7,13 @@ For more information see docs/topics/architecture.rst
 
 from __future__ import annotations
 
+import asyncio
 import logging
 from time import time
 from traceback import format_exc
 from typing import TYPE_CHECKING, Any, TypeVar, cast
 
-from twisted.internet.defer import Deferred, inlineCallbacks, succeed
+from twisted.internet.defer import CancelledError, Deferred, inlineCallbacks, succeed
 from twisted.internet.task import LoopingCall
 from twisted.python.failure import Failure
 
@@ -102,6 +103,8 @@ class ExecutionEngine:
         )
         self.start_time: float | None = None
         self._start: AsyncIterator[Any] | None = None
+        self._closewait: Deferred[None] | None = None
+        self._start_request_processing_dfd: Deferred[None] | None = None
         downloader_cls: type[Downloader] = load_object(self.settings["DOWNLOADER"])
         try:
             self.scheduler_cls: type[BaseScheduler] = self._get_scheduler_class(
@@ -133,9 +136,9 @@ class ExecutionEngine:
             self.signals.send_catch_log_deferred(signal=signals.engine_started)
         )
         self.running = True
-        self._closewait: Deferred[None] = Deferred()
+        self._closewait = Deferred()
         if _start_request_processing:
-            self._start_request_processing()
+            self._start_request_processing_dfd = self._start_request_processing()
         await maybe_deferred_to_future(self._closewait)
 
     def stop(self) -> Deferred[None]:
@@ -146,12 +149,16 @@ class ExecutionEngine:
             await maybe_deferred_to_future(
                 self.signals.send_catch_log_deferred(signal=signals.engine_stopped)
             )
-            self._closewait.callback(None)
+            if self._closewait:
+                self._closewait.callback(None)
 
         if not self.running:
             raise RuntimeError("Engine not running")
 
         self.running = False
+        if self._start_request_processing_dfd is not None:
+            self._start_request_processing_dfd.cancel()
+            self._start_request_processing_dfd = None
         dfd = (
             self.close_spider(self.spider, reason="shutdown")
             if self.spider is not None
@@ -213,17 +220,30 @@ class ExecutionEngine:
         # Starts the processing of scheduled requests, as well as a periodic
         # call to that processing method for scenarios where the scheduler
         # reports having pending requests but returns none.
-        assert self._slot is not None  # typing
-        self._slot.nextcall.schedule()
-        self._slot.heartbeat.start(self._SLOT_HEARTBEAT_INTERVAL)
+        try:
+            assert self._slot is not None  # typing
+            self._slot.nextcall.schedule()
+            self._slot.heartbeat.start(self._SLOT_HEARTBEAT_INTERVAL)
 
-        while self._start and self.spider:
-            await self._process_start_next()
-            if not self.needs_backout():
-                # Give room for the outcome of self._process_start_next() to be
-                # processed before continuing with the next iteration.
-                self._slot.nextcall.schedule()
-                await self._slot.nextcall.wait()
+            while self._start and self.spider:
+                await self._process_start_next()
+                if not self.needs_backout():
+                    # Give room for the outcome of self._process_start_next() to be
+                    # processed before continuing with the next iteration.
+                    self._slot.nextcall.schedule()
+                    await self._slot.nextcall.wait()
+        except (asyncio.exceptions.CancelledError, CancelledError):
+            # self.stop() has cancelled us, nothing to do
+            return
+        except Exception:
+            # an error happened, log it and stop the engine
+            self._start_request_processing_dfd = None
+            logger.error(
+                "Error while processing requests from start()",
+                exc_info=True,
+                extra={"spider": self.spider},
+            )
+            await maybe_deferred_to_future(self.stop())
 
     def _start_scheduled_requests(self) -> None:
         if self._slot is None or self._slot.closing is not None or self.paused:
diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py
index ac720e03f..a2fe281e0 100644
--- a/scrapy/core/scraper.py
+++ b/scrapy/core/scraper.py
@@ -139,7 +139,6 @@ class Scraper:
 
     def _check_if_closing(self) -> None:
         assert self.slot is not None  # typing
-        assert self.crawler.spider
         if self.slot.closing and self.slot.is_idle():
             assert self.crawler.spider
             self.slot.closing.callback(self.crawler.spider)
diff --git a/tests/test_engine.py b/tests/test_engine.py
index b2e436425..a3c5243d5 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -25,6 +25,7 @@ import attr
 import pytest
 from itemadapter import ItemAdapter
 from pydispatch import dispatcher
+from testfixtures import LogCapture
 from twisted.internet import defer, reactor
 from twisted.trial import unittest
 from twisted.web import server, static, util
@@ -448,7 +449,7 @@ class TestEngine(TestEngineBase):
     @defer.inlineCallbacks
     def test_start_already_running_exception(self):
         e = ExecutionEngine(get_crawler(MySpider), lambda _: None)
-        yield e.open_spider(MySpider(), [])
+        yield e.open_spider(MySpider())
         e.start()
 
         def cb(exc: BaseException) -> None:
@@ -459,6 +460,26 @@ class TestEngine(TestEngineBase):
         finally:
             yield e.stop()
 
+    @defer.inlineCallbacks
+    def test_start_request_processing_exception(self):
+        class BadRequestFingerprinter:
+            def fingerprint(self, request):
+                raise ValueError  # to make Scheduler.enqueue_request() fail
+
+        class SimpleSpider(Spider):
+            name = "simple"
+
+            async def start(self):
+                yield Request("data:,")
+
+        crawler = get_crawler(
+            SimpleSpider, {"REQUEST_FINGERPRINTER_CLASS": BadRequestFingerprinter}
+        )
+        with LogCapture() as log:
+            yield crawler.crawl()
+        assert "Error while processing requests from start()" in str(log)
+        assert "Spider closed (shutdown)" in str(log)
+
     def test_short_timeout(self):
         args = (
             sys.executable,