Replace RuntimeError handling with Andrey’s lock-based approach

scrapy/core/engine.py

@@ -113,6 +113,7 @@ class ExecutionEngine:
         self.start_time: float | None = None
         self._load_seeding_policy()
         self._seeds: AsyncIterable[Any] | None = None
+        self._waiting_for_seed: bool = False
 
     def _load_seeding_policy(self) -> None:
         try:
@@ -192,15 +193,13 @@ class ExecutionEngine:
 
     @inlineCallbacks
     def _process_next_seed(self):
+        if self._waiting_for_seed:
+            return
+        self._waiting_for_seed = True
         try:
             seed = yield deferred_from_coro(self._seeds.__anext__())
         except StopAsyncIteration:
             self._seeds = None
-        except RuntimeError:
-            # “RuntimeError: anext(): asynchronous generator is already
-            # running” happens if yield_seeds is taking long to yield the
-            # next seed.
-            pass
         except Exception:
             self._seeds = None
             logger.error(
@@ -214,6 +213,8 @@ class ExecutionEngine:
             else:
                 self.scraper.start_itemproc(seed, response=None)
                 self._slot.nextcall.schedule()
+        finally:
+            self._waiting_for_seed = False
 
     @inlineCallbacks
     def _start_next_requests(self) -> Generator[Deferred[Any], Any, None]: