diff --git a/docs/03_guides/06_scrapy.mdx b/docs/03_guides/06_scrapy.mdx
index 49755de6..53d909e1 100644
--- a/docs/03_guides/06_scrapy.mdx
+++ b/docs/03_guides/06_scrapy.mdx
@@ -104,7 +104,7 @@ The following example shows a Scrapy Actor that scrapes page titles and enqueues
 
 ## Dealing with imminent migration to another host
 
-Under some circumstances, the platform may decide to [migrate your Actor](https://docs.apify.com/academy/expert-scraping-with-apify/migrations-maintaining-state) from one piece of infrastructure to another while it's in progress. While [Crawlee](https://crawlee.dev/python)-based projects can pause and resume their work after a restart, achieving the same with a Scrapy-based project can be challenging.
+Under some circumstances, the platform may decide to [migrate your Actor](https://docs.apify.com/academy/expert-scraping-with-apify/migrations-maintaining-state) from one piece of infrastructure to another while it's in progress. Requests that Scrapy hasn't finished when the run stops stay unhandled in the request queue, so the next run picks them up and downloads them from scratch. A Scrapy-based project doesn't resume where it left off the way a [Crawlee](https://crawlee.dev/python)-based one does, so items their callbacks already pushed can land in the dataset twice.
 
 As a workaround for this issue (tracked as [apify/actor-templates#303](https://github.com/apify/actor-templates/issues/303)), turn on caching with `HTTPCACHE_ENABLED` and set `HTTPCACHE_EXPIRATION_SECS` to at least a few minutes—the exact value depends on your use case. If your Actor gets migrated and restarted, the subsequent run will hit the cache, making it fast and avoiding unnecessary resource consumption.
 
diff --git a/src/apify/scrapy/_async_thread.py b/src/apify/scrapy/_async_thread.py
index 90f6f4cb..d35dec3a 100644
--- a/src/apify/scrapy/_async_thread.py
+++ b/src/apify/scrapy/_async_thread.py
@@ -12,6 +12,9 @@
 
 logger = getLogger(__name__)
 
+SUBMITTED_PRUNE_THRESHOLD = 128
+"""How many `submit_coro` futures may pile up before the finished ones are dropped from the tracking list."""
+
 
 class AsyncThread:
     """Run an asyncio event loop in a dedicated background thread.
@@ -26,6 +29,9 @@ def __init__(self, default_timeout: timedelta = timedelta(seconds=60)) -> None:
         self._default_timeout = default_timeout
         self._eventloop = asyncio.new_event_loop()
 
+        self._submitted: list[futures.Future] = []
+        """Futures of the coroutines submitted via `submit_coro` that may still be running."""
+
         # Start the event loop in a dedicated daemon thread.
         self._thread = threading.Thread(
             target=self._start_event_loop,
@@ -74,6 +80,50 @@ def run_coro(
             future.cancel()
             raise
 
+    def submit_coro(self, coro: Coroutine) -> None:
+        """Schedule a coroutine on the event loop without waiting for its result.
+
+        Use this for work whose result nothing depends on, so the calling thread is not blocked by the round
+        trip. Failures are logged, as no caller is left to propagate them to, and `close` cancels whatever is
+        still pending - call `wait_for_submitted` first if that matters.
+
+        Args:
+            coro: The coroutine to run.
+
+        Raises:
+            RuntimeError: If the event loop has been closed.
+        """
+        if self._eventloop.is_closed():
+            raise RuntimeError(f'The coroutine {coro} cannot be executed because the event loop is closed.')
+
+        # `wait_for_submitted` only runs once Scrapy goes idle, so without pruning here the list would hold
+        # every coroutine the whole crawl ever submitted, with its result.
+        if len(self._submitted) >= SUBMITTED_PRUNE_THRESHOLD:
+            self._submitted = [submitted for submitted in self._submitted if not submitted.done()]
+
+        future = asyncio.run_coroutine_threadsafe(coro, self._eventloop)
+        future.add_done_callback(self._log_failure)
+        self._submitted.append(future)
+
+    def wait_for_submitted(self, timeout: timedelta | None = None) -> None:
+        """Block until the coroutines submitted via `submit_coro` have finished.
+
+        Use this before anything that would observe their effects, or before `close`, which cancels whatever is
+        still running. Coroutines that do not finish within the timeout stay tracked for the next call.
+
+        Args:
+            timeout: The maximum time to wait for the submitted coroutines. Pass `None` to use the
+                `default_timeout` passed to the constructor.
+        """
+        if timeout is None:
+            timeout = self._default_timeout
+
+        self._submitted = list(futures.wait(self._submitted, timeout=timeout.total_seconds()).not_done)
+
+        # Callers rely on the effects having landed, so a timeout has to be visible.
+        if self._submitted:
+            logger.warning(f'{len(self._submitted)} submitted coroutines did not finish within the timeout.')
+
     def close(self, timeout: timedelta | None = None) -> None:
         """Close the event loop and its thread gracefully.
 
@@ -110,6 +160,15 @@ def close(self, timeout: timedelta | None = None) -> None:
                 logger.warning('Event loop thread did not exit cleanly! Forcing shutdown...')
                 self._force_exit_event_loop()
 
+    @staticmethod
+    def _log_failure(future: futures.Future) -> None:
+        """Log the failure of a coroutine submitted via `submit_coro`."""
+        if future.cancelled():
+            return
+
+        if (exc := future.exception()) is not None:
+            logger.error('A coroutine submitted to the event loop failed.', exc_info=exc)
+
     def _start_event_loop(self) -> None:
         """Set up and run the asyncio event loop in the dedicated thread."""
         asyncio.set_event_loop(self._eventloop)
diff --git a/src/apify/scrapy/requests.py b/src/apify/scrapy/requests.py
index 8bf99ec5..caa62413 100644
--- a/src/apify/scrapy/requests.py
+++ b/src/apify/scrapy/requests.py
@@ -76,8 +76,14 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ
     try:
         if scrapy_request.dont_filter:
             request_kwargs['always_enqueue'] = True
-        elif scrapy_request.meta.get('apify_request_unique_key'):
-            request_kwargs['unique_key'] = scrapy_request.meta['apify_request_unique_key']
+        # Reuse the queue's unique key only while this is still the request it was minted for. Redirects
+        # (`Request.replace()`) and spiders forwarding `meta` to another URL both inherit the stamp, and
+        # reusing it there deduplicates the derived request against its parent. A stamp without a URL beside
+        # it was set by hand, so it is taken at face value.
+        elif (unique_key := scrapy_request.meta.get('apify_request_unique_key')) and (
+            scrapy_request.meta.get('apify_request_url', scrapy_request.url) == scrapy_request.url
+        ):
+            request_kwargs['unique_key'] = unique_key
 
         # Serialize the Scrapy request now, before `Request.from_url()` runs below. `from_url()` mutates the
         # `user_data` dict it receives in place (it injects a live `CrawleeRequestData` under `__crawlee`), and that
@@ -187,21 +193,14 @@ def to_scrapy_request(apify_request: ApifyRequest, spider: Spider) -> ScrapyRequ
         if not isinstance(scrapy_request, ScrapyRequest):
             raise TypeError('scrapy_request must be an instance of the ScrapyRequest class')
 
-        # Update the meta field with the meta field from the apify_request
-        meta = scrapy_request.meta or {}
-        meta.update({'apify_request_unique_key': apify_request.unique_key})
-        # scrapy_request.meta is a property, so we have to set it like this
-        scrapy_request._meta = meta  # noqa: SLF001
-
     # If the apify_request comes directly from the Scrapy, typically start URLs.
     else:
-        scrapy_request = ScrapyRequest(
-            url=apify_request.url,
-            method=apify_request.method,
-            meta={
-                'apify_request_unique_key': apify_request.unique_key,
-            },
-        )
+        scrapy_request = ScrapyRequest(url=apify_request.url, method=apify_request.method)
+
+    # Stamp the unique key together with the URL it belongs to, so `to_apify_request` can tell this request
+    # apart from the ones Scrapy derives from it.
+    scrapy_request.meta['apify_request_unique_key'] = apify_request.unique_key
+    scrapy_request.meta['apify_request_url'] = scrapy_request.url
 
     # Add optional 'headers' field
     if apify_request.headers:
diff --git a/src/apify/scrapy/scheduler.py b/src/apify/scrapy/scheduler.py
index 0646d3d6..72f5656c 100644
--- a/src/apify/scrapy/scheduler.py
+++ b/src/apify/scrapy/scheduler.py
@@ -19,6 +19,8 @@
     from scrapy.http.request import Request
     from twisted.internet.defer import Deferred
 
+    from apify import Request as ApifyRequest
+
 logger = getLogger(__name__)
 
 
@@ -28,7 +30,11 @@ class ApifyScheduler(BaseScheduler):
     This scheduler requires the asyncio Twisted reactor to be installed.
     """
 
-    def __init__(self, async_thread_timeout: timedelta = timedelta(seconds=60)) -> None:
+    def __init__(
+        self,
+        async_thread_timeout: timedelta = timedelta(seconds=60),
+        crawler: Crawler | None = None,
+    ) -> None:
         if not is_asyncio_reactor_installed():
             raise ValueError(
                 f'{ApifyScheduler.__qualname__} requires the asyncio Twisted reactor. '
@@ -37,6 +43,10 @@ def __init__(self, async_thread_timeout: timedelta = timedelta(seconds=60)) -> N
             )
         self._rq: RequestQueue | None = None
         self.spider: Spider | None = None
+        self._crawler = crawler
+
+        self._requests_in_flight: list[tuple[ApifyRequest, Request]] = []
+        """Requests handed over to Scrapy and not resolved in the request queue yet."""
 
         # A thread with the asyncio event loop to run coroutines on.
         self._async_thread = AsyncThread(default_timeout=async_thread_timeout)
@@ -49,7 +59,7 @@ def from_crawler(cls, crawler: Crawler) -> ApifyScheduler:
         background event loop may take before timing out; it defaults to 60 seconds.
         """
         timeout_secs = crawler.settings.getint('APIFY_ASYNC_THREAD_TIMEOUT_SECS', 60)
-        return cls(async_thread_timeout=timedelta(seconds=timeout_secs))
+        return cls(async_thread_timeout=timedelta(seconds=timeout_secs), crawler=crawler)
 
     def open(self, spider: Spider) -> Deferred[None] | None:
         """Open the scheduler.
@@ -86,12 +96,34 @@ async def open_rq() -> RequestQueue:
     def close(self, reason: str) -> None:
         """Close the scheduler.
 
-        Shut down the event loop and its thread gracefully.
+        Resolve the requests Scrapy still holds, then shut down the event loop and its thread gracefully.
 
         Args:
             reason: The reason for closing the spider.
         """
         logger.debug(f'Closing {self.__class__.__name__} due to {reason}...')
+
+        rq = self._rq
+        if isinstance(rq, RequestQueue):
+            try:
+                self._resolve_finished_requests(wait=True)
+            except Exception:
+                logger.exception('Failed to resolve the requests still in flight in the request queue.')
+
+            # Whatever Scrapy did not finish goes back to the queue, so the next run gets it as pending.
+            # One failed reclaim must not strand the rest.
+            for apify_request, _ in self._requests_in_flight:
+                try:
+                    self._async_thread.run_coro(rq.reclaim_request(apify_request))
+                except Exception:
+                    logger.exception(f'Failed to reclaim the request {apify_request} in the request queue.')
+
+            self._requests_in_flight.clear()
+
+        # Closing the event loop cancels the updates fired off on the hot path silently, leaving those
+        # requests unhandled.
+        self._async_thread.wait_for_submitted()
+
         try:
             self._async_thread.close()
 
@@ -107,12 +139,21 @@ def close(self, reason: str) -> None:
     def has_pending_requests(self) -> bool:
         """Check if the scheduler has any pending requests.
 
+        Resolves the requests Scrapy has finished with first, as their outcome is what decides the answer.
+
         Returns:
             True if the scheduler has any pending requests, False otherwise.
         """
         if not isinstance(self._rq, RequestQueue):
             raise TypeError('self._rq must be an instance of the RequestQueue class')
 
+        # Scrapy asks this only once both its downloader and its scraper are idle, so everything still tracked
+        # as in flight is provably finished.
+        self._resolve_finished_requests(wait=True)
+
+        # The queue answers from its own bookkeeping, which a pending update has not reached yet.
+        self._async_thread.wait_for_submitted()
+
         # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is
         # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery.
         try:
@@ -164,6 +205,10 @@ def next_request(self) -> Request | None:
         if not isinstance(self._rq, RequestQueue):
             raise TypeError('self._rq must be an instance of the RequestQueue class')
 
+        # The engine polls this method throughout the crawl, so resolving here keeps the queue current
+        # without blocking on the round trips.
+        self._resolve_finished_requests(wait=False)
+
         # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is
         # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery.
         try:
@@ -178,26 +223,68 @@ def next_request(self) -> Request | None:
         if not isinstance(self.spider, Spider):
             raise TypeError('self.spider must be an instance of the Spider class')
 
-        # Reconstruct the Scrapy request before consuming the queue entry. A malformed entry must not crash
-        # the whole run, so on failure it is logged and skipped (None) rather than propagating.
+        # A corrupt or legacy payload must not crash the run, and is marked as handled right away, otherwise
+        # the queue would keep handing it back forever.
         try:
             scrapy_request = to_scrapy_request(apify_request, spider=self.spider)
         except Exception as exc:
             logger.warning(f'Failed to convert Apify request {apify_request} to a Scrapy request; skipping it: {exc}')
-            scrapy_request = None
-
-        # Mark the request as handled. This runs even when reconstruction failed above: an unrecoverable entry
-        # (a corrupt or legacy payload) must still be consumed, otherwise the queue would keep handing it back
-        # forever. Retrying genuine failures is the RetryMiddleware's job.
-        # Log here before re-raising: this coroutine ran on a separate event-loop thread, and the failure is
-        # otherwise easy to lose as it crosses that thread boundary back into Scrapy's synchronous machinery.
-        try:
-            self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request))
-        except Exception:
-            logger.exception('Failed to mark the request as handled in the request queue.')
-            raise
-
-        if scrapy_request is None:
+            try:
+                self._async_thread.run_coro(self._rq.mark_request_as_handled(apify_request))
+            except Exception:
+                logger.exception('Failed to mark the request as handled in the request queue.')
+                raise
             return None
 
+        # The entry stays unresolved until Scrapy is done with the request, so a run interrupted mid-flight
+        # leaves it pending instead of silently handled.
+        self._requests_in_flight.append((apify_request, scrapy_request))
+
         return scrapy_request
+
+    def _requests_busy_in_scrapy(self) -> set[Request]:
+        """Return the requests Scrapy is still working on.
+
+        A request joins the downloader's active set before the middleware chain runs and leaves the scraper's
+        only once the callback and the item pipeline are done, so absence from both means Scrapy has finished
+        with it - downloaded, dropped by a middleware or errored out alike.
+        """
+        engine = self._crawler.engine if self._crawler is not None else None
+        if engine is None:
+            return set()
+
+        scraper_slot = engine.scraper.slot
+        return engine.downloader.active | (scraper_slot.active if scraper_slot is not None else set())
+
+    def _resolve_finished_requests(self, *, wait: bool) -> None:
+        """Mark every request Scrapy has finished processing as handled in the request queue.
+
+        A request whose update cannot be dispatched stays tracked for the next call to retry, without holding
+        up the rest of the list.
+
+        Args:
+            wait: Whether to block until the queue has been updated. Pass False on the crawl's hot path, where
+                nothing depends on the result and blocking would stall the Twisted reactor.
+        """
+        rq = self._rq
+        if not self._requests_in_flight or not isinstance(rq, RequestQueue):
+            return
+
+        busy = self._requests_busy_in_scrapy()
+        unresolved: list[tuple[ApifyRequest, Request]] = []
+
+        for apify_request, scrapy_request in self._requests_in_flight:
+            if scrapy_request in busy:
+                unresolved.append((apify_request, scrapy_request))
+                continue
+
+            try:
+                if wait:
+                    self._async_thread.run_coro(rq.mark_request_as_handled(apify_request))
+                else:
+                    self._async_thread.submit_coro(rq.mark_request_as_handled(apify_request))
+            except Exception:
+                logger.exception(f'Failed to mark the request {apify_request} as handled in the request queue.')
+                unresolved.append((apify_request, scrapy_request))
+
+        self._requests_in_flight = unresolved
diff --git a/tests/unit/scrapy/requests/test_to_apify_request.py b/tests/unit/scrapy/requests/test_to_apify_request.py
index 97902f7d..754e0290 100644
--- a/tests/unit/scrapy/requests/test_to_apify_request.py
+++ b/tests/unit/scrapy/requests/test_to_apify_request.py
@@ -10,6 +10,7 @@
 
 from crawlee._types import HttpHeaders
 
+from apify import Request as ApifyRequest
 from apify.scrapy.requests import to_apify_request, to_scrapy_request
 
 
@@ -187,3 +188,36 @@ def test_apify_request_id_in_meta_is_ignored(spider: Spider) -> None:
 
     assert apify_request is not None
     assert apify_request.unique_key == 'https://example.com'
+
+
+def test_unchanged_request_keeps_the_unique_key_it_was_stamped_with(spider: Spider) -> None:
+    """A request handed to Scrapy and enqueued again unchanged reuses the unique key it was minted for."""
+    scrapy_request = to_scrapy_request(ApifyRequest.from_url('https://example.com'), spider)
+
+    apify_request = to_apify_request(scrapy_request, spider)
+
+    assert apify_request is not None
+    assert apify_request.unique_key == scrapy_request.meta['apify_request_unique_key']
+
+
+def test_redirected_request_does_not_inherit_the_parents_unique_key(spider: Spider) -> None:
+    """A redirect derived from a fetched request gets its own unique key instead of the parent's stamp."""
+    parent = to_scrapy_request(ApifyRequest.from_url('https://example.com/redirect'), spider)
+    redirected = parent.replace(url='https://example.com/target')
+
+    apify_request = to_apify_request(redirected, spider)
+
+    assert apify_request is not None
+    assert apify_request.url == 'https://example.com/target'
+    assert apify_request.unique_key != parent.meta['apify_request_unique_key']
+
+
+def test_follow_up_request_with_propagated_meta_gets_its_own_unique_key(spider: Spider) -> None:
+    """A spider callback forwarding `meta` verbatim to another URL must not reuse the parent's unique key."""
+    parent = to_scrapy_request(ApifyRequest.from_url('https://example.com/listing'), spider)
+    follow_up = Request(url='https://example.com/detail', meta=parent.meta)
+
+    apify_request = to_apify_request(follow_up, spider)
+
+    assert apify_request is not None
+    assert apify_request.unique_key != parent.meta['apify_request_unique_key']
diff --git a/tests/unit/scrapy/requests/test_to_scrapy_request.py b/tests/unit/scrapy/requests/test_to_scrapy_request.py
index 898312f2..c3803d7b 100644
--- a/tests/unit/scrapy/requests/test_to_scrapy_request.py
+++ b/tests/unit/scrapy/requests/test_to_scrapy_request.py
@@ -68,6 +68,21 @@ def test_without_reconstruction(spider: Spider) -> None:
     assert apify_request.unique_key == scrapy_request.meta.get('apify_request_unique_key')
 
 
+def test_unique_key_is_stamped_together_with_its_url(spider: Spider) -> None:
+    """The queue's unique key is stamped alongside the URL it belongs to, so derived requests can be told apart."""
+    apify_request = ApifyRequest(
+        url='https://example.com',
+        method='GET',
+        unique_key='https://example.com',
+        user_data={},
+    )
+
+    scrapy_request = to_scrapy_request(apify_request, spider)
+
+    assert scrapy_request.meta['apify_request_unique_key'] == apify_request.unique_key
+    assert scrapy_request.meta['apify_request_url'] == scrapy_request.url
+
+
 def test_without_reconstruction_with_optional_fields(spider: Spider) -> None:
     """The without-reconstruction path also carries optional headers and user data to the Scrapy request."""
     apify_request = ApifyRequest(
diff --git a/tests/unit/scrapy/test_async_thread.py b/tests/unit/scrapy/test_async_thread.py
index 3cf51b62..7030d51f 100644
--- a/tests/unit/scrapy/test_async_thread.py
+++ b/tests/unit/scrapy/test_async_thread.py
@@ -11,7 +11,7 @@
 import pytest
 
 from ..._utils import poll_until_condition
-from apify.scrapy._async_thread import AsyncThread
+from apify.scrapy._async_thread import SUBMITTED_PRUNE_THRESHOLD, AsyncThread
 
 
 async def _return(value: int) -> int:
@@ -161,3 +161,117 @@ async def boom() -> None:
     # The loop was stopped and its thread joined despite the failing cancellation, so nothing is left running.
     assert not thread._thread.is_alive()
     assert thread._eventloop.is_closed()
+
+
+def test_submit_coro_runs_the_coroutine_without_blocking() -> None:
+    """`submit_coro` schedules the coroutine on the background loop and returns before it completes."""
+    thread = AsyncThread()
+    _wait_until_running(thread)
+
+    release = threading.Event()
+    finished = threading.Event()
+
+    async def gated() -> None:
+        await asyncio.to_thread(release.wait)
+        finished.set()
+
+    thread.submit_coro(gated())
+
+    # The call returned while the coroutine is still parked on the gate.
+    assert not finished.is_set()
+
+    release.set()
+    assert finished.wait(timeout=2)
+
+    thread.close()
+
+
+def test_submit_coro_logs_a_failing_coroutine(caplog: pytest.LogCaptureFixture) -> None:
+    """A coroutine submitted without a caller to propagate to has its failure logged instead of swallowed."""
+    thread = AsyncThread()
+    _wait_until_running(thread)
+
+    async def boom() -> None:
+        raise RuntimeError('boom')
+
+    with caplog.at_level(logging.ERROR, logger='apify.scrapy._async_thread'):
+        thread.submit_coro(boom())
+        thread.close()
+
+    errors = [record for record in caplog.records if record.levelno >= logging.ERROR]
+    assert len(errors) == 1
+    assert errors[0].exc_info is not None
+    assert str(errors[0].exc_info[1]) == 'boom'
+
+
+def test_submit_coro_raises_after_close() -> None:
+    """`submit_coro` raises `RuntimeError` once the loop has been closed."""
+    thread = AsyncThread()
+    thread.close()
+
+    coro = _return(42)
+    with pytest.raises(RuntimeError):
+        thread.submit_coro(coro)
+    coro.close()
+
+
+def test_wait_for_submitted_blocks_until_the_coroutines_finish() -> None:
+    """`wait_for_submitted` waits for the fire-and-forget coroutines, so `close` cannot cancel them."""
+    thread = AsyncThread()
+    _wait_until_running(thread)
+
+    release = threading.Event()
+    finished = threading.Event()
+
+    async def gated() -> None:
+        await asyncio.to_thread(release.wait)
+        finished.set()
+
+    thread.submit_coro(gated())
+    release.set()
+
+    thread.wait_for_submitted()
+
+    assert finished.is_set()
+    thread.close()
+
+
+def test_wait_for_submitted_keeps_an_unfinished_coroutine_tracked(caplog: pytest.LogCaptureFixture) -> None:
+    """A coroutine that outlasts the timeout stays tracked and is reported, so a later call can wait for it."""
+    thread = AsyncThread()
+    _wait_until_running(thread)
+
+    release = threading.Event()
+
+    async def gated() -> None:
+        await asyncio.to_thread(release.wait)
+
+    thread.submit_coro(gated())
+
+    with caplog.at_level(logging.WARNING, logger='apify.scrapy._async_thread'):
+        thread.wait_for_submitted(timeout=timedelta(seconds=0.01))
+    assert len(thread._submitted) == 1
+    assert [record for record in caplog.records if record.levelno == logging.WARNING]
+
+    release.set()
+    thread.wait_for_submitted()
+    assert thread._submitted == []
+
+    thread.close()
+
+
+def test_submit_coro_drops_the_finished_futures() -> None:
+    """Only the coroutines still running stay tracked, so a long crawl does not pile up finished futures."""
+    thread = AsyncThread()
+    _wait_until_running(thread)
+
+    for _ in range(SUBMITTED_PRUNE_THRESHOLD):
+        thread.submit_coro(_return(1))
+
+    assert futures.wait(list(thread._submitted), timeout=2).not_done == set()
+
+    # Every tracked coroutine has finished, so this submission drops them instead of growing the list.
+    thread.submit_coro(_return(1))
+    assert len(thread._submitted) == 1
+
+    thread.close()
diff --git a/tests/unit/scrapy/test_scheduler.py b/tests/unit/scrapy/test_scheduler.py
index a7cc4445..c4032955 100644
--- a/tests/unit/scrapy/test_scheduler.py
+++ b/tests/unit/scrapy/test_scheduler.py
@@ -25,6 +25,20 @@ def spider() -> DummySpider:
     return DummySpider()
 
 
+def fake_crawler(
+    *,
+    downloader_busy: set[Request] | None = None,
+    scraper_busy: set[Request] | None = None,
+) -> Any:
+    """Build a crawler stub reporting the given requests as busy; without `scraper_busy` its scraper slot is None."""
+    scraper_slot = SimpleNamespace(active=scraper_busy) if scraper_busy is not None else None
+    engine = SimpleNamespace(
+        downloader=SimpleNamespace(active=downloader_busy if downloader_busy is not None else set()),
+        scraper=SimpleNamespace(slot=scraper_slot),
+    )
+    return SimpleNamespace(engine=engine)
+
+
 @pytest.fixture
 def scheduler(monkeypatch: pytest.MonkeyPatch, spider: DummySpider) -> ApifyScheduler:
     """Create a scheduler with its reactor check and async thread stubbed out."""
@@ -124,7 +138,7 @@ def test_next_request_skips_request_that_fails_to_convert(
 
 
 def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> None:
-    """A valid queue entry is reconstructed into a Scrapy request and marked handled."""
+    """A valid queue entry is reconstructed into a Scrapy request and left unhandled until Scrapy is done."""
     rq = cast('mock.MagicMock', scheduler._rq)
     async_thread = cast('mock.MagicMock', scheduler._async_thread)
 
@@ -134,13 +148,13 @@ def test_next_request_returns_converted_request(scheduler: ApifyScheduler) -> No
         unique_key='https://example.com',
         user_data={},
     )
-    async_thread.run_coro.side_effect = [apify_request, None]
+    async_thread.run_coro.return_value = apify_request
 
     result = scheduler.next_request()
 
     assert isinstance(result, Request)
     assert result.url == apify_request.url
-    rq.mark_request_as_handled.assert_called_once_with(apify_request)
+    rq.mark_request_as_handled.assert_not_called()
 
 
 def test_next_request_returns_none_when_queue_empty(scheduler: ApifyScheduler) -> None:
@@ -190,3 +204,209 @@ def __init__(self, default_timeout: timedelta | None = None) -> None:
     ApifyScheduler.from_crawler(cast('Any', crawler))
 
     assert captured['default_timeout'] == timedelta(seconds=123)
+
+
+def test_has_pending_requests_marks_finished_requests_as_handled(scheduler: ApifyScheduler) -> None:
+    """Requests Scrapy has finished with are marked as handled once it goes idle and asks about pending work."""
+    rq = cast('mock.MagicMock', scheduler._rq)
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+
+    apify_request = ApifyRequest(
+        url='https://example.com',
+        method='GET',
+        unique_key='https://example.com',
+        user_data={},
+    )
+    async_thread.run_coro.return_value = apify_request
+    scheduler.next_request()
+    rq.mark_request_as_handled.assert_not_called()
+
+    # Scrapy asks about pending work only once its downloader and its scraper are both idle.
+    scheduler._crawler = fake_crawler(scraper_busy=set())
+    async_thread.run_coro.return_value = True  # the queue reports itself finished
+    assert scheduler.has_pending_requests() is False
+
+    rq.mark_request_as_handled.assert_called_once_with(apify_request)
+
+
+def test_next_request_marks_finished_requests_without_blocking(scheduler: ApifyScheduler) -> None:
+    """On the crawl's hot path a finished request is marked as handled without blocking the reactor on it."""
+    rq = cast('mock.MagicMock', scheduler._rq)
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+
+    apify_request = ApifyRequest(
+        url='https://example.com',
+        method='GET',
+        unique_key='https://example.com',
+        user_data={},
+    )
+    async_thread.run_coro.return_value = apify_request
+    scrapy_request = scheduler.next_request()
+
+    # The queue is drained from here on, so no further request is handed out.
+    async_thread.run_coro.return_value = None
+
+    # Scrapy is still downloading the request, so it stays unresolved.
+    scheduler._crawler = fake_crawler(downloader_busy={cast('Request', scrapy_request)})
+    assert scheduler.next_request() is None
+    async_thread.submit_coro.assert_not_called()
+
+    # Scrapy is done with it, so it is resolved off the reactor thread instead of blocking on the round trip.
+    scheduler._crawler = fake_crawler(scraper_busy=set())
+    assert scheduler.next_request() is None
+
+    rq.mark_request_as_handled.assert_called_once_with(apify_request)
+    async_thread.submit_coro.assert_called_once_with(rq.mark_request_as_handled.return_value)
+
+
+def test_has_pending_requests_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler) -> None:
+    """The queue is asked whether it is finished only after the updates fired off on the hot path have landed."""
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+    scheduler._crawler = fake_crawler()
+
+    async_thread.run_coro.return_value = True  # the queue reports itself finished
+    assert scheduler.has_pending_requests() is False
+
+    # The queue answers from its own bookkeeping, which a pending update has not reached yet.
+    assert async_thread.mock_calls == [
+        mock.call.wait_for_submitted(),
+        mock.call.run_coro(cast('mock.MagicMock', scheduler._rq).is_finished()),
+    ]
+
+
+@pytest.mark.parametrize(
+    'busy_kwarg',
+    [
+        pytest.param('downloader_busy', id='busy in the downloader'),
+        pytest.param('scraper_busy', id='busy in the scraper slot'),
+    ],
+)
+def test_close_reclaims_requests_scrapy_never_finished(scheduler: ApifyScheduler, busy_kwarg: str) -> None:
+    """Requests still being processed when the scheduler closes go back to the queue instead of being lost."""
+    rq = cast('mock.MagicMock', scheduler._rq)
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+
+    apify_request = ApifyRequest(
+        url='https://example.com',
+        method='GET',
+        unique_key='https://example.com',
+        user_data={},
+    )
+    async_thread.run_coro.return_value = apify_request
+    scrapy_request = scheduler.next_request()
+
+    # Scrapy is still working on the request when the run is interrupted.
+    scheduler._crawler = fake_crawler(**{busy_kwarg: {cast('Request', scrapy_request)}})
+
+    scheduler.close('shutdown')
+
+    rq.reclaim_request.assert_called_once_with(apify_request)
+    rq.mark_request_as_handled.assert_not_called()
+
+
+def test_close_marks_the_requests_scrapy_finished_as_handled(scheduler: ApifyScheduler) -> None:
+    """Requests Scrapy drained before the shutdown are marked as handled rather than reclaimed."""
+    rq = cast('mock.MagicMock', scheduler._rq)
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+
+    apify_request = ApifyRequest(
+        url='https://example.com',
+        method='GET',
+        unique_key='https://example.com',
+        user_data={},
+    )
+    async_thread.run_coro.return_value = apify_request
+    scheduler.next_request()
+
+    # Scrapy drains its downloader and its scraper before the scheduler is closed.
+    scheduler._crawler = fake_crawler(scraper_busy=set())
+
+    scheduler.close('finished')
+
+    rq.mark_request_as_handled.assert_called_once_with(apify_request)
+    rq.reclaim_request.assert_not_called()
+
+
+def test_close_reclaims_the_other_requests_after_a_failed_reclaim(
+    scheduler: ApifyScheduler,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """One failing reclaim does not stop the other in-flight requests from going back to the queue."""
+    rq = cast('mock.MagicMock', scheduler._rq)
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+
+    apify_requests = [
+        ApifyRequest(
+            url=f'https://example.com/{index}',
+            method='GET',
+            unique_key=f'https://example.com/{index}',
+            user_data={},
+        )
+        for index in range(2)
+    ]
+
+    # The crawler stub keeps a reference to this set, so both requests stay busy as they are handed out.
+    busy: set[Request] = set()
+    scheduler._crawler = fake_crawler(downloader_busy=busy)
+
+    async_thread.run_coro.side_effect = apify_requests
+    for _ in apify_requests:
+        busy.add(cast('Request', scheduler.next_request()))
+
+    async_thread.run_coro.side_effect = [RuntimeError('boom'), None]
+
+    with caplog.at_level(logging.ERROR, logger='apify.scrapy.scheduler'):
+        scheduler.close('shutdown')
+
+    assert rq.reclaim_request.call_count == len(apify_requests)
+    errors = [record for record in caplog.records if record.levelno >= logging.ERROR]
+    assert len(errors) == 1
+
+
+def test_close_waits_for_the_non_blocking_updates(scheduler: ApifyScheduler) -> None:
+    """The event loop is not torn down before the updates fired off on the hot path have landed."""
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+
+    scheduler.close('finished')
+
+    calls = async_thread.mock_calls
+    assert calls.index(mock.call.wait_for_submitted()) < calls.index(mock.call.close())
+
+
+def test_a_failed_mark_keeps_the_request_tracked(
+    scheduler: ApifyScheduler,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A request whose mark-as-handled fails stays tracked, so the next resolution retries it."""
+    async_thread = cast('mock.MagicMock', scheduler._async_thread)
+
+    apify_request = ApifyRequest(
+        url='https://example.com',
+        method='GET',
+        unique_key='https://example.com',
+        user_data={},
+    )
+    async_thread.run_coro.return_value = apify_request
+    scheduler.next_request()
+
+    scheduler._crawler = fake_crawler()
+    # The mark fails, then the queue reports itself unfinished because the request is still in progress.
+    async_thread.run_coro.side_effect = [RuntimeError('boom'), False]
+
+    with caplog.at_level(logging.ERROR, logger='apify.scrapy.scheduler'):
+        assert scheduler.has_pending_requests() is True
+
+    assert scheduler._requests_in_flight
+    errors = [record for record in caplog.records if record.levelno >= logging.ERROR]
+    assert len(errors) == 1
+
+
+def test_from_crawler_keeps_the_crawler(monkeypatch: pytest.MonkeyPatch) -> None:
+    """`from_crawler` keeps the crawler, which is how the scheduler learns what Scrapy is still working on."""
+    monkeypatch.setattr('apify.scrapy.scheduler.is_asyncio_reactor_installed', lambda: True)
+    monkeypatch.setattr('apify.scrapy.scheduler.AsyncThread', mock.MagicMock())
+
+    crawler = SimpleNamespace(settings=Settings())
+    scheduler = ApifyScheduler.from_crawler(cast('Any', crawler))
+
+    assert scheduler._crawler is crawler