From cb1074e989bc6c5de0ed98875d30b605c9871fde Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Wed, 15 Jul 2026 11:47:01 +0100
Subject: [PATCH 1/9] Fix cat_ranges on a cold cache in whole-file caching
 filesystems

SimpleCacheFileSystem.cat_ranges compared _check_file() results with
"is False", but its _check_file returns None for missing entries, so
uncached files were never downloaded and the subsequent local read
failed. WholeFileCacheFileSystem._cat_ranges had the inverse problem
("is None" while the metadata-backed _check_file returns False), never
resolved a local path for repeats of one path within a single call, and
forwarded on_error into self.fs._get, which passes it on to the target
filesystem's _get_file (aiohttp, for one, rejects it).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 fsspec/implementations/cached.py            | 42 ++++++----
 fsspec/implementations/tests/test_cached.py | 89 +++++++++++++++++++++
 2 files changed, 115 insertions(+), 16 deletions(-)

diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 08b4ad579..04a936560 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import inspect
 import logging
 import os
@@ -784,20 +785,25 @@ async def _cat_ranges(
     ):
         logger.debug("async cat ranges %s", paths)
         lpaths = []
-        rset = set()
-        download = []
-        rpaths = []
+        need = {}
         for p in paths:
             fn = self._check_file(p)
-            if fn is None and p not in rset:
-                sha = self._mapper(p)
-                fn = os.path.join(self.storage[-1], sha)
-                download.append(fn)
-                rset.add(p)
-                rpaths.append(p)
+            if isinstance(fn, tuple):
+                fn = fn[1]
+            if not fn:
+                fn = need.setdefault(p, os.path.join(self.storage[-1], self._mapper(p)))
             lpaths.append(fn)
-        if download:
-            await self.fs._get(rpaths, download, on_error=on_error)
+        if need:
+            # a batch self.fs._get would forward on_error to the target
+            # filesystem's _get_file, which does not accept it
+            results = await asyncio.gather(
+                *(self.fs._get_file(rpath, lpath) for rpath, lpath in need.items()),
+                return_exceptions=True,
+            )
+            if on_error == "raise":
+                for res in results:
+                    if isinstance(res, BaseException):
+                        raise res
 
         return LocalFileSystem().cat_ranges(
             lpaths, starts, ends, max_gap=max_gap, on_error=on_error, **kwargs
@@ -912,12 +918,16 @@ def cat_ranges(
     ):
         logger.debug("cat ranges %s", paths)
         lpaths = [self._check_file(p) for p in paths]
-        rpaths = [p for l, p in zip(lpaths, paths) if l is False]
-        lpaths = [l for l, p in zip(lpaths, paths) if l is False]
-        self.fs.get(rpaths, lpaths)
-        paths = [self._check_file(p) for p in paths]
+        need = {
+            p: os.path.join(self.storage[-1], self._mapper(p))
+            for l, p in zip(lpaths, paths)
+            if l is None
+        }
+        if need:
+            self.fs.get(list(need), list(need.values()))
+        lpaths = [need[p] if l is None else l for l, p in zip(lpaths, paths)]
         return LocalFileSystem().cat_ranges(
-            paths, starts, ends, max_gap=max_gap, on_error=on_error, **kwargs
+            lpaths, starts, ends, max_gap=max_gap, on_error=on_error, **kwargs
         )
 
     def _get_cached_file_before_open(self, path, **kwargs):
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index 794eea061..a55e9751b 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -1,7 +1,12 @@
+import asyncio
 import json
 import os
+import random
 import shutil
 import tempfile
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 import pytest
 
@@ -1405,3 +1410,87 @@ def test_class_has_cat_file_and_cat_ranges(tmp_path, protocol):
     for attr in ("_cat_file", "_cat_ranges"):
         assert hasattr(fs, attr), f"instance missing {attr}"
         assert hasattr(type(fs), attr), f"class missing {attr}"
+
+
+@pytest.fixture(scope="module")
+def slow_http_server():
+    """A local HTTP server that streams a 1 MiB payload slowly.
+
+    Downloads take long enough that staggered concurrent reads of the same
+    URL overlap with an in-flight download, exercising the cache-write race
+    of issue #639.
+    """
+    pytest.importorskip("aiohttp")
+
+    payload = random.Random(42).randbytes(2**20)
+
+    class Handler(BaseHTTPRequestHandler):
+        protocol_version = "HTTP/1.1"
+
+        def do_HEAD(self):
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+
+        def do_GET(self):
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(payload)))
+            self.end_headers()
+            for i in range(0, len(payload), 2**16):
+                try:
+                    self.wfile.write(payload[i : i + 2**16])
+                except (BrokenPipeError, ConnectionResetError):
+                    return
+                time.sleep(0.005)
+
+        def log_message(self, format, *args):
+            pass
+
+    class QuietServer(ThreadingHTTPServer):
+        def handle_error(self, request, client_address):
+            pass
+
+    server = QuietServer(("127.0.0.1", 0), Handler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    yield f"http://127.0.0.1:{server.server_address[1]}/data", payload
+    server.shutdown()
+
+
+def test_simplecache_cat_ranges_cold_cache(tmp_path):
+    # SimpleCacheFileSystem.cat_ranges compared _check_file() results with
+    # ``is False``, but _check_file returns None for missing entries, so
+    # uncached files were never downloaded and cat_ranges failed on a cold
+    # cache
+    mem = fsspec.filesystem("memory")
+    mem.pipe("/raw/one", b"0123456789")
+    mem.pipe("/raw/two", b"abcdefghij")
+    fs = fsspec.filesystem(
+        "simplecache",
+        fs=mem,
+        cache_storage=str(tmp_path / "cr"),
+        skip_instance_cache=True,
+    )
+    out = fs.cat_ranges(["/raw/one", "/raw/two", "/raw/one"], [0, 2, 4], [4, 6, 8])
+    assert out == [b"0123", b"cdef", b"4567"]
+
+
+@pytest.mark.parametrize("protocol", ["simplecache", "filecache"])
+def test_async_cat_ranges_cold_cache(slow_http_server, tmp_path, protocol):
+    # _cat_ranges compared _check_file() results with ``is None``, but
+    # filecache's _check_file returns False for missing entries, so uncached
+    # files were never downloaded; and repeats of one path within a single
+    # call (e.g. several ranges of one file) got no local path at all
+    url, payload = slow_http_server
+
+    async def run():
+        fs = fsspec.filesystem(
+            protocol,
+            target_protocol="http",
+            cache_storage=str(tmp_path / "cr"),
+            asynchronous=True,
+            target_options={"asynchronous": True, "skip_instance_cache": True},
+            skip_instance_cache=True,
+        )
+        return await fs._cat_ranges([url, url], [0, 10], [10, 20])
+
+    assert asyncio.run(run()) == [payload[0:10], payload[10:20]]

From 4246449bb7c21b335214d9b0b8225a1f2f2f4943 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Wed, 15 Jul 2026 11:49:40 +0100
Subject: [PATCH 2/9] Make whole-file cache writes atomic (download to temp
 file, then rename)

The whole-file caching filesystems (filecache, simplecache) downloaded
cache misses directly to the final cache filename, while readers treat
the existence of that filename as "download complete". A concurrent
reader of the same key could therefore observe - and return - a
partially-written cache file (issue #639).

All whole-file download paths (_open via _get_cached_file_before_open
including its compression branch, cat, open_many, cat_ranges, _cat_file
and _cat_ranges) now download to a temporary file created next to the
final name and os.replace() it into place, so an incomplete download is
never visible under the final filename. A lost race means the same
bytes are fetched twice and the last rename wins; no locking is needed.
Temporary files are removed on failure.

The regression tests stagger reader starts across the duration of a
throttled download: simultaneous starts would all miss the cache and
write identical bytes to identical offsets, hiding the race.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 fsspec/implementations/cached.py            | 129 ++++++++++++++------
 fsspec/implementations/tests/test_cached.py |  79 ++++++++++++
 2 files changed, 171 insertions(+), 37 deletions(-)

diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 04a936560..8fc6abee7 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -30,6 +30,86 @@
 logger = logging.getLogger("fsspec.cached")
 
 
+# Cache misses are downloaded to a temporary file in the cache directory and
+# then renamed into place, so that a partially-written download is never
+# visible under its final cache filename: readers treat the existence of that
+# filename as "download complete" (see issue #639). The rename is atomic, and
+# a concurrent duplicate download of the same path is harmless: both
+# temporary files hold identical bytes and the last rename wins. These
+# helpers are module-level functions rather than methods because
+# CachingFileSystem.__getattribute__ dispatches only known method names to
+# the caching class, delegating anything else to the wrapped filesystem.
+
+
+def _temppath(lpath):
+    fd, tmp = tempfile.mkstemp(
+        dir=os.path.dirname(lpath),
+        prefix=os.path.basename(lpath) + ".",
+        suffix=".part",
+    )
+    os.close(fd)
+    return tmp
+
+
+def _remove_tempfile(tmp):
+    try:
+        os.remove(tmp)
+    except OSError:
+        pass
+
+
+def _atomic_get_file(fs, rpath, lpath):
+    tmp = _temppath(lpath)
+    try:
+        fs.get_file(rpath, tmp)
+        os.replace(tmp, lpath)
+    except BaseException:
+        _remove_tempfile(tmp)
+        raise
+
+
+async def _atomic_get_file_async(fs, rpath, lpath, **kwargs):
+    tmp = _temppath(lpath)
+    try:
+        await fs._get_file(rpath, tmp, **kwargs)
+        os.replace(tmp, lpath)
+    except BaseException:
+        _remove_tempfile(tmp)
+        raise
+
+
+def _atomic_get(fs, rpaths, lpaths):
+    tmps = [_temppath(lpath) for lpath in lpaths]
+    try:
+        fs.get(rpaths, tmps)
+        for tmp, lpath in zip(tmps, lpaths):
+            os.replace(tmp, lpath)
+    except BaseException:
+        for tmp in tmps:
+            _remove_tempfile(tmp)
+        raise
+
+
+def _atomic_get_file_decompressed(fs, rpath, lpath, compression, **kwargs):
+    tmp = _temppath(lpath)
+    try:
+        with fs._open(rpath, mode="rb", **kwargs) as f, open(tmp, "wb") as f2:
+            if isinstance(f, AbstractBufferedFile):
+                # want no type of caching if just downloading whole thing
+                f.cache = BaseCache(0, f.cache.fetcher, f.size)
+            comp = infer_compression(rpath) if compression == "infer" else compression
+            f = compr[comp](f, mode="rb")
+            data = True
+            while data:
+                block = getattr(f, "blocksize", 5 * 2**20)
+                data = f.read(block)
+                f2.write(data)
+        os.replace(tmp, lpath)
+    except BaseException:
+        _remove_tempfile(tmp)
+        raise
+
+
 class WriteCachedTransaction(Transaction):
     def complete(self, commit=True):
         rpaths = [f.path for f in self.files]
@@ -603,7 +683,7 @@ def open_many(self, open_files, **kwargs):
         downfn = [fn for fn, d in zip(downfn0, details) if not d]
         if downpath:
             # skip if all files are already cached and up to date
-            self.fs.get(downpath, downfn)
+            _atomic_get(self.fs, downpath, downfn)
 
             # update metadata - only happens when downloads are successful
             newdetail = [
@@ -687,7 +767,7 @@ def cat(
                 paths.remove(p)
 
         if getpaths:
-            self.fs.get(getpaths, storepaths)
+            _atomic_get(self.fs, getpaths, storepaths)
             self.save_cache()
 
         callback.set_size(len(paths))
@@ -704,23 +784,9 @@ def _get_cached_file_before_open(self, path, **kwargs):
         # call target filesystems open
         self._mkcache()
         if self.compression:
-            with self.fs._open(path, mode="rb", **kwargs) as f, open(fn, "wb") as f2:
-                if isinstance(f, AbstractBufferedFile):
-                    # want no type of caching if just downloading whole thing
-                    f.cache = BaseCache(0, f.cache.fetcher, f.size)
-                comp = (
-                    infer_compression(path)
-                    if self.compression == "infer"
-                    else self.compression
-                )
-                f = compr[comp](f, mode="rb")
-                data = True
-                while data:
-                    block = getattr(f, "blocksize", 5 * 2**20)
-                    data = f.read(block)
-                    f2.write(data)
+            _atomic_get_file_decompressed(self.fs, path, fn, self.compression, **kwargs)
         else:
-            self.fs.get_file(path, fn)
+            _atomic_get_file(self.fs, path, fn)
         self.save_cache()
 
     def _open(self, path, mode="rb", **kwargs):
@@ -772,7 +838,7 @@ async def _cat_file(self, path, start=None, end=None, **kwargs):
 
         if not fn:
             fn = os.path.join(self.storage[-1], sha)
-            await self.fs._get_file(path, fn, **kwargs)
+            await _atomic_get_file_async(self.fs, path, fn, **kwargs)
 
         with open(fn, "rb") as f:  # noqa ASYNC230
             if start:
@@ -797,7 +863,10 @@ async def _cat_ranges(
             # a batch self.fs._get would forward on_error to the target
             # filesystem's _get_file, which does not accept it
             results = await asyncio.gather(
-                *(self.fs._get_file(rpath, lpath) for rpath, lpath in need.items()),
+                *(
+                    _atomic_get_file_async(self.fs, rpath, lpath)
+                    for rpath, lpath in need.items()
+                ),
                 return_exceptions=True,
             )
             if on_error == "raise":
@@ -924,7 +993,7 @@ def cat_ranges(
             if l is None
         }
         if need:
-            self.fs.get(list(need), list(need.values()))
+            _atomic_get(self.fs, list(need), list(need.values()))
         lpaths = [need[p] if l is None else l for l, p in zip(lpaths, paths)]
         return LocalFileSystem().cat_ranges(
             lpaths, starts, ends, max_gap=max_gap, on_error=on_error, **kwargs
@@ -939,23 +1008,9 @@ def _get_cached_file_before_open(self, path, **kwargs):
         self._cache_size = None
 
         if self.compression:
-            with self.fs._open(path, mode="rb", **kwargs) as f, open(fn, "wb") as f2:
-                if isinstance(f, AbstractBufferedFile):
-                    # want no type of caching if just downloading whole thing
-                    f.cache = BaseCache(0, f.cache.fetcher, f.size)
-                comp = (
-                    infer_compression(path)
-                    if self.compression == "infer"
-                    else self.compression
-                )
-                f = compr[comp](f, mode="rb")
-                data = True
-                while data:
-                    block = getattr(f, "blocksize", 5 * 2**20)
-                    data = f.read(block)
-                    f2.write(data)
+            _atomic_get_file_decompressed(self.fs, path, fn, self.compression, **kwargs)
         else:
-            self.fs.get_file(path, fn)
+            _atomic_get_file(self.fs, path, fn)
 
     def _open(self, path, mode="rb", **kwargs):
         path = self._strip_protocol(path)
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index a55e9751b..1ea8937ca 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -6,6 +6,7 @@
 import tempfile
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 import pytest
@@ -1456,6 +1457,84 @@ def handle_error(self, request, client_address):
     server.shutdown()
 
 
+def _staggered_async_cat_file(protocol, url, cache_dir, n):
+    # readers must start while a download is still in flight: simultaneous
+    # starts all miss the cache and write identical bytes to identical
+    # offsets, which would hide the race
+    async def run():
+        fs = fsspec.filesystem(
+            protocol,
+            target_protocol="http",
+            cache_storage=cache_dir,
+            asynchronous=True,
+            target_options={"asynchronous": True, "skip_instance_cache": True},
+            skip_instance_cache=True,
+        )
+
+        async def one(i):
+            await asyncio.sleep(i * 0.03)
+            return await fs._cat_file(url)
+
+        return await asyncio.gather(*[one(i) for i in range(n)])
+
+    return asyncio.run(run())
+
+
+def _staggered_threaded_cat_file(protocol, url, cache_dir, n):
+    fs = fsspec.filesystem(
+        protocol,
+        target_protocol="http",
+        cache_storage=cache_dir,
+        target_options={"skip_instance_cache": True},
+        skip_instance_cache=True,
+    )
+
+    def one(i):
+        time.sleep(i * 0.03)
+        return fs.cat_file(url)
+
+    with ThreadPoolExecutor(n) as pool:
+        return list(pool.map(one, range(n)))
+
+
+@pytest.mark.parametrize("protocol", ["simplecache", "filecache"])
+def test_concurrent_cat_file_async(slow_http_server, tmp_path, protocol):
+    """Concurrent reads of one uncached URL must all see complete bytes.
+
+    Regression test for https://github.com/fsspec/filesystem_spec/issues/639:
+    cache misses were downloaded directly to the final cache filename, so a
+    concurrent reader of the same key could observe (and return) a
+    partially-written file.
+    """
+    url, payload = slow_http_server
+    for trial in range(3):
+        data = _staggered_async_cat_file(
+            protocol, url, str(tmp_path / f"async{trial}"), 8
+        )
+        assert [len(d) for d in data] == [len(payload)] * 8
+        assert all(d == payload for d in data)
+
+
+def test_concurrent_cat_file_threads(slow_http_server, tmp_path):
+    """Same as test_concurrent_cat_file_async, for the sync open() path."""
+    url, payload = slow_http_server
+    for trial in range(3):
+        data = _staggered_threaded_cat_file(
+            "simplecache", url, str(tmp_path / f"thr{trial}"), 8
+        )
+        assert [len(d) for d in data] == [len(payload)] * 8
+        assert all(d == payload for d in data)
+
+
+def test_concurrent_downloads_leave_no_tempfiles(slow_http_server, tmp_path):
+    url, payload = slow_http_server
+    cache_dir = str(tmp_path / "clean")
+    _staggered_async_cat_file("simplecache", url, cache_dir, 4)
+    entries = os.listdir(cache_dir)
+    assert [fn for fn in entries if fn.endswith(".part")] == []
+    assert len(entries) == 1
+
+
 def test_simplecache_cat_ranges_cold_cache(tmp_path):
     # SimpleCacheFileSystem.cat_ranges compared _check_file() results with
     # ``is False``, but _check_file returns None for missing entries, so

From 029bdb5bc01159f73399a4b1ba69c63175558fa4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Thu, 16 Jul 2026 11:42:44 +0100
Subject: [PATCH 3/9] Clarify that mkstemp guarantees unique temp files for
 concurrent downloads

Review feedback on #2075: make explicit in the code that the temp
filename already contains a random token and is created with O_EXCL,
so concurrent downloads of the same key cannot clash.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 fsspec/implementations/cached.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 8fc6abee7..61e2cfee2 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -42,6 +42,8 @@
 
 
 def _temppath(lpath):
+    # mkstemp inserts a random token between prefix and suffix and creates the
+    # file with O_EXCL, so concurrent downloads of the same key never clash.
     fd, tmp = tempfile.mkstemp(
         dir=os.path.dirname(lpath),
         prefix=os.path.basename(lpath) + ".",

From d9ffb534eb9e00cf640adc338f7ae154a07e9df6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Tue, 28 Jul 2026 11:48:23 +0100
Subject: [PATCH 4/9] Generate temp filename without mkstemp; document leftover
 .part files

Per review: mkstemp's fd handling and 0o600 mode are POSIX-specific
choices, so _temppath now only generates a random name and the download's
own open() creates the file. Document (module comment and features.rst)
that .part files are removed when a download raises but can be left
behind on abrupt process exit, and are safe to delete.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 docs/source/features.rst         |  7 +++++++
 fsspec/implementations/cached.py | 23 ++++++++++++-----------
 2 files changed, 19 insertions(+), 11 deletions(-)

diff --git a/docs/source/features.rst b/docs/source/features.rst
index 18f1e80e6..8eed2d969 100644
--- a/docs/source/features.rst
+++ b/docs/source/features.rst
@@ -290,6 +290,13 @@ except without the options for cache expiry and to check the original source - i
 target can be considered static, and particularly where a large number of target files are expected
 (because no metadata is written to disc). Only "simplecache" is guaranteed thread/process-safe.
 
+With "filecache" and "simplecache", each file is first downloaded to a temporary file in the
+cache directory (the final cache filename plus a random token and a ``.part`` suffix) and then
+renamed into place, so that a partially-downloaded file is never visible under its final cache
+name. The temporary file is removed if the download raises an exception, but *not* if the
+process exits abruptly (e.g., it is killed) mid-download: leftover ``*.part`` files may then
+remain in the cache directory. They are ignored by the cache and are safe to delete.
+
 Remote Write Caching
 --------------------
 
diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 61e2cfee2..35aee9c0d 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -4,6 +4,7 @@
 import inspect
 import logging
 import os
+import secrets
 import tempfile
 import time
 import weakref
@@ -35,22 +36,22 @@
 # visible under its final cache filename: readers treat the existence of that
 # filename as "download complete" (see issue #639). The rename is atomic, and
 # a concurrent duplicate download of the same path is harmless: both
-# temporary files hold identical bytes and the last rename wins. These
-# helpers are module-level functions rather than methods because
+# temporary files hold identical bytes and the last rename wins. Temporary
+# files are removed when a download raises, but not on early exit (e.g. the
+# process being killed mid-download): stale "*.part" files may then be left
+# in the cache directory, are ignored by the cache, and are safe to delete.
+# These helpers are module-level functions rather than methods because
 # CachingFileSystem.__getattribute__ dispatches only known method names to
 # the caching class, delegating anything else to the wrapped filesystem.
 
 
 def _temppath(lpath):
-    # mkstemp inserts a random token between prefix and suffix and creates the
-    # file with O_EXCL, so concurrent downloads of the same key never clash.
-    fd, tmp = tempfile.mkstemp(
-        dir=os.path.dirname(lpath),
-        prefix=os.path.basename(lpath) + ".",
-        suffix=".part",
-    )
-    os.close(fd)
-    return tmp
+    # Only generates a name; the file itself is created by whatever plain
+    # open() downloads it, so it gets ordinary umask-derived permissions
+    # (mkstemp's fd handling and 0o600 mode are not portable choices for a
+    # shared cache directory). The random token keeps concurrent downloads
+    # of the same key on distinct temp files.
+    return f"{lpath}.{secrets.token_hex(8)}.part"
 
 
 def _remove_tempfile(tmp):

From 9028ef944d4ebd97136afb6b28197f6344cd02d8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Tue, 28 Jul 2026 11:56:23 +0100
Subject: [PATCH 5/9] Document rename-failure cleanup; test failed-download and
 stale .part paths

Follow-up from multi-agent verification of the review response: the docs
now name the rename as well as the download as cleanup triggers (the
clause the reviewer spelled out), the _temppath comment no longer
misattributes mkstemp's drawbacks to portability, and two tests pin the
newly documented guarantees (failing download leaves no .part nor final
file and the next read succeeds; a stale .part file is never mistaken
for a cache entry).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 docs/source/features.rst                    |  7 ++--
 fsspec/implementations/cached.py            | 14 ++++---
 fsspec/implementations/tests/test_cached.py | 46 +++++++++++++++++++++
 3 files changed, 58 insertions(+), 9 deletions(-)

diff --git a/docs/source/features.rst b/docs/source/features.rst
index 8eed2d969..c6f335cc4 100644
--- a/docs/source/features.rst
+++ b/docs/source/features.rst
@@ -293,9 +293,10 @@ target can be considered static, and particularly where a large number of target
 With "filecache" and "simplecache", each file is first downloaded to a temporary file in the
 cache directory (the final cache filename plus a random token and a ``.part`` suffix) and then
 renamed into place, so that a partially-downloaded file is never visible under its final cache
-name. The temporary file is removed if the download raises an exception, but *not* if the
-process exits abruptly (e.g., it is killed) mid-download: leftover ``*.part`` files may then
-remain in the cache directory. They are ignored by the cache and are safe to delete.
+name. The temporary file is removed if the download (or the final rename) raises an
+exception, but *not* if the process exits abruptly (e.g., it is killed) mid-download:
+leftover ``*.part`` files may then remain in the cache directory. They are ignored by the
+cache and are safe to delete.
 
 Remote Write Caching
 --------------------
diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 35aee9c0d..43cea7ce7 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -37,9 +37,10 @@
 # filename as "download complete" (see issue #639). The rename is atomic, and
 # a concurrent duplicate download of the same path is harmless: both
 # temporary files hold identical bytes and the last rename wins. Temporary
-# files are removed when a download raises, but not on early exit (e.g. the
-# process being killed mid-download): stale "*.part" files may then be left
-# in the cache directory, are ignored by the cache, and are safe to delete.
+# files are removed when the download or the final rename raises, but not on
+# early exit (e.g. the process being killed mid-download): stale "*.part"
+# files may then be left in the cache directory, are ignored by the cache,
+# and are safe to delete.
 # These helpers are module-level functions rather than methods because
 # CachingFileSystem.__getattribute__ dispatches only known method names to
 # the caching class, delegating anything else to the wrapped filesystem.
@@ -48,9 +49,10 @@
 def _temppath(lpath):
     # Only generates a name; the file itself is created by whatever plain
     # open() downloads it, so it gets ordinary umask-derived permissions
-    # (mkstemp's fd handling and 0o600 mode are not portable choices for a
-    # shared cache directory). The random token keeps concurrent downloads
-    # of the same key on distinct temp files.
+    # (unlike mkstemp, whose open fd is awkward to hand to a downloader and
+    # whose 0o600 mode is too restrictive for a shared cache directory). The
+    # random token keeps concurrent downloads of the same key on distinct
+    # temp files.
     return f"{lpath}.{secrets.token_hex(8)}.part"
 
 
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index 1ea8937ca..9c1cd2c6e 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -1535,6 +1535,52 @@ def test_concurrent_downloads_leave_no_tempfiles(slow_http_server, tmp_path):
     assert len(entries) == 1
 
 
+@pytest.mark.parametrize("failure_mode", ["before_write", "mid_write"])
+def test_failed_download_cleans_up_tempfiles(tmp_path, monkeypatch, failure_mode):
+    # a failing download must leave neither a .part temp file nor the final
+    # cache filename behind, and the next read must succeed; "before_write"
+    # covers cleanup of a temp path that was never created
+    mem = fsspec.filesystem("memory")
+    mem.pipe("/raw/data", b"0123456789")
+    cache_dir = str(tmp_path / "fail")
+    fs = fsspec.filesystem(
+        "simplecache", fs=mem, cache_storage=cache_dir, skip_instance_cache=True
+    )
+
+    def boom(rpath, lpath, **kwargs):
+        if failure_mode == "mid_write":
+            with open(lpath, "wb") as f:
+                f.write(b"0123")
+        raise OSError("simulated download failure")
+
+    with monkeypatch.context() as m:
+        m.setattr(mem, "get_file", boom)
+        with pytest.raises(OSError, match="simulated download failure"):
+            with fs.open("/raw/data", "rb") as f:
+                f.read()
+    assert os.listdir(cache_dir) == []
+
+    with fs.open("/raw/data", "rb") as f:
+        assert f.read() == b"0123456789"
+
+
+def test_stale_part_file_is_ignored(tmp_path):
+    # a *.part file left behind by a hard kill mid-download must not be
+    # mistaken for a cache entry
+    mem = fsspec.filesystem("memory")
+    mem.pipe("/raw/stale", b"real content")
+    cache_dir = tmp_path / "stale"
+    fs = fsspec.filesystem(
+        "simplecache", fs=mem, cache_storage=str(cache_dir), skip_instance_cache=True
+    )
+    sha = fs._mapper("/raw/stale")
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    (cache_dir / f"{sha}.0123456789abcdef.part").write_bytes(b"garbage")
+
+    with fs.open("/raw/stale", "rb") as f:
+        assert f.read() == b"real content"
+
+
 def test_simplecache_cat_ranges_cold_cache(tmp_path):
     # SimpleCacheFileSystem.cat_ranges compared _check_file() results with
     # ``is False``, but _check_file returns None for missing entries, so

From 5b94d782c211e66ec6fdbe6d0d43d2935f6eabde Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Tue, 28 Jul 2026 13:32:36 +0100
Subject: [PATCH 6/9] Record filecache downloads in metadata; tolerate busy
 destination on Windows

Two review comments from Sanjays2402:

- filecache's _cat_file/_cat_ranges downloaded cache misses without
  recording them in CacheMetadata, so every async read of the same key
  was a fresh miss and downloaded the file again; _cat_file also passed
  the (detail, fn) tuple a metadata hit returns straight to open().
  Misses now go through _make_local_details_async (async-safe uid
  lookup) followed by save_cache(), mirroring the sync cat() path;
  simplecache overrides it to keep its no-metadata behavior.

- "last rename wins" does not hold on Windows, where os.replace onto a
  destination another process holds open raises PermissionError. The
  new _replace_tempfile discards the redundant temp copy instead when
  the destination already exists (identical bytes by construction).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 fsspec/implementations/cached.py            | 72 +++++++++++++++---
 fsspec/implementations/tests/test_cached.py | 84 +++++++++++++++++++++
 2 files changed, 147 insertions(+), 9 deletions(-)

diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 43cea7ce7..819f4ce05 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -9,6 +9,7 @@
 import time
 import weakref
 from collections.abc import Callable
+from hashlib import sha256
 from shutil import rmtree
 from typing import TYPE_CHECKING, Any, ClassVar
 
@@ -21,7 +22,7 @@
 from fsspec.implementations.cache_metadata import CacheMetadata
 from fsspec.implementations.chained import ChainedFileSystem
 from fsspec.implementations.local import LocalFileSystem
-from fsspec.spec import AbstractBufferedFile
+from fsspec.spec import AbstractBufferedFile, AbstractFileSystem
 from fsspec.transaction import Transaction
 from fsspec.utils import infer_compression
 
@@ -36,7 +37,9 @@
 # visible under its final cache filename: readers treat the existence of that
 # filename as "download complete" (see issue #639). The rename is atomic, and
 # a concurrent duplicate download of the same path is harmless: both
-# temporary files hold identical bytes and the last rename wins. Temporary
+# temporary files hold identical bytes and the last rename wins (except on
+# Windows, where replacing a file that a reader holds open fails, and the
+# redundant copy is discarded instead — see _replace_tempfile). Temporary
 # files are removed when the download or the final rename raises, but not on
 # early exit (e.g. the process being killed mid-download): stale "*.part"
 # files may then be left in the cache directory, are ignored by the cache,
@@ -63,11 +66,24 @@ def _remove_tempfile(tmp):
         pass
 
 
+def _replace_tempfile(tmp, lpath):
+    try:
+        os.replace(tmp, lpath)
+    except OSError:
+        # On Windows, replacing a file that another process holds open
+        # raises PermissionError. The destination can then only already
+        # exist because a concurrent download of the same path renamed
+        # identical bytes into place, so this copy is redundant.
+        if not os.path.exists(lpath):
+            raise
+        _remove_tempfile(tmp)
+
+
 def _atomic_get_file(fs, rpath, lpath):
     tmp = _temppath(lpath)
     try:
         fs.get_file(rpath, tmp)
-        os.replace(tmp, lpath)
+        _replace_tempfile(tmp, lpath)
     except BaseException:
         _remove_tempfile(tmp)
         raise
@@ -77,7 +93,7 @@ async def _atomic_get_file_async(fs, rpath, lpath, **kwargs):
     tmp = _temppath(lpath)
     try:
         await fs._get_file(rpath, tmp, **kwargs)
-        os.replace(tmp, lpath)
+        _replace_tempfile(tmp, lpath)
     except BaseException:
         _remove_tempfile(tmp)
         raise
@@ -88,7 +104,7 @@ def _atomic_get(fs, rpaths, lpaths):
     try:
         fs.get(rpaths, tmps)
         for tmp, lpath in zip(tmps, lpaths):
-            os.replace(tmp, lpath)
+            _replace_tempfile(tmp, lpath)
     except BaseException:
         for tmp in tmps:
             _remove_tempfile(tmp)
@@ -109,7 +125,7 @@ def _atomic_get_file_decompressed(fs, rpath, lpath, compression, **kwargs):
                 block = getattr(f, "blocksize", 5 * 2**20)
                 data = f.read(block)
                 f2.write(data)
-        os.replace(tmp, lpath)
+        _replace_tempfile(tmp, lpath)
     except BaseException:
         _remove_tempfile(tmp)
         raise
@@ -536,6 +552,8 @@ def __getattribute__(self, item):
             "__getattribute__",
             "__reduce__",
             "_make_local_details",
+            "_make_local_details_async",
+            "_ukey_async",
             "open",
             "cat",
             "cat_file",
@@ -739,6 +757,30 @@ def _make_local_details(self, path):
         logger.debug("Copying %s to local cache", path)
         return fn
 
+    async def _ukey_async(self, path):
+        if type(self.fs).ukey is AbstractFileSystem.ukey:
+            # replicate the default ukey without calling sync code, which
+            # for a target created with asynchronous=True would fail inside
+            # the caller's running event loop
+            return sha256(str(await self.fs._info(path)).encode()).hexdigest()
+        # an overriding ukey (e.g. http's, which does no I/O) is called
+        # directly, as there is no async counterpart to delegate to
+        return self.fs.ukey(path)
+
+    async def _make_local_details_async(self, path):
+        hash = self._mapper(path)
+        fn = os.path.join(self.storage[-1], hash)
+        detail = {
+            "original": path,
+            "fn": hash,
+            "blocks": True,
+            "time": time.time(),
+            "uid": await self._ukey_async(path),
+        }
+        self._metadata.update_file(path, detail)
+        logger.debug("Copying %s to local cache", path)
+        return fn
+
     def cat(
         self,
         path,
@@ -838,12 +880,14 @@ def _open(self, path, mode="rb", **kwargs):
     async def _cat_file(self, path, start=None, end=None, **kwargs):
         logger.debug("async cat_file %s", path)
         path = self._strip_protocol(path)
-        sha = self._mapper(path)
         fn = self._check_file(path)
+        if isinstance(fn, tuple):
+            fn = fn[1]
 
         if not fn:
-            fn = os.path.join(self.storage[-1], sha)
+            fn = await self._make_local_details_async(path)
             await _atomic_get_file_async(self.fs, path, fn, **kwargs)
+            self.save_cache()
 
         with open(fn, "rb") as f:  # noqa ASYNC230
             if start:
@@ -862,7 +906,9 @@ async def _cat_ranges(
             if isinstance(fn, tuple):
                 fn = fn[1]
             if not fn:
-                fn = need.setdefault(p, os.path.join(self.storage[-1], self._mapper(p)))
+                if p not in need:
+                    need[p] = await self._make_local_details_async(p)
+                fn = need[p]
             lpaths.append(fn)
         if need:
             # a batch self.fs._get would forward on_error to the target
@@ -878,6 +924,9 @@ async def _cat_ranges(
                 for res in results:
                     if isinstance(res, BaseException):
                         raise res
+            # metadata entries whose download failed are harmless:
+            # check_file only trusts entries whose file exists
+            self.save_cache()
 
         return LocalFileSystem().cat_ranges(
             lpaths, starts, ends, max_gap=max_gap, on_error=on_error, **kwargs
@@ -927,6 +976,11 @@ def save_cache(self):
     def load_cache(self):
         pass
 
+    async def _make_local_details_async(self, path):
+        # no metadata is kept, so only the local filename is needed; this
+        # also skips the remote ukey call the metadata would require
+        return os.path.join(self.storage[-1], self._mapper(path))
+
     def pipe_file(self, path, value=None, **kwargs):
         if self._intrans:
             with self.open(path, "wb") as f:
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index 9c1cd2c6e..e0d9a4c2e 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -23,6 +23,7 @@
     CachingFileSystem,
     LocalTempFile,
     WholeFileCacheFileSystem,
+    _replace_tempfile,
 )
 from fsspec.implementations.local import make_path_posix
 from fsspec.implementations.zip import ZipFileSystem
@@ -1619,3 +1620,86 @@ async def run():
         return await fs._cat_ranges([url, url], [0, 10], [10, 20])
 
     assert asyncio.run(run()) == [payload[0:10], payload[10:20]]
+
+
+def _async_caching_fs(protocol, cache_dir):
+    return fsspec.filesystem(
+        protocol,
+        target_protocol="http",
+        cache_storage=cache_dir,
+        asynchronous=True,
+        target_options={"asynchronous": True, "skip_instance_cache": True},
+        skip_instance_cache=True,
+    )
+
+
+def _count_downloads(fs):
+    downloads = []
+    inner_get_file = fs.fs._get_file
+
+    async def counting_get_file(rpath, lpath, **kwargs):
+        downloads.append(rpath)
+        return await inner_get_file(rpath, lpath, **kwargs)
+
+    fs.fs._get_file = counting_get_file
+    return downloads
+
+
+@pytest.mark.parametrize("protocol", ["simplecache", "filecache"])
+def test_async_cat_file_downloads_once(slow_http_server, tmp_path, protocol):
+    # filecache's _cat_file did not record the download in its metadata, so
+    # every subsequent _cat_file was a fresh cache miss and downloaded the
+    # file again; and once metadata was recorded (e.g. by cat()), the
+    # (detail, fn) tuple from _check_file was passed straight to open()
+    url, payload = slow_http_server
+
+    async def run():
+        fs = _async_caching_fs(protocol, str(tmp_path / "once"))
+        downloads = _count_downloads(fs)
+        out = [await fs._cat_file(url), await fs._cat_file(url)]
+        return out, len(downloads)
+
+    out, n_downloads = asyncio.run(run())
+    assert out == [payload, payload]
+    assert n_downloads == 1
+
+
+@pytest.mark.parametrize("protocol", ["simplecache", "filecache"])
+def test_async_cat_ranges_downloads_once(slow_http_server, tmp_path, protocol):
+    # same as test_async_cat_file_downloads_once, for _cat_ranges
+    url, payload = slow_http_server
+    url2 = url + "2"
+
+    async def run():
+        fs = _async_caching_fs(protocol, str(tmp_path / "ranges_once"))
+        downloads = _count_downloads(fs)
+        out1 = await fs._cat_ranges([url, url2, url], [0, 5, 10], [10, 15, 20])
+        out2 = await fs._cat_ranges([url, url2], [0, 5], [10, 15])
+        return out1, out2, len(downloads)
+
+    out1, out2, n_downloads = asyncio.run(run())
+    assert out1 == [payload[0:10], payload[5:15], payload[10:20]]
+    assert out2 == [payload[0:10], payload[5:15]]
+    assert n_downloads == 2
+
+
+def test_replace_tempfile_busy_destination(tmp_path, monkeypatch):
+    # on Windows, os.replace onto a destination another process holds open
+    # raises PermissionError; the redundant temp copy must be discarded,
+    # while a failure with no destination in place must still propagate
+    def busy_replace(src, dst):
+        raise PermissionError("destination is open in another process")
+
+    monkeypatch.setattr(os, "replace", busy_replace)
+
+    tmp, dst = tmp_path / "x.abcd1234.part", tmp_path / "x"
+    tmp.write_bytes(b"data")
+    dst.write_bytes(b"data")
+    _replace_tempfile(str(tmp), str(dst))
+    assert not tmp.exists()
+    assert dst.read_bytes() == b"data"
+
+    tmp.write_bytes(b"data")
+    with pytest.raises(PermissionError):
+        _replace_tempfile(str(tmp), str(tmp_path / "missing"))
+    assert tmp.exists()

From aeb6cc3d1d5ff95fa29f56afd2c64f0df46e6825 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Tue, 28 Jul 2026 13:54:00 +0100
Subject: [PATCH 7/9] Fix async cache paths found by review: strip, on_error,
 check_files, stale replace

Follow-up to the multi-agent review of the Sanjays2402 fixes:

- _cat_ranges now strips protocols before metadata lookup/recording, so
  the downloads-once fix works for non-identity _strip_protocol targets
  (s3-style paths) and no longer persists unstripped junk keys.
- The miss-path metadata/ukey lookup moved inside the gathered download
  coroutine, restoring on_error="return" semantics for missing paths.
- New _check_file_async performs the check_files uid comparison and
  expiry check with the async-safe ukey (the sync comparison crashed
  for asynchronous=True targets once entries existed), and treats
  partial block-cache entries as misses instead of serving sparse bytes.
- _replace_tempfile only discards the temp copy when the busy
  destination holds identical bytes; a differing destination (cache
  refresh racing a reader) raises instead of passing stale bytes off
  as fresh.
- compression= set skips async metadata recording so the sync open()
  path can still self-heal with a decompressed download.
- New tests cover non-identity strip + save_cache persistence (fresh
  instance, zero downloads), check_files round-trip, on_error, and the
  differing-bytes replace branch; all verified failing pre-fix by the
  review agents' repros.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 fsspec/implementations/cached.py            |  86 ++++++++++----
 fsspec/implementations/tests/test_cached.py | 120 ++++++++++++++++++--
 2 files changed, 179 insertions(+), 27 deletions(-)

diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 819f4ce05..90e95078e 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import filecmp
 import inspect
 import logging
 import os
@@ -37,9 +38,10 @@
 # visible under its final cache filename: readers treat the existence of that
 # filename as "download complete" (see issue #639). The rename is atomic, and
 # a concurrent duplicate download of the same path is harmless: both
-# temporary files hold identical bytes and the last rename wins (except on
-# Windows, where replacing a file that a reader holds open fails, and the
-# redundant copy is discarded instead — see _replace_tempfile). Temporary
+# temporary files hold identical bytes and the last rename wins (on Windows,
+# where replacing a file that a reader holds open fails, an identical
+# redundant copy is discarded instead, while a differing one — a cache
+# refresh racing a reader — raises; see _replace_tempfile). Temporary
 # files are removed when the download or the final rename raises, but not on
 # early exit (e.g. the process being killed mid-download): stale "*.part"
 # files may then be left in the cache directory, are ignored by the cache,
@@ -71,10 +73,12 @@ def _replace_tempfile(tmp, lpath):
         os.replace(tmp, lpath)
     except OSError:
         # On Windows, replacing a file that another process holds open
-        # raises PermissionError. The destination can then only already
-        # exist because a concurrent download of the same path renamed
-        # identical bytes into place, so this copy is redundant.
-        if not os.path.exists(lpath):
+        # raises PermissionError. If a concurrent download of the same
+        # path already renamed identical bytes into place, this copy is
+        # redundant and safe to discard; a differing destination (e.g. a
+        # cache refresh racing a reader of the stale copy) must surface
+        # the error rather than let stale bytes pass for fresh ones.
+        if not (os.path.exists(lpath) and filecmp.cmp(tmp, lpath, shallow=False)):
             raise
         _remove_tempfile(tmp)
 
@@ -554,6 +558,7 @@ def __getattribute__(self, item):
             "_make_local_details",
             "_make_local_details_async",
             "_ukey_async",
+            "_check_file_async",
             "open",
             "cat",
             "cat_file",
@@ -764,9 +769,33 @@ async def _ukey_async(self, path):
             # the caller's running event loop
             return sha256(str(await self.fs._info(path)).encode()).hexdigest()
         # an overriding ukey (e.g. http's, which does no I/O) is called
-        # directly, as there is no async counterpart to delegate to
+        # directly: there is no async counterpart to delegate to, and an
+        # override that itself performs sync I/O will surface the usual
+        # sync-within-running-loop error
         return self.fs.ukey(path)
 
+    async def _check_file_async(self, path):
+        # _check_file validates check_files with a sync self.fs.ukey call,
+        # which for a target created with asynchronous=True would fail
+        # inside the caller's running event loop; redo the checks here
+        # with the async-safe ukey. Unlike _check_file, a failed check on
+        # one storage location rejects the entry outright instead of
+        # falling through to the next location.
+        self._check_cache()
+        detail = self._metadata.check_file(path, None)
+        if not detail:
+            return False
+        detail, fn = detail
+        if self.check_files and detail["uid"] != await self._ukey_async(path):
+            return False
+        if self.expiry and time.time() - detail["time"] > self.expiry:
+            return False
+        if detail["blocks"] is not True:
+            # a partial (block-cache) entry sharing this cache directory;
+            # re-download the whole file rather than serve sparse bytes
+            return False
+        return detail, fn
+
     async def _make_local_details_async(self, path):
         hash = self._mapper(path)
         fn = os.path.join(self.storage[-1], hash)
@@ -880,14 +909,21 @@ def _open(self, path, mode="rb", **kwargs):
     async def _cat_file(self, path, start=None, end=None, **kwargs):
         logger.debug("async cat_file %s", path)
         path = self._strip_protocol(path)
-        fn = self._check_file(path)
+        fn = await self._check_file_async(path)
         if isinstance(fn, tuple):
             fn = fn[1]
 
         if not fn:
-            fn = await self._make_local_details_async(path)
-            await _atomic_get_file_async(self.fs, path, fn, **kwargs)
-            self.save_cache()
+            if self.compression:
+                # a decompressing download would need a sync self.fs._open;
+                # record no metadata for the raw copy, so the sync open()
+                # path can still download and decompress it properly
+                fn = os.path.join(self.storage[-1], self._mapper(path))
+                await _atomic_get_file_async(self.fs, path, fn, **kwargs)
+            else:
+                fn = await self._make_local_details_async(path)
+                await _atomic_get_file_async(self.fs, path, fn, **kwargs)
+                self.save_cache()
 
         with open(fn, "rb") as f:  # noqa ASYNC230
             if start:
@@ -902,22 +938,27 @@ async def _cat_ranges(
         lpaths = []
         need = {}
         for p in paths:
-            fn = self._check_file(p)
+            p = self._strip_protocol(p)
+            fn = await self._check_file_async(p)
             if isinstance(fn, tuple):
                 fn = fn[1]
             if not fn:
-                if p not in need:
-                    need[p] = await self._make_local_details_async(p)
-                fn = need[p]
+                fn = need.setdefault(p, os.path.join(self.storage[-1], self._mapper(p)))
             lpaths.append(fn)
         if need:
+
+            async def _get_one(rpath, lpath):
+                if not self.compression:
+                    # recorded inside the gather so a failure (e.g. the
+                    # ukey lookup on a missing path) honors on_error
+                    # instead of escaping the whole call
+                    await self._make_local_details_async(rpath)
+                await _atomic_get_file_async(self.fs, rpath, lpath)
+
             # a batch self.fs._get would forward on_error to the target
             # filesystem's _get_file, which does not accept it
             results = await asyncio.gather(
-                *(
-                    _atomic_get_file_async(self.fs, rpath, lpath)
-                    for rpath, lpath in need.items()
-                ),
+                *(_get_one(rpath, lpath) for rpath, lpath in need.items()),
                 return_exceptions=True,
             )
             if on_error == "raise":
@@ -981,6 +1022,11 @@ async def _make_local_details_async(self, path):
         # also skips the remote ukey call the metadata would require
         return os.path.join(self.storage[-1], self._mapper(path))
 
+    async def _check_file_async(self, path):
+        # existence-based and performs no remote I/O, so the sync version
+        # is safe inside a running event loop
+        return self._check_file(path)
+
     def pipe_file(self, path, value=None, **kwargs):
         if self._intrans:
             with self.open(path, "wb") as f:
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index e0d9a4c2e..9c3b508af 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -14,6 +14,7 @@
 import fsspec
 from fsspec.compression import compr
 from fsspec.exceptions import BlocksizeMismatchError
+from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
 from fsspec.implementations.cache_mapper import (
     BasenameCacheMapper,
     HashCacheMapper,
@@ -26,6 +27,7 @@
     _replace_tempfile,
 )
 from fsspec.implementations.local import make_path_posix
+from fsspec.implementations.memory import MemoryFileSystem
 from fsspec.implementations.zip import ZipFileSystem
 from fsspec.tests.conftest import win
 
@@ -1650,7 +1652,9 @@ def test_async_cat_file_downloads_once(slow_http_server, tmp_path, protocol):
     # filecache's _cat_file did not record the download in its metadata, so
     # every subsequent _cat_file was a fresh cache miss and downloaded the
     # file again; and once metadata was recorded (e.g. by cat()), the
-    # (detail, fn) tuple from _check_file was passed straight to open()
+    # (detail, fn) tuple from _check_file was passed straight to open().
+    # simplecache never had the bug (existence-based _check_file) and is
+    # included as a regression guard.
     url, payload = slow_http_server
 
     async def run():
@@ -1685,21 +1689,123 @@ async def run():
 
 def test_replace_tempfile_busy_destination(tmp_path, monkeypatch):
     # on Windows, os.replace onto a destination another process holds open
-    # raises PermissionError; the redundant temp copy must be discarded,
-    # while a failure with no destination in place must still propagate
+    # raises PermissionError; an identical redundant temp copy must be
+    # discarded, while a differing destination (cache refresh racing a
+    # reader of the stale copy) or an absent one must still propagate
     def busy_replace(src, dst):
         raise PermissionError("destination is open in another process")
 
     monkeypatch.setattr(os, "replace", busy_replace)
 
     tmp, dst = tmp_path / "x.abcd1234.part", tmp_path / "x"
-    tmp.write_bytes(b"data")
-    dst.write_bytes(b"data")
+    tmp.write_bytes(b"same bytes")
+    dst.write_bytes(b"same bytes")
     _replace_tempfile(str(tmp), str(dst))
     assert not tmp.exists()
-    assert dst.read_bytes() == b"data"
+    assert dst.read_bytes() == b"same bytes"
+
+    tmp.write_bytes(b"fresh bytes")
+    dst.write_bytes(b"stale bytes")
+    with pytest.raises(PermissionError):
+        _replace_tempfile(str(tmp), str(dst))
+    assert tmp.exists()
+    assert dst.read_bytes() == b"stale bytes"
 
-    tmp.write_bytes(b"data")
     with pytest.raises(PermissionError):
         _replace_tempfile(str(tmp), str(tmp_path / "missing"))
     assert tmp.exists()
+
+
+class _AsyncMemoryFileSystem(AsyncFileSystemWrapper):
+    # gives the wrapper memory's protocol handling, so _strip_protocol is
+    # NOT the identity ("memory://afile" -> "/afile") — unlike http's —
+    # and the default AbstractFileSystem.ukey (no override) is exercised
+    protocol = "memory"
+    _strip_protocol = MemoryFileSystem._strip_protocol
+
+
+def _async_mem_caching_fs(cache_dir, **kwargs):
+    return fsspec.filesystem(
+        "filecache",
+        fs=_AsyncMemoryFileSystem(fs=fsspec.filesystem("memory")),
+        cache_storage=cache_dir,
+        skip_instance_cache=True,
+        **kwargs,
+    )
+
+
+def test_filecache_async_nonidentity_strip_and_persistence(tmp_path):
+    # exercises what the http-based tests cannot: a target whose
+    # _strip_protocol is not the identity (metadata must be recorded under
+    # the stripped path or every read is a miss), the default-ukey branch
+    # of _ukey_async, and persistence via save_cache (a fresh instance
+    # must hit the cache without downloading)
+    mem = fsspec.filesystem("memory")
+    mem.pipe("/afile", b"0123456789")
+    cache_dir = str(tmp_path / "strip")
+
+    async def run():
+        fs = _async_mem_caching_fs(cache_dir)
+        downloads = _count_downloads(fs)
+        out1 = await fs._cat_ranges(
+            ["memory://afile", "memory://afile"], [0, 4], [4, 8]
+        )
+        out2 = await fs._cat_file("memory://afile")
+        n_first = len(downloads)
+
+        fs2 = _async_mem_caching_fs(cache_dir)
+        downloads2 = _count_downloads(fs2)
+        out3 = await fs2._cat_file("memory://afile")
+        return out1, out2, n_first, out3, len(downloads2)
+
+    out1, out2, n_first, out3, n_second = asyncio.run(run())
+    assert out1 == [b"0123", b"4567"]
+    assert out2 == b"0123456789" == out3
+    assert n_first == 1
+    assert n_second == 0
+
+
+def test_filecache_async_check_files(tmp_path):
+    # check_files=True must work on the async paths: the uid recorded by
+    # _make_local_details_async has to round-trip through
+    # _check_file_async's async-safe comparison (a sync ukey call here
+    # would fail inside the running loop), and a remote change must
+    # trigger a re-download
+    mem = fsspec.filesystem("memory")
+    mem.pipe("/cfile", b"version one")
+
+    async def run():
+        fs = _async_mem_caching_fs(str(tmp_path / "cf"), check_files=True)
+        downloads = _count_downloads(fs)
+        first = await fs._cat_file("memory://cfile")
+        again = await fs._cat_file("memory://cfile")
+        mem.pipe("/cfile", b"version two, longer")
+        refreshed = await fs._cat_file("memory://cfile")
+        return first, again, refreshed, len(downloads)
+
+    first, again, refreshed, n_downloads = asyncio.run(run())
+    assert (first, again) == (b"version one", b"version one")
+    assert refreshed == b"version two, longer"
+    assert n_downloads == 2
+
+
+def test_filecache_async_cat_ranges_on_error(tmp_path):
+    # a missing remote path must be reported per-range with
+    # on_error="return" (the default), not raised out of the whole call:
+    # the metadata/ukey lookup for a miss runs inside the gathered
+    # download so its failure is captured like a download failure
+    mem = fsspec.filesystem("memory")
+    mem.pipe("/exists", b"0123456789")
+
+    async def run():
+        fs = _async_mem_caching_fs(str(tmp_path / "oe"))
+        returned = await fs._cat_ranges(
+            ["memory://exists", "memory://missing"], [0, 0], [4, 4], on_error="return"
+        )
+        with pytest.raises(FileNotFoundError):
+            await fs._cat_ranges(["memory://missing"], [0], [4], on_error="raise")
+        return returned
+
+    returned = asyncio.run(run())
+    assert returned[0] == b"0123"
+    assert isinstance(returned[1], Exception)

From 4e6817bbf8a86e6babf45e2de3fd405a2f1a2e08 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Tue, 28 Jul 2026 14:08:15 +0100
Subject: [PATCH 8/9] Simplify atomic-write helpers and async cache paths

Behavior-preserving cleanups from the simplification review: a single
_tempfile context manager replaces the temp/replace/cleanup dance
repeated across the four _atomic_* helpers; _make_local_details_async
delegates to _make_local_details via an optional uid argument;
_check_file_async returns only the local filename (neither caller used
the detail); one _fetch_file_async helper carries the compression/
metadata guard for both _cat_file and _cat_ranges; tests reuse one
fs-constructor helper and the leave-no-tempfiles assertion folded into
the concurrency test, where it now covers both protocols and all
trials.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 fsspec/implementations/cached.py            | 139 +++++++++-----------
 fsspec/implementations/tests/test_cached.py |  54 +++-----
 2 files changed, 77 insertions(+), 116 deletions(-)

diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 90e95078e..380b81753 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -10,6 +10,7 @@
 import time
 import weakref
 from collections.abc import Callable
+from contextlib import ExitStack, contextmanager
 from hashlib import sha256
 from shutil import rmtree
 from typing import TYPE_CHECKING, Any, ClassVar
@@ -38,14 +39,12 @@
 # visible under its final cache filename: readers treat the existence of that
 # filename as "download complete" (see issue #639). The rename is atomic, and
 # a concurrent duplicate download of the same path is harmless: both
-# temporary files hold identical bytes and the last rename wins (on Windows,
-# where replacing a file that a reader holds open fails, an identical
-# redundant copy is discarded instead, while a differing one — a cache
-# refresh racing a reader — raises; see _replace_tempfile). Temporary
-# files are removed when the download or the final rename raises, but not on
-# early exit (e.g. the process being killed mid-download): stale "*.part"
-# files may then be left in the cache directory, are ignored by the cache,
-# and are safe to delete.
+# temporary files hold identical bytes and the last rename wins (or, on
+# Windows, a redundant identical copy is discarded; see _replace_tempfile).
+# Temporary files are removed when the download or the final rename raises,
+# but not on early exit (e.g. the process being killed mid-download): stale
+# "*.part" files may then be left in the cache directory, are ignored by the
+# cache, and are safe to delete.
 # These helpers are module-level functions rather than methods because
 # CachingFileSystem.__getattribute__ dispatches only known method names to
 # the caching class, delegating anything else to the wrapped filesystem.
@@ -83,56 +82,60 @@ def _replace_tempfile(tmp, lpath):
         _remove_tempfile(tmp)
 
 
-def _atomic_get_file(fs, rpath, lpath):
+@contextmanager
+def _tempfile(lpath):
+    # yield a temp name to download into; rename it to lpath on success,
+    # remove it on error
     tmp = _temppath(lpath)
     try:
-        fs.get_file(rpath, tmp)
+        yield tmp
         _replace_tempfile(tmp, lpath)
     except BaseException:
         _remove_tempfile(tmp)
         raise
 
 
+def _atomic_get_file(fs, rpath, lpath):
+    with _tempfile(lpath) as tmp:
+        fs.get_file(rpath, tmp)
+
+
 async def _atomic_get_file_async(fs, rpath, lpath, **kwargs):
-    tmp = _temppath(lpath)
-    try:
+    with _tempfile(lpath) as tmp:
         await fs._get_file(rpath, tmp, **kwargs)
-        _replace_tempfile(tmp, lpath)
-    except BaseException:
-        _remove_tempfile(tmp)
-        raise
+
+
+async def _fetch_file_async(cfs, rpath, lpath, **kwargs):
+    # a decompressing download would need a sync cfs.fs._open; record no
+    # metadata for the raw copy, so the sync open() path can still
+    # download and decompress it properly
+    if not cfs.compression:
+        await cfs._make_local_details_async(rpath)
+    await _atomic_get_file_async(cfs.fs, rpath, lpath, **kwargs)
 
 
 def _atomic_get(fs, rpaths, lpaths):
-    tmps = [_temppath(lpath) for lpath in lpaths]
-    try:
+    with ExitStack() as stack:
+        tmps = [stack.enter_context(_tempfile(lpath)) for lpath in lpaths]
         fs.get(rpaths, tmps)
-        for tmp, lpath in zip(tmps, lpaths):
-            _replace_tempfile(tmp, lpath)
-    except BaseException:
-        for tmp in tmps:
-            _remove_tempfile(tmp)
-        raise
 
 
 def _atomic_get_file_decompressed(fs, rpath, lpath, compression, **kwargs):
-    tmp = _temppath(lpath)
-    try:
-        with fs._open(rpath, mode="rb", **kwargs) as f, open(tmp, "wb") as f2:
-            if isinstance(f, AbstractBufferedFile):
-                # want no type of caching if just downloading whole thing
-                f.cache = BaseCache(0, f.cache.fetcher, f.size)
-            comp = infer_compression(rpath) if compression == "infer" else compression
-            f = compr[comp](f, mode="rb")
-            data = True
-            while data:
-                block = getattr(f, "blocksize", 5 * 2**20)
-                data = f.read(block)
-                f2.write(data)
-        _replace_tempfile(tmp, lpath)
-    except BaseException:
-        _remove_tempfile(tmp)
-        raise
+    with (
+        _tempfile(lpath) as tmp,
+        fs._open(rpath, mode="rb", **kwargs) as f,
+        open(tmp, "wb") as f2,
+    ):
+        if isinstance(f, AbstractBufferedFile):
+            # want no type of caching if just downloading whole thing
+            f.cache = BaseCache(0, f.cache.fetcher, f.size)
+        comp = infer_compression(rpath) if compression == "infer" else compression
+        f = compr[comp](f, mode="rb")
+        data = True
+        while data:
+            block = getattr(f, "blocksize", 5 * 2**20)
+            data = f.read(block)
+            f2.write(data)
 
 
 class WriteCachedTransaction(Transaction):
@@ -748,7 +751,7 @@ def commit_many(self, open_files):
                 pass
         self._cache_size = None
 
-    def _make_local_details(self, path):
+    def _make_local_details(self, path, uid=None):
         hash = self._mapper(path)
         fn = os.path.join(self.storage[-1], hash)
         detail = {
@@ -756,7 +759,7 @@ def _make_local_details(self, path):
             "fn": hash,
             "blocks": True,
             "time": time.time(),
-            "uid": self.fs.ukey(path),
+            "uid": self.fs.ukey(path) if uid is None else uid,
         }
         self._metadata.update_file(path, detail)
         logger.debug("Copying %s to local cache", path)
@@ -794,21 +797,10 @@ async def _check_file_async(self, path):
             # a partial (block-cache) entry sharing this cache directory;
             # re-download the whole file rather than serve sparse bytes
             return False
-        return detail, fn
+        return fn
 
     async def _make_local_details_async(self, path):
-        hash = self._mapper(path)
-        fn = os.path.join(self.storage[-1], hash)
-        detail = {
-            "original": path,
-            "fn": hash,
-            "blocks": True,
-            "time": time.time(),
-            "uid": await self._ukey_async(path),
-        }
-        self._metadata.update_file(path, detail)
-        logger.debug("Copying %s to local cache", path)
-        return fn
+        return self._make_local_details(path, uid=await self._ukey_async(path))
 
     def cat(
         self,
@@ -910,19 +902,11 @@ async def _cat_file(self, path, start=None, end=None, **kwargs):
         logger.debug("async cat_file %s", path)
         path = self._strip_protocol(path)
         fn = await self._check_file_async(path)
-        if isinstance(fn, tuple):
-            fn = fn[1]
 
         if not fn:
-            if self.compression:
-                # a decompressing download would need a sync self.fs._open;
-                # record no metadata for the raw copy, so the sync open()
-                # path can still download and decompress it properly
-                fn = os.path.join(self.storage[-1], self._mapper(path))
-                await _atomic_get_file_async(self.fs, path, fn, **kwargs)
-            else:
-                fn = await self._make_local_details_async(path)
-                await _atomic_get_file_async(self.fs, path, fn, **kwargs)
+            fn = os.path.join(self.storage[-1], self._mapper(path))
+            await _fetch_file_async(self, path, fn, **kwargs)
+            if not self.compression:
                 self.save_cache()
 
         with open(fn, "rb") as f:  # noqa ASYNC230
@@ -940,25 +924,20 @@ async def _cat_ranges(
         for p in paths:
             p = self._strip_protocol(p)
             fn = await self._check_file_async(p)
-            if isinstance(fn, tuple):
-                fn = fn[1]
             if not fn:
                 fn = need.setdefault(p, os.path.join(self.storage[-1], self._mapper(p)))
             lpaths.append(fn)
         if need:
-
-            async def _get_one(rpath, lpath):
-                if not self.compression:
-                    # recorded inside the gather so a failure (e.g. the
-                    # ukey lookup on a missing path) honors on_error
-                    # instead of escaping the whole call
-                    await self._make_local_details_async(rpath)
-                await _atomic_get_file_async(self.fs, rpath, lpath)
-
             # a batch self.fs._get would forward on_error to the target
-            # filesystem's _get_file, which does not accept it
+            # filesystem's _get_file, which does not accept it; metadata is
+            # recorded inside the gather so a failure (e.g. the ukey lookup
+            # on a missing path) honors on_error instead of escaping the
+            # whole call
             results = await asyncio.gather(
-                *(_get_one(rpath, lpath) for rpath, lpath in need.items()),
+                *(
+                    _fetch_file_async(self, rpath, lpath)
+                    for rpath, lpath in need.items()
+                ),
                 return_exceptions=True,
             )
             if on_error == "raise":
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index 9c3b508af..55a7e7f56 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -1460,19 +1460,23 @@ def handle_error(self, request, client_address):
     server.shutdown()
 
 
+def _async_caching_fs(protocol, cache_dir):
+    return fsspec.filesystem(
+        protocol,
+        target_protocol="http",
+        cache_storage=cache_dir,
+        asynchronous=True,
+        target_options={"asynchronous": True, "skip_instance_cache": True},
+        skip_instance_cache=True,
+    )
+
+
 def _staggered_async_cat_file(protocol, url, cache_dir, n):
     # readers must start while a download is still in flight: simultaneous
     # starts all miss the cache and write identical bytes to identical
     # offsets, which would hide the race
     async def run():
-        fs = fsspec.filesystem(
-            protocol,
-            target_protocol="http",
-            cache_storage=cache_dir,
-            asynchronous=True,
-            target_options={"asynchronous": True, "skip_instance_cache": True},
-            skip_instance_cache=True,
-        )
+        fs = _async_caching_fs(protocol, cache_dir)
 
         async def one(i):
             await asyncio.sleep(i * 0.03)
@@ -1516,6 +1520,11 @@ def test_concurrent_cat_file_async(slow_http_server, tmp_path, protocol):
         )
         assert [len(d) for d in data] == [len(payload)] * 8
         assert all(d == payload for d in data)
+        # no leftover .part temp files; only the payload (plus filecache's
+        # "cache" metadata file) remains
+        entries = os.listdir(str(tmp_path / f"async{trial}"))
+        assert [fn for fn in entries if fn.endswith(".part")] == []
+        assert len(entries) == (1 if protocol == "simplecache" else 2)
 
 
 def test_concurrent_cat_file_threads(slow_http_server, tmp_path):
@@ -1529,15 +1538,6 @@ def test_concurrent_cat_file_threads(slow_http_server, tmp_path):
         assert all(d == payload for d in data)
 
 
-def test_concurrent_downloads_leave_no_tempfiles(slow_http_server, tmp_path):
-    url, payload = slow_http_server
-    cache_dir = str(tmp_path / "clean")
-    _staggered_async_cat_file("simplecache", url, cache_dir, 4)
-    entries = os.listdir(cache_dir)
-    assert [fn for fn in entries if fn.endswith(".part")] == []
-    assert len(entries) == 1
-
-
 @pytest.mark.parametrize("failure_mode", ["before_write", "mid_write"])
 def test_failed_download_cleans_up_tempfiles(tmp_path, monkeypatch, failure_mode):
     # a failing download must leave neither a .part temp file nor the final
@@ -1611,30 +1611,12 @@ def test_async_cat_ranges_cold_cache(slow_http_server, tmp_path, protocol):
     url, payload = slow_http_server
 
     async def run():
-        fs = fsspec.filesystem(
-            protocol,
-            target_protocol="http",
-            cache_storage=str(tmp_path / "cr"),
-            asynchronous=True,
-            target_options={"asynchronous": True, "skip_instance_cache": True},
-            skip_instance_cache=True,
-        )
+        fs = _async_caching_fs(protocol, str(tmp_path / "cr"))
         return await fs._cat_ranges([url, url], [0, 10], [10, 20])
 
     assert asyncio.run(run()) == [payload[0:10], payload[10:20]]
 
 
-def _async_caching_fs(protocol, cache_dir):
-    return fsspec.filesystem(
-        protocol,
-        target_protocol="http",
-        cache_storage=cache_dir,
-        asynchronous=True,
-        target_options={"asynchronous": True, "skip_instance_cache": True},
-        skip_instance_cache=True,
-    )
-
-
 def _count_downloads(fs):
     downloads = []
     inner_get_file = fs.fs._get_file

From 2b9ec831475d118914fbc28370354fbde6e934c6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Lo=C3=AFc=20Houpert?=
 <10154151+lhoupert@users.noreply.github.com>
Date: Tue, 28 Jul 2026 15:00:00 +0100
Subject: [PATCH 9/9] Replace content comparison with stat checks in
 _replace_tempfile

Per review: filecmp re-read both files in full on every swallowed
rename failure. Only complete downloads are ever renamed onto the
final cache name, so stat checks suffice: the temp copy is discarded
only when the busy destination is a regular file of the same size
whose mtime is at or after the download start. The start stamp is the
pre-created temp file's own mtime, so both timestamps come from the
cache filesystem's clock (no wall-clock/filesystem skew), and only
PermissionError is swallowed. An older destination (a cache refresh
racing a reader of the stale copy), a size mismatch, or a non-file
still raises rather than passing stale bytes off as fresh.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 fsspec/implementations/cached.py            | 60 ++++++++++++++-------
 fsspec/implementations/tests/test_cached.py | 40 ++++++++++----
 2 files changed, 72 insertions(+), 28 deletions(-)

diff --git a/fsspec/implementations/cached.py b/fsspec/implementations/cached.py
index 380b81753..295d13ce5 100644
--- a/fsspec/implementations/cached.py
+++ b/fsspec/implementations/cached.py
@@ -1,11 +1,11 @@
 from __future__ import annotations
 
 import asyncio
-import filecmp
 import inspect
 import logging
 import os
 import secrets
+import stat
 import tempfile
 import time
 import weakref
@@ -40,7 +40,7 @@
 # filename as "download complete" (see issue #639). The rename is atomic, and
 # a concurrent duplicate download of the same path is harmless: both
 # temporary files hold identical bytes and the last rename wins (or, on
-# Windows, a redundant identical copy is discarded; see _replace_tempfile).
+# Windows, a redundant concurrent copy is discarded; see _replace_tempfile).
 # Temporary files are removed when the download or the final rename raises,
 # but not on early exit (e.g. the process being killed mid-download): stale
 # "*.part" files may then be left in the cache directory, are ignored by the
@@ -51,12 +51,12 @@
 
 
 def _temppath(lpath):
-    # Only generates a name; the file itself is created by whatever plain
-    # open() downloads it, so it gets ordinary umask-derived permissions
-    # (unlike mkstemp, whose open fd is awkward to hand to a downloader and
-    # whose 0o600 mode is too restrictive for a shared cache directory). The
-    # random token keeps concurrent downloads of the same key on distinct
-    # temp files.
+    # Only generates a name; the file is created by plain open() calls
+    # (pre-created empty in _tempfile, then written by the downloader), so
+    # it gets ordinary umask-derived permissions (unlike mkstemp, whose
+    # open fd is awkward to hand to a downloader and whose 0o600 mode is
+    # too restrictive for a shared cache directory). The random token
+    # keeps concurrent downloads of the same key on distinct temp files.
     return f"{lpath}.{secrets.token_hex(8)}.part"
 
 
@@ -67,17 +67,35 @@ def _remove_tempfile(tmp):
         pass
 
 
-def _replace_tempfile(tmp, lpath):
+def _replace_tempfile(tmp, lpath, start):
     try:
         os.replace(tmp, lpath)
-    except OSError:
+    except PermissionError:
         # On Windows, replacing a file that another process holds open
-        # raises PermissionError. If a concurrent download of the same
-        # path already renamed identical bytes into place, this copy is
-        # redundant and safe to discard; a differing destination (e.g. a
-        # cache refresh racing a reader of the stale copy) must surface
-        # the error rather than let stale bytes pass for fresh ones.
-        if not (os.path.exists(lpath) and filecmp.cmp(tmp, lpath, shallow=False)):
+        # raises PermissionError. Only complete downloads are ever renamed
+        # onto the final name, so a regular file of the same size written
+        # there after this download started can only be a complete, recent
+        # download of the same remote path, making this copy redundant.
+        # Anything else — notably an older destination, i.e. a cache
+        # refresh racing a reader of the stale copy — re-raises rather
+        # than let stale bytes pass for fresh ones. `start` is the mtime
+        # of the pre-created temp file, so both timestamps come from the
+        # cache filesystem's clock (in-tree get_file implementations all
+        # write the local copy in place, so its mtime is the local write
+        # time, not the remote object's). The check is deliberately
+        # narrower than a byte comparison: a duplicate whose writes all
+        # landed before this download began raises a loud spurious error
+        # instead of being discarded.
+        try:
+            st_dst, st_tmp = os.stat(lpath), os.stat(tmp)
+            redundant = (
+                stat.S_ISREG(st_dst.st_mode)
+                and st_dst.st_size == st_tmp.st_size
+                and st_dst.st_mtime >= start
+            )
+        except OSError:
+            redundant = False
+        if not redundant:
             raise
         _remove_tempfile(tmp)
 
@@ -85,11 +103,17 @@ def _replace_tempfile(tmp, lpath):
 @contextmanager
 def _tempfile(lpath):
     # yield a temp name to download into; rename it to lpath on success,
-    # remove it on error
+    # remove it on error. The temp file is pre-created empty (with a plain
+    # open(), keeping umask-derived permissions) so the download start is
+    # stamped on the cache filesystem's own clock, comparable with the
+    # destination's mtime in _replace_tempfile.
     tmp = _temppath(lpath)
     try:
+        with open(tmp, "wb"):
+            pass
+        start = os.stat(tmp).st_mtime
         yield tmp
-        _replace_tempfile(tmp, lpath)
+        _replace_tempfile(tmp, lpath, start)
     except BaseException:
         _remove_tempfile(tmp)
         raise
diff --git a/fsspec/implementations/tests/test_cached.py b/fsspec/implementations/tests/test_cached.py
index 55a7e7f56..078a77376 100644
--- a/fsspec/implementations/tests/test_cached.py
+++ b/fsspec/implementations/tests/test_cached.py
@@ -1542,7 +1542,7 @@ def test_concurrent_cat_file_threads(slow_http_server, tmp_path):
 def test_failed_download_cleans_up_tempfiles(tmp_path, monkeypatch, failure_mode):
     # a failing download must leave neither a .part temp file nor the final
     # cache filename behind, and the next read must succeed; "before_write"
-    # covers cleanup of a temp path that was never created
+    # fails before the downloader itself writes anything to the temp path
     mem = fsspec.filesystem("memory")
     mem.pipe("/raw/data", b"0123456789")
     cache_dir = str(tmp_path / "fail")
@@ -1671,30 +1671,50 @@ async def run():
 
 def test_replace_tempfile_busy_destination(tmp_path, monkeypatch):
     # on Windows, os.replace onto a destination another process holds open
-    # raises PermissionError; an identical redundant temp copy must be
-    # discarded, while a differing destination (cache refresh racing a
-    # reader of the stale copy) or an absent one must still propagate
+    # raises PermissionError; a same-size destination written at or after
+    # our download start is a redundant concurrent download and the temp
+    # copy is discarded, while an older destination (cache refresh racing
+    # a reader of the stale copy), a different size, a non-file, or an
+    # absent one must propagate
     def busy_replace(src, dst):
         raise PermissionError("destination is open in another process")
 
     monkeypatch.setattr(os, "replace", busy_replace)
 
     tmp, dst = tmp_path / "x.abcd1234.part", tmp_path / "x"
-    tmp.write_bytes(b"same bytes")
-    dst.write_bytes(b"same bytes")
-    _replace_tempfile(str(tmp), str(dst))
+    tmp.write_bytes(b"payload")
+    dst.write_bytes(b"payload")
+    _replace_tempfile(str(tmp), str(dst), start=dst.stat().st_mtime - 1)
+    assert not tmp.exists()
+    assert dst.read_bytes() == b"payload"
+
+    # the tie discards too: on coarse-granularity filesystems a concurrent
+    # duplicate's mtime commonly equals the captured start exactly
+    tmp.write_bytes(b"payload")
+    _replace_tempfile(str(tmp), str(dst), start=dst.stat().st_mtime)
     assert not tmp.exists()
-    assert dst.read_bytes() == b"same bytes"
 
     tmp.write_bytes(b"fresh bytes")
     dst.write_bytes(b"stale bytes")
     with pytest.raises(PermissionError):
-        _replace_tempfile(str(tmp), str(dst))
+        _replace_tempfile(str(tmp), str(dst), start=dst.stat().st_mtime + 1)
     assert tmp.exists()
     assert dst.read_bytes() == b"stale bytes"
 
+    tmp.write_bytes(b"short")
+    dst.write_bytes(b"much longer content")
+    with pytest.raises(PermissionError):
+        _replace_tempfile(str(tmp), str(dst), start=dst.stat().st_mtime - 1)
+    assert tmp.exists()
+
+    dst_dir = tmp_path / "adir"
+    dst_dir.mkdir()
+    with pytest.raises(PermissionError):
+        _replace_tempfile(str(tmp), str(dst_dir), start=0.0)
+    assert tmp.exists()
+
     with pytest.raises(PermissionError):
-        _replace_tempfile(str(tmp), str(tmp_path / "missing"))
+        _replace_tempfile(str(tmp), str(tmp_path / "missing"), start=0.0)
     assert tmp.exists()