Repository: osbuild/osbuild
Pull request #1212 (2022-12-20 13:30:12)

util/fscache: add cachedir-tag support
The cachedir-tag specification defines how to mark directories as cache-directories. This allows tools like `tar` to ignore those directories if desired (e.g., see `tar --ignore-caches`). This is very useful to avoid huge cache-directories in backups and remote synchronizations.

The spec simply defines a file called `CACHEDIR.TAG` with the first 43 bytes to be: "Signature: 8a477f597d28d172789f06886806bc55" (which happens to be the MD5-checksum of ".IsCacheDirectory". Further content is to be ignored. Any such files marks the directory in question as a cache-directory.

The cachedir-tag has been successfully deployed in tools like `cargo` and `VLC`, and is currently discussed to be implemented in Firefox. More information is available here: https://bford.info/cachedir/


Patch:
diff --git a/osbuild/objectstore.py b/osbuild/objectstore.py
index 9f1dcef4..d8f2cba8 100644
--- a/osbuild/objectstore.py
+++ b/osbuild/objectstore.py
@@ -171,13 +171,27 @@ class Object:
         self._check_mode(Object.Mode.WRITE)
         assert self.active
         assert self._path
-        base.clone(self._path)
+
+        subprocess.run(
+            [
+                "cp",
+                "--reflink=auto",
+                "-a",
+                os.fspath(base.path) + "/.",
+                os.fspath(self.path),
+            ],
+            check=True,
+        )
 
     @property
-    def tree(self) -> str:
+    def path(self) -> str:
         assert self.active
         assert self._path
-        return os.path.join(self._path, "tree")
+        return self._path
+
+    @property
+    def tree(self) -> str:
+        return os.path.join(self.path, "tree")
 
     @property
     def meta(self) -> Metadata:
@@ -243,22 +257,6 @@ class Object:
             check=True,
         )
 
-    def clone(self, to_directory: PathLike):
-        """Clone the object to the specified directory"""
-
-        assert self._path
-
-        subprocess.run(
-            [
-                "cp",
-                "--reflink=auto",
-                "-a",
-                os.fspath(self._path) + "/.",
-                os.fspath(to_directory),
-            ],
-            check=True,
-        )
-
     def __fspath__(self):
         return self.tree
 
@@ -421,13 +419,12 @@ class ObjectStore(contextlib.AbstractContextManager):
 
         assert self.active
 
-        with self.cache.store(object_id) as name:
-            path = os.path.join(self.cache, name)
-            # we clamp the mtime of `obj` itself so that it
-            # resuming a snapshop and building with a snapshot
-            # goes through the same code path
-            obj.clamp_mtime()
-            obj.clone(path)
+        # we clamp the mtime of `obj` itself so that it
+        # resuming a snapshop and building with a snapshot
+        # goes through the same code path
+        obj.clamp_mtime()
+
+        self.cache.store_tree(object_id, obj.path + "/.")
 
     def cleanup(self):
         """Cleanup all created Objects that are still alive"""
diff --git a/osbuild/util/fscache.py b/osbuild/util/fscache.py
index f2f1c4fc..e99ae87d 100644
--- a/osbuild/util/fscache.py
+++ b/osbuild/util/fscache.py
@@ -9,10 +9,10 @@ the cache under a given limit.
 # pylint: disable=too-many-lines
 
 import contextlib
-import ctypes
 import errno
 import json
 import os
+import subprocess
 import uuid
 from typing import Any, Dict, NamedTuple, Optional, Tuple, Union
 
@@ -59,19 +59,19 @@ class FsCacheInfo(NamedTuple):
         version = None
 
         # parse "creation-boot-id"
-        _creation_boot_id = data.get("creation-boot-id", None)
+        _creation_boot_id = data.get("creation-boot-id")
         if isinstance(_creation_boot_id, str) and len(_creation_boot_id) == 32:
             creation_boot_id = _creation_boot_id
 
         # parse "maximum-size"
-        _maximum_size = data.get("maximum-size", None)
+        _maximum_size = data.get("maximum-size")
         if isinstance(_maximum_size, int):
             maximum_size = _maximum_size
         elif isinstance(_maximum_size, str) and _maximum_size == "unlimited":
             maximum_size = "unlimited"
 
         # parse "version"
-        _version = data.get("version", None)
+        _version = data.get("version")
         if isinstance(_version, int):
             version = _version
 
@@ -204,6 +204,7 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
     _filename_cache_info = "cache.info"
     _filename_cache_lock = "cache.lock"
     _filename_cache_size = "cache.size"
+    _filename_cache_tag = "CACHEDIR.TAG"
     _filename_object_info = "object.info"
     _filename_object_lock = "object.lock"
     _version_current = 1
@@ -211,7 +212,7 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
 
     # constant properties
     _appid: str
-    _libc: linux.Libc
+    _tracers: Dict[str, Any]
     _path_cache: Any
 
     # context-manager properties
@@ -241,7 +242,7 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
         """
 
         self._appid = appid
-        self._libc = linux.Libc.default()
+        self._tracers = {}
         self._path_cache = os.fspath(path_cache)
 
         self._active = False
@@ -250,6 +251,26 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
         self._info = FsCacheInfo()
         self._info_maximum_size = 0
 
+    def _trace(self, trace: str):
+        """Trace execution
+
+        Execute registered trace-hooks for the given trace string. This allows
+        tests to register callbacks that are executed at runtime at a specific
+        location in the code. During normal operation, no such hooks should be
+        used.
+
+        The trace-hooks are used to trigger race-conditions during tests and
+        verify they are handled gracefully.
+
+        Parameters:
+        -----------
+        trace
+            The trace-hook to run.
+        """
+
+        if trace in self._tracers:
+            self._tracers[trace]()
+
     @staticmethod
     def _calculate_size(path_target: str) -> int:
         """Calculate total size of a directory tree
@@ -349,16 +370,41 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
                 if write:
                     flags = flags | os.O_RDWR
                     lock = linux.fcntl.F_WRLCK
+                self._trace("_atomic_open:open")
                 fd = os.open(path, flags, 0o644)
+                self._trace("_atomic_open:lock")
                 linux.fcntl_flock(fd, lock, wait=wait)
 
                 # The file might have been replaced between opening it and
                 # acquiring the lock. Hence, run `stat(2)` on the path again
                 # and compare it to `fstat(2)` of the open file. If they differ
                 # simply retry.
-                st_fd = os.stat(fd)
-                st_path = os.stat(path)
-                if st_fd.st_dev != st_path.st_dev or st_fd.st_ino != st_path.st_ino:
+                # On NFS, the lock-acquisition has invalidated the caches, hence
+                # the metadata is refetched. On linux, the first query will
+                # succeed and reflect the drop in link-count. Every further
+                # query will yield `ESTALE`. Yet, we cannot rely on being the
+                # first to query, so proceed carefully.
+                # On non-NFS, information is coherent and we can simply proceed
+                # comparing the DEV+INO information to see whether the file was
+                # replaced.
+
+                retry = False
+
+                try:
+                    st_fd = os.stat(fd)
+                except OSError as e:
+                    if e.errno != errno.ESTALE:
+                        raise
+                    retry = True
+
+                try:
+                    st_path = os.stat(path)
+                except OSError as e:
+                    if e.errno not in [errno.ENOENT, errno.ESTALE]:
+                        raise
+                    retry = True
+
+                if retry or st_fd.st_dev != st_path.st_dev or st_fd.st_ino != st_path.st_ino:
                     linux.fcntl_flock(fd, linux.fcntl.F_UNLCK)
                     os.close(fd)
                     fd = None
@@ -437,28 +483,35 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
                 with os.fdopen(fd, "r+", closefd=False, encoding="utf8") as file:
                     yield file
 
-                if replace:
-                    flags = ctypes.c_uint(0)
-                else:
-                    flags = self._libc.RENAME_NOREPLACE
-
                 suppress = []
                 if ignore_exist:
                     suppress.append(errno.EEXIST)
 
-                # As a last step, move the file to the desired location.
-                with ctx.suppress_oserror(*suppress):
-                    self._libc.renameat2(
-                        oldpath=self._path(rpath_tmp).encode(),
-                        newpath=self._path(rpath).encode(),
-                        flags=flags,
+                if replace:
+                    # Move the file into the desired location, possibly
+                    # replacing any existing entry.
+                    os.rename(
+                        src=self._path(rpath_tmp),
+                        dst=self._path(rpath),
                     )
+                else:
+                    # Preferably, we used `RENAME_NOREPLACE`, but this is not
+                    # supported on NFS. Instead, we create a hard-link, which
+                    # will fail if the target already exists. We rely on the
+                    # cleanup-path to drop the original link.
+                    with ctx.suppress_oserror(*suppress):
+                        os.link(
+                            src=self._path(rpath_tmp),
+                            dst=self._path(rpath),
+                            follow_symlinks=False,
+                        )
         finally:
             if rpath_tmp is not None:
-                # If the temporary file exists, we delete it on error. If we
-                # haven't created it, or if we already moved it, this will be a
-                # no-op. Due to the unique name, we will never delete a file we
-                # do not own.
+                # If the temporary file exists, we delete it. If we haven't
+                # created it, or if we already moved it, this will be a no-op.
+                # Due to the unique name, we will never delete a file we do not
+                # own. If we hard-linked the file, this merely deletes the
+                # original temporary link.
                 # On fatal errors, we leak the file into the object store. Due
                 # to the released lock and UUID name, cache management will
                 # clean it up.
@@ -563,6 +616,11 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
 
         # Create the file-scaffolding of the cache. We fill in the default
         # information and ignore racing operations.
+        with self._atomic_file(self._filename_cache_tag, self._dirname_objects, ignore_exist=True) as f:
+            f.write(
+                "Signature: 8a477f597d28d172789f06886806bc55\n"
+                "# This is a cache directory tag created by osbuild (see https://bford.info/cachedir/)\n"
+            )
         with self._atomic_file(self._filename_cache_info, self._dirname_objects, ignore_exist=True) as f:
             json.dump({"version": self._version_current}, f)
         with self._atomic_file(self._filename_cache_lock, self._dirname_objects, ignore_exist=True) as f:
@@ -892,12 +950,16 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
                 # As last step move the entry to the desired location. If the
                 # target name is already taken, we bail out and pretend the
                 # entry was immediately overwritten by another one.
+                #
+                # Preferably, we used RENAME_NOREPLACE, but this is not
+                # available on all file-systems. Hence, we rely on the fact
+                # that non-empty directories cannot be replaced, so we
+                # automatically get RENAME_NOREPLACE behavior.
                 path_name = self._path(self._dirname_objects, name)
                 try:
-                    self._libc.renameat2(
-                        oldpath=path_uuid.encode(),
-                        newpath=path_name.encode(),
-                        flags=self._libc.RENAME_NOREPLACE,
+                    os.rename(
+                        src=path_uuid,
+                        dst=path_name,
                     )
                 except OSError as e:
                     ignore = [errno.EEXIST, errno.ENOTDIR, errno.ENOTEMPTY]
@@ -1044,3 +1106,49 @@ class FsCache(contextlib.AbstractContextManager, os.PathLike):
                         json.dump(info_raw, f)
 
         self._load_cache_info(info)
+
+    def store_tree(self, name: str, tree: Any):
+        """Store file system tree in cache
+
+        Create a new entry in the object store containing a copy of the file
+        system tree specified as `tree`. This behaves like `store()` but instead
+        of providing a context to the caller it will copy the specified tree.
+
+        Similar to `store()`, when the entry is committed it is immediately
+        unlocked and released to the cache. This means it might vanish at any
+        moment due to a parallel cleanup. Hence, a caller cannot rely on the
+        object being available in the cache once this call returns.
+
+        If `tree` points to a file, the file is copied. If it points to a
+        directory, the entire directory tree is copied including the root entry
+        itself. To copy an entire directory without its root entry, use the
+        `path/.` notation. Links are never followed but copied verbatim.
+        All metadata is preserved, if possible.
+
+        Parameters:
+        -----------
+        name
+            Name to store the object under.
+        tree:
+            Path to the file system tree to copy.
+        """
+
+        with self.store(name) as rpath_data:
+            r = subprocess.run(
+                [
+                    "cp",
+                    "--reflink=auto",
+                    "-a",
+                    "--",
+                    os.fspath(tree),
+                    self._path(rpath_data),
+                ],
+                check=False,
+                encoding="utf-8",
+                stderr=subprocess.STDOUT,
+                stdout=subprocess.PIPE,
+            )
+            if r.returncode != 0:
+                code = r.returncode
+                msg = r.stdout.strip()
+                raise RuntimeError(f"Cannot copy into file-system cache ({code}): {msg}")


Test patch:
diff --git a/test/mod/test_util_fscache.py b/test/mod/test_util_fscache.py
index 334f4736..b81b00d5 100644
--- a/test/mod/test_util_fscache.py
+++ b/test/mod/test_util_fscache.py
@@ -190,10 +190,12 @@ def test_scaffolding(tmpdir):
     with cache:
         pass
 
-    assert len(list(os.scandir(tmpdir))) == 5
+    assert len(list(os.scandir(tmpdir))) == 6
     assert len(list(os.scandir(os.path.join(tmpdir, cache._dirname_objects)))) == 0
     assert len(list(os.scandir(os.path.join(tmpdir, cache._dirname_stage)))) == 0
 
+    with open(os.path.join(tmpdir, cache._filename_cache_tag), "r", encoding="utf8") as f:
+        assert len(f.read()) > 0
     with open(os.path.join(tmpdir, cache._filename_cache_info), "r", encoding="utf8") as f:
         assert json.load(f) == {"version": 1}
     with open(os.path.join(tmpdir, cache._filename_cache_lock), "r", encoding="utf8") as f:
@@ -202,6 +204,20 @@ def test_scaffolding(tmpdir):
         assert f.read() == "0"
 
 
+def test_cachedir_tag(tmpdir):
+    #
+    # Verify compatibility to the cachedir-tag specification.
+    #
+
+    cache = fscache.FsCache("osbuild-test-appid", tmpdir)
+
+    with cache:
+        pass
+
+    with open(os.path.join(tmpdir, "CACHEDIR.TAG"), "r", encoding="utf8") as f:
+        assert f.read(43) == "Signature: 8a477f597d28d172789f06886806bc55"
+
+
 def test_cache_info(tmpdir):
     #
     # Verify that the cache reads and augments cache information. Also verify
@@ -290,6 +306,44 @@ def test_load(tmpdir):
                 pass
 
 
+def test_store_tree(tmpdir):
+    #
+    # API tests for the `store_tree()` method.
+    #
+
+    cache = fscache.FsCache("osbuild-test-appid", tmpdir)
+
+    with pytest.raises(AssertionError):
+        cache.store_tree("foobar", "invalid/dir")
+
+    with cache:
+        cache.info = cache.info._replace(maximum_size=1024*1024*1024)
+
+        with pytest.raises(ValueError):
+            cache.store_tree("", "invalid/dir")
+        with pytest.raises(RuntimeError):
+            cache.store_tree("key", "invalid/dir")
+
+        with tempfile.TemporaryDirectory(dir="/var/tmp") as tmp:
+            with open(os.path.join(tmp, "outside"), "x", encoding="utf8") as f:
+                f.write("foo")
+            os.mkdir(os.path.join(tmp, "tree"))
+            with open(os.path.join(tmp, "tree", "inside"), "x", encoding="utf8") as f:
+                f.write("bar")
+            with open(os.path.join(tmp, "tree", "more-inside"), "x", encoding="utf8") as f:
+                f.write("foobar")
+
+            cache.store_tree("key", os.path.join(tmp, "tree"))
+
+            with cache.load("key") as rpath:
+                assert len(list(os.scandir(os.path.join(cache, rpath)))) == 1
+                assert len(list(os.scandir(os.path.join(cache, rpath, "tree")))) == 2
+                with open(os.path.join(cache, rpath, "tree", "inside"), "r", encoding="utf8") as f:
+                    assert f.read() == "bar"
+                with open(os.path.join(cache, rpath, "tree", "more-inside"), "r", encoding="utf8") as f:
+                    assert f.read() == "foobar"
+
+
 def test_basic(tmpdir):
     #
     # A basic cache store+load test.
diff --git a/test/mod/test_util_fscache_coherency.py b/test/mod/test_util_fscache_coherency.py
new file mode 100644
index 00000000..9895d523
--- /dev/null
+++ b/test/mod/test_util_fscache_coherency.py
@@ -0,0 +1,277 @@
+#
+# Coherency tests for the 'osbuild.util.fscache' module.
+#
+
+# pylint: disable=protected-access
+
+import contextlib
+import errno
+import os
+import socket
+import subprocess
+import tempfile
+
+import pytest
+
+from osbuild.util import fscache, linux
+
+from .. import test
+
+
+def nfsd_available():
+    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+        s.settimeout(1)
+        return s.connect_ex(("localhost", 2049)) == 0
+
+
+@contextlib.contextmanager
+def mount_nfs(src: str, dst: str):
+    r = subprocess.run(
+        [
+            "mount",
+            "-t", "nfs",
+            "-o", "nosharecache,vers=4",
+            src,
+            dst,
+        ],
+        check=False,
+        encoding="utf-8",
+        stderr=subprocess.STDOUT,
+        stdout=subprocess.PIPE,
+    )
+    if r.returncode != 0:
+        code = r.returncode
+        msg = r.stdout.strip()
+        raise RuntimeError(f"{msg} (code: {code})")
+
+    try:
+        yield dst
+    finally:
+        subprocess.run(
+            ["umount", dst],
+            check=True,
+        )
+
+
+@pytest.fixture(name="nfsmnt")
+def nfsmnt_fixture():
+    tmpmnt = None
+    try:
+        tmpmnt = tempfile.mkdtemp(prefix="osbuild-test-", dir="/var/tmp")
+        with mount_nfs("localhost:/", tmpmnt) as tmpnfs:
+            with tempfile.TemporaryDirectory(dir=tmpnfs) as tmpdir:
+                yield tmpdir
+    finally:
+        os.rmdir(tmpmnt)
+
+
+@pytest.fixture(name="nfsmnts")
+def nfsmnts_fixture():
+    tmpmnt = None
+    try:
+        tmpmnt = tempfile.mkdtemp(prefix="osbuild-test-", dir="/var/tmp")
+        os.mkdir(os.path.join(tmpmnt, "a"))
+        os.mkdir(os.path.join(tmpmnt, "b"))
+        with mount_nfs("localhost:/", os.path.join(tmpmnt, "a")), \
+                mount_nfs("localhost:/", os.path.join(tmpmnt, "b")):
+            with tempfile.TemporaryDirectory(dir=os.path.join(tmpmnt, "a")) as tmpdir:
+                dirname = os.path.basename(os.path.normpath(tmpdir))
+                a = os.path.join(tmpmnt, "a", dirname)
+                b = os.path.join(tmpmnt, "b", dirname)
+                yield (a, b)
+    finally:
+        os.rmdir(os.path.join(tmpmnt, "b"))
+        os.rmdir(os.path.join(tmpmnt, "a"))
+        os.rmdir(tmpmnt)
+
+
+@pytest.mark.skipif(not nfsd_available(), reason="NFSv4 daemon required")
+@pytest.mark.skipif(not test.TestBase.can_bind_mount(), reason="Bind-mounting required")
+def test_nfs_characteristics(nfsmnts):
+    #
+    # Test NFS Characteristic
+    #
+    # This mounts a single NFS instance with `nosharedcache` twice. It then
+    # runs a series of custom tests to verify cache-behavior of NFS and how
+    # different operations will cause stale caches and thus validate our
+    # assumptions on how to avoid them.
+    #
+
+    a = nfsmnts[0]
+    b = nfsmnts[1]
+
+    # `stat` does not invalidate caches
+    #
+    # Write a fresh file on A, then stat it on B. This will properly re-fetch
+    # all information since caches on both sides are empty. Then re-write the
+    # file on A and, again, stat it on B and verify that the metadata was *NOT*
+    # re-fetched, since NFS caches this information.
+    # As last step read the file on B and verify that `open()` will properly
+    # re-fetch all information.
+
+    with open(os.path.join(a, "0_foo"), "x", encoding="utf8") as f:
+        f.write("foo")
+
+    assert os.stat(os.path.join(b, "0_foo")).st_size == 3
+
+    with open(os.path.join(a, "0_foo"), "w", encoding="utf8") as f:
+        f.write("foobar")
+
+    assert os.stat(os.path.join(b, "0_foo")).st_size == 3
+
+    with open(os.path.join(b, "0_foo"), "r", encoding="utf8") as f:
+        assert f.read() == "foobar"
+
+    assert os.stat(os.path.join(b, "0_foo")).st_size == 6
+
+    # Lock-acquisition invalidates caches
+    #
+    # Create a file on A and commit it to disk. Open it for reading on A, but
+    # then delete it on B. Verify on B it is gone. On A continue reading the
+    # file. Stat it on A to verify the caches have not been invalidated. Then
+    # acquire a read-lock on the open file and try the same again, this time
+    # the cache should reflect the unlink.
+
+    with open(os.path.join(a, "1_foo"), "x", encoding="utf8") as f:
+        f.write("foo")
+
+    with open(os.path.join(a, "1_foo"), "r", encoding="utf8") as f:
+        os.unlink(os.path.join(b, "1_foo"))
+        assert not os.access(os.path.join(b, "1_foo"), os.R_OK)
+
+        assert f.read() == "foo"
+        assert os.stat(f.fileno()).st_nlink == 1
+        assert os.stat(os.path.join(a, "1_foo")).st_ino != 0
+        assert os.access(os.path.join(a, "1_foo"), os.R_OK)
+
+        linux.fcntl_flock(f.fileno(), linux.fcntl.F_RDLCK, wait=True)
+
+        # The first STAT after an unlink returns a link-count of 0,
+        # while every following STAT raises ESTALE. Lets try
+        # verifying that, but do not depend on it and allow the
+        # first STAT to raise ESTALE as well.
+        try:
+            assert os.stat(f.fileno()).st_nlink == 0
+        except OSError as e:
+            assert e.errno == errno.ESTALE
+
+        with pytest.raises(OSError):
+            os.stat(f.fileno())
+        with pytest.raises(OSError):
+            os.stat(os.path.join(a, "1_foo"))
+
+        assert not os.access(os.path.join(a, "1_foo"), os.R_OK)
+
+    # Inode changes on replacement
+    #
+    # Create a file, STAT it on A and B and verify they match and the
+    # caches are active. Then replace the file on A and STAT again. The
+    # replacement will be visible on A, but the caches on B still yield
+    # the same old value. However, after opening on B, the file content
+    # will yield the updated data and so will STAT.
+
+    with open(os.path.join(a, "2_foo"), "x", encoding="utf8") as f:
+        f.write("foo")
+
+    st_a0 = os.stat(os.path.join(a, "2_foo"))
+    st_b0 = os.stat(os.path.join(b, "2_foo"))
+    assert st_a0.st_ino == st_b0.st_ino
+
+    with open(os.path.join(a, "2_bar"), "x", encoding="utf8") as f:
+        f.write("bar")
+    os.rename(os.path.join(a, "2_bar"), os.path.join(a, "2_foo"))
+
+    st_a1 = os.stat(os.path.join(a, "2_foo"))
+    st_b1 = os.stat(os.path.join(b, "2_foo"))
+
+    assert st_a1.st_ino != st_b1.st_ino
+    assert st_b0.st_ino == st_b1.st_ino
+
+    with open(os.path.join(b, "2_foo"), "r", encoding="utf8") as f:
+        assert f.read() == "bar"
+
+    st_a2 = os.stat(os.path.join(a, "2_foo"))
+    st_b2 = os.stat(os.path.join(b, "2_foo"))
+
+    assert st_a2.st_ino == st_b2.st_ino
+
+
+def _test_atomics_with(a: str, b: str):
+    with fscache.FsCache("osbuild-test-appid", a) as cache:
+        cache.info = cache.info._replace(maximum_size=1024)
+
+        # Test _atomic_open() with open+lock race
+        #
+        # Create a file `0_foo` and OPEN+LOCK it with _atomic_open(). Use a
+        # tracer to hook between OPEN and LOCK. First time unlinke the target
+        # file and recreate it. The second time, replace it instead.
+        #
+        # Verify that `_atomic_open()` needs 3 attempts to OPEN+LOCK the file,
+        # and verify the content is ultimately correct.
+
+        def _trace_lock(state: dict):
+            # Use `open(..., "x")` to force an invalidation of NFS caches.
+            # Otherwise, VFS would just deny our operations based on outdated
+            # NFS-caches. This would be coherent, but overly restrictive.
+            with pytest.raises(OSError):
+                with open(os.path.join(b, "0_foo"), "x", encoding="utf8") as f:
+                    pass
+
+            if state["lock"] == 0:
+                with open(os.path.join(b, "0_foo"), "r", encoding="utf8") as f:
+                    assert f.read() == "foo"
+                os.unlink(os.path.join(b, "0_foo"))
+                with open(os.path.join(b, "0_foo"), "x", encoding="utf8") as f:
+                    f.write("bar")
+            elif state["lock"] == 1:
+                with open(os.path.join(b, "0_foo"), "r", encoding="utf8") as f:
+                    assert f.read() == "bar"
+                with open(os.path.join(b, "0_foo2"), "x", encoding="utf8") as f:
+                    f.write("foobar")
+                os.rename(os.path.join(b, "0_foo2"), os.path.join(b, "0_foo"))
+
+            state["lock"] = state["lock"] + 1
+
+        state = {"lock": 0}
+        cache._tracers = {
+            "_atomic_open:lock": lambda: _trace_lock(state)
+        }
+        with open(os.path.join(a, "0_foo"), "x", encoding="utf8") as f:
+            f.write("foo")
+        with cache._atomic_open("0_foo", wait=True, write=False) as fd:
+            with os.fdopen(fd, "r", closefd=False, encoding="utf8") as f:
+                assert f.read() == "foobar"
+
+        assert state["lock"] == 3
+
+
+@pytest.mark.skipif(not test.TestBase.can_bind_mount(), reason="Bind-mounting required")
+def test_atomics():
+    #
+    # Test FsCache Atomics (native)
+    #
+    # Verify the behavior of the `_atomic_*()` helpers of FsCache. Use the
+    # trace-hooks of FsCache to trigger the race-conditions we want to test.
+    #
+
+    with tempfile.TemporaryDirectory(dir="/var/tmp") as tmpdir:
+        _test_atomics_with(tmpdir, tmpdir)
+
+
+@pytest.mark.skipif(not nfsd_available(), reason="NFSv4 daemon required")
+@pytest.mark.skipif(not test.TestBase.can_bind_mount(), reason="Bind-mounting required")
+def test_atomics_nfs(nfsmnts):
+    #
+    # Test FsCache Atomics (NFS)
+    #
+    # Same as `test_atomics()` but on NFS.
+    #
+
+    with tempfile.TemporaryDirectory(dir=nfsmnts[0]) as tmpdir:
+        _test_atomics_with(tmpdir, tmpdir)
+
+    # Preferably, we would now run the same tests on two distinct
+    # NFS mounts with no shared caches. Unfortunately, this keeps
+    # triggering kernel-oopses, so we disable the tests for now:
+
+    #_test_atomics_with(nfsmnts[0], nfsmnts[1])