diff --git a/docs/source/features.rst b/docs/source/features.rst index 18f1e80e6..c6f335cc4 100644 --- a/docs/source/features.rst +++ b/docs/source/features.rst @@ -290,6 +290,14 @@ 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 (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 08b4ad579..295d13ce5 100644 --- a/fsspec/implementations/cached.py +++ b/fsspec/implementations/cached.py @@ -1,12 +1,17 @@ from __future__ import annotations +import asyncio import inspect import logging import os +import secrets +import stat import tempfile 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 @@ -19,7 +24,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 @@ -29,6 +34,134 @@ 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 (or, on +# 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 +# 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): + # 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" + + +def _remove_tempfile(tmp): + try: + os.remove(tmp) + except OSError: + pass + + +def _replace_tempfile(tmp, lpath, start): + try: + os.replace(tmp, lpath) + except PermissionError: + # On Windows, replacing a file that another process holds open + # 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) + + +@contextmanager +def _tempfile(lpath): + # yield a temp name to download into; rename it to lpath on success, + # 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, start) + 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): + with _tempfile(lpath) as tmp: + await fs._get_file(rpath, tmp, **kwargs) + + +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): + with ExitStack() as stack: + tmps = [stack.enter_context(_tempfile(lpath)) for lpath in lpaths] + fs.get(rpaths, tmps) + + +def _atomic_get_file_decompressed(fs, rpath, lpath, compression, **kwargs): + 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): def complete(self, commit=True): rpaths = [f.path for f in self.files] @@ -450,6 +583,9 @@ def __getattribute__(self, item): "__getattribute__", "__reduce__", "_make_local_details", + "_make_local_details_async", + "_ukey_async", + "_check_file_async", "open", "cat", "cat_file", @@ -602,7 +738,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 = [ @@ -639,7 +775,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 = { @@ -647,12 +783,49 @@ 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) 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: 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 fn + + async def _make_local_details_async(self, path): + return self._make_local_details(path, uid=await self._ukey_async(path)) + def cat( self, path, @@ -686,7 +859,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)) @@ -703,23 +876,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): @@ -766,12 +925,13 @@ 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) + fn = await self._check_file_async(path) if not fn: - fn = os.path.join(self.storage[-1], sha) - await self.fs._get_file(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 if start: @@ -784,20 +944,33 @@ 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) + p = self._strip_protocol(p) + fn = await self._check_file_async(p) + 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; 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( + *( + _fetch_file_async(self, 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 + # 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 @@ -847,6 +1020,16 @@ 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)) + + 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: @@ -912,12 +1095,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: + _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( - 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): @@ -929,23 +1116,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 794eea061..078a77376 100644 --- a/fsspec/implementations/tests/test_cached.py +++ b/fsspec/implementations/tests/test_cached.py @@ -1,13 +1,20 @@ +import asyncio import json import os +import random import shutil import tempfile +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import pytest 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, @@ -17,8 +24,10 @@ CachingFileSystem, LocalTempFile, WholeFileCacheFileSystem, + _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 @@ -1405,3 +1414,400 @@ 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 _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 = _async_caching_fs(protocol, cache_dir) + + 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) + # 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): + """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) + + +@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" + # 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") + 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 + # 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 = _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 _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(). + # simplecache never had the bug (existence-based _check_file) and is + # included as a regression guard. + 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; 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"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() + + tmp.write_bytes(b"fresh bytes") + dst.write_bytes(b"stale bytes") + with pytest.raises(PermissionError): + _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"), start=0.0) + 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)