lib.FcCache: Add a write-through read cache

Add a write-through read cache in front of a file context.

stat() and get() results are cached: stat per (path, follow_symlinks), get
per (path, wd, throw). Successful stats and FileNotFoundErrors are cached;
other errors are re-probed, so a transient transport failure does not
poison the cache. Writes pass through to the wrapped context and drop the
cache entries of the affected paths. The existence and directory checks are
rerouted through the cached stat, so they pay the cache instead of the
wrapped context's own fast path.

warm() fills the stat cache of a set of paths, as file_exists() would stat
them, without following symlinks. If the wrapped context is an exec
context, one host command stats all of the paths; otherwise, one stat per
path runs.

The GNU stat format of ExecContext is lifted to the module constant
STAT_GNU_FORMAT, and the field parser is exported as parse_stat_fields(),
so warm() fills entries that are indistinguishable from a real stat().

Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.1.0
Signed-off-by: Jan Lindemann <jan@janware.com>

src/python/jw/pkg/lib/ExecContext.py

@@ -17,6 +17,24 @@ from .FileContext import FileContext as Base
 from .log import DEBUG, ERR, NOTICE, log
 
 _US = '\x1f'  # unlikely to appear in numeric output
+
+STAT_GNU_FORMAT = _US.join(
+    [
+        '%f',  # st_mode in hex
+        '%i',  # st_ino
+        '%d',  # st_dev
+        '%h',  # st_nlink
+        '%U',  # st_uid
+        '%G',  # st_gid
+        '%s',  # st_size
+        '%.9X',  # st_atime
+        '%.9Y',  # st_mtime
+        '%.9Z',  # st_ctime
+        '%o',  # st_blksize hint
+        '%b',  # st_blocks
+        '%r',  # st_rdev
+    ]
+)
 _BILLION = Decimal(1_000_000_000)
 
 def _looks_like_option_error(stderr: str | None) -> bool:
@@ -54,7 +72,10 @@ def _parse_epoch(value: str) -> tuple[int, float, int]:
     ns = int((dec * _BILLION).to_integral_value(rounding = ROUND_FLOOR))
     return sec, float(dec), ns
 
-def _build_stat_result(fields: list[str], mode_base: int) -> StatResult:
+def parse_stat_fields(stdout: str, mode_base: int = 16) -> StatResult:  # export
+    return build_stat_result(stdout.split(_US), mode_base = mode_base)
+
+def build_stat_result(fields: list[str], mode_base: int) -> StatResult:  # export
     if len(fields) != 13:
         raise ValueError(
             f'unexpected stat output: expected 13 fields, got {len(fields)}: {fields!r}'
@@ -621,27 +642,9 @@ class ExecContext(Base):
             )
 
         # GNU coreutils stat
-        gnu_format = _US.join(
-            [
-                '%f',  # st_mode in hex
-                '%i',  # st_ino
-                '%d',  # st_dev
-                '%h',  # st_nlink
-                '%U',  # st_uid
-                '%G',  # st_gid
-                '%s',  # st_size
-                '%.9X',  # st_atime
-                '%.9Y',  # st_mtime
-                '%.9Z',  # st_ctime
-                '%o',  # st_blksize hint
-                '%b',  # st_blocks
-                '%r',  # st_rdev
-            ]
-        )
-
-        result = await __stat(['--printf', gnu_format])
+        result = await __stat(['--printf', STAT_GNU_FORMAT])
         if result.status == 0 and result.stdout is not None:
-            return _build_stat_result(result.stdout_str.split(_US), mode_base = 16)
+            return parse_stat_fields(result.stdout_str, mode_base = 16)
 
         if not _looks_like_option_error(result.stderr_str_or_none):
             # log(DEBUG, f'GNU stat attempt failed on "{path}" ({str(e)})')
@@ -671,7 +674,7 @@ class ExecContext(Base):
         if result.status != 0 or stdout is None:
             _raise_stat_error(path, result)
         assert stdout is not None  # Just there to pacify the linter
-        return _build_stat_result(stdout.rstrip('\n').split(_US), mode_base = 8)
+        return parse_stat_fields(stdout.rstrip('\n'), mode_base = 8)
 
     @override
     async def _chown(self, path: str, owner: str | None, group: str | None) -> None:

src/python/jw/pkg/lib/FcCache.py

@@ -0,0 +1,234 @@
+from __future__ import annotations
+
+import errno
+import stat
+
+from typing import TYPE_CHECKING, Iterable, override
+
+if TYPE_CHECKING:
+    from .base import Result
+    from .FileContext import FileContext
+    from .file_types import FileType
+
+from .base import StatResult
+from .ExecContext import STAT_GNU_FORMAT, ExecContext, parse_stat_fields
+from .FcWrapper import FcWrapper
+
+class FcCache(FcWrapper):  # export
+    """A write-through read cache in front of a file context.
+
+    stat and get results are cached: stat per (path,
+    follow_symlinks), get per (path, wd, throw). Successful stats
+    and FileNotFoundErrors are cached; other errors are re-probed,
+    so a transient transport failure does not poison the cache.
+    Writes pass through to the wrapped context and drop the cache
+    entries of the affected paths.
+
+    warm() probes a set of paths without following symlinks and
+    fills the stat cache with the results: an absent path is
+    absent, and a non-link is the same node, for both follow
+    variants, so the result is cached for both; a link is cached
+    without following, its followed variant probed on first use.
+    If the wrapped context is an exec context, one host command
+    stats all of the paths; otherwise, one stat per path runs.
+    Paths containing tab or newline characters are not supported.
+    """
+
+    def __init__(self, wrapped: FileContext) -> None:
+        super().__init__(wrapped)
+        self.__stat_cache: dict[tuple[str, bool], StatResult | Exception] = {}
+        self.__get_cache: dict[tuple[str, str | None, bool], Result] = {}
+
+    @override
+    async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
+        key = (path, follow_symlinks)
+        hit = self.__stat_cache.get(key)
+        if hit is not None:
+            if isinstance(hit, StatResult):
+                return hit
+            raise hit
+        try:
+            st = await super()._stat(path, follow_symlinks)
+        except FileNotFoundError as e:
+            self.__stat_cache[key] = e
+            raise
+        self.__stat_cache[key] = st
+        return st
+
+    @override
+    async def _get(
+        self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str
+    ) -> Result:
+        key = (path, wd, throw)
+        hit = self.__get_cache.get(key)
+        if hit is None:
+            hit = await super()._get(path, wd, throw, verbose, title)
+            self.__get_cache[key] = hit
+        return hit
+
+    @override
+    async def _file_exists(self, path: str) -> bool:
+        # -- Reroute the existence check through the cached stat
+        #    instead of the wrapped context's own fast path. A path
+        #    below a file cannot exist: the exec contexts report it
+        #    as absent, so the local one is treated the same
+        try:
+            await self._stat(path, False)
+        except FileNotFoundError:
+            return False
+        except NotADirectoryError:
+            return False
+        return True
+
+    @override
+    async def _is_dir(self, path: str, follow_symlinks: bool) -> bool:
+        try:
+            return stat.S_ISDIR((await self._stat(path, follow_symlinks)).mode)
+        except FileNotFoundError:
+            return False
+        except NotADirectoryError:
+            return False
+
+    async def warm(self, paths: Iterable[str]) -> None:
+        """Probe a set of paths without following symlinks and fill
+        the stat cache, as the class docstring describes.
+
+        A path whose both follow variants are already in the cache
+        is skipped, so a warm of a set that is cached already does
+        nothing. If the wrapped context is an exec context, one
+        host command stats all of the remaining paths; otherwise,
+        one stat per path runs.
+        """
+        composed = []
+        for p in paths:
+            if '\t' in p or '\n' in p:
+                raise ValueError(f'warm: path {p!r} holds a tab or newline')
+            c = self._chroot(p)
+            if (c, False) not in self.__stat_cache or \
+                    (c, True) not in self.__stat_cache:
+                composed.append(c)
+        if not composed:
+            return
+        ec = self.wrapped
+        if not isinstance(ec, ExecContext):
+            for p in composed:
+                try:
+                    st = await super()._stat(p, False)
+                except FileNotFoundError as e:
+                    self.__cache_probe(p, e)
+                except NotADirectoryError:
+                    # -- A path below a file is not statable; it is
+                    #    not cached, and file_exists() judges it
+                    #    absent
+                    pass
+                else:
+                    self.__cache_probe(p, st)
+            return
+        script = (
+            'while IFS= read -r p; do '
+            "printf '%s\\t' \"$p\"; "
+            f"stat --printf '{STAT_GNU_FORMAT}\\n' \"$p\" 2>/dev/null || "
+            "printf 'ABSENT\\n'; "
+            'done'
+        )
+        ret = await ec.run(
+            ['sh', '-c', script],
+            cmd_input = ('\n'.join(composed) + '\n').encode(),
+            mod_env = {'LC_ALL': 'C'},
+            throw = True,
+        )
+        self.__parse_probe(ret.stdout_str, composed)
+
+    def __cache_probe(self, path: str, result: StatResult | FileNotFoundError) -> None:
+        # -- Cache a probe that did not follow symlinks: an absent
+        #    path is absent, and a non-link is the same node, for
+        #    both follow variants; a link is cached without
+        #    following, its followed variant probed on first use
+        self.__stat_cache[(path, False)] = result
+        if isinstance(result, FileNotFoundError) or not stat.S_ISLNK(result.mode):
+            self.__stat_cache[(path, True)] = result
+
+    def __parse_probe(self, stdout: str, composed: list[str]) -> None:
+        # -- One line per path: the path, a tab, then the stat
+        #    fields or ABSENT
+        lines = stdout.splitlines()
+        if len(lines) != len(composed):
+            raise RuntimeError(
+                f'warm: probe reported {len(lines)} lines for {len(composed)} paths'
+            )
+        for p, line in zip(composed, lines):
+            name, sep, rest = line.partition('\t')
+            if name != p or not sep:
+                raise RuntimeError(f'warm: probe line does not name "{p}": {line!r}')
+            if rest == 'ABSENT':
+                self.__cache_probe(p, FileNotFoundError(errno.ENOENT, f'stat: {p}'))
+            else:
+                self.__cache_probe(p, parse_stat_fields(rest))
+
+    def __invalidate(self, *paths: str) -> None:
+        # -- A write changes the stat and get answers of its paths
+        for p in paths:
+            for fl in (True, False):
+                self.__stat_cache.pop((p, fl), None)
+            self.__get_cache = {k: r for k, r in self.__get_cache.items() if k[0] != p}
+
+    @override
+    async def _put(
+        self,
+        path: str,
+        content: bytes,
+        wd: str | None,
+        throw: bool,
+        verbose: bool | None,
+        title: str,
+        owner: str | None,
+        group: str | None,
+        mode: str | None,
+        atomic: bool,
+        ftype: FileType,
+    ) -> Result:
+        r = await super()._put(
+            path,
+            content,
+            wd,
+            throw,
+            verbose,
+            title,
+            owner,
+            group,
+            mode,
+            atomic,
+            ftype,
+        )
+        self.__invalidate(path)
+        return r
+
+    @override
+    async def _mkdir(self, path: str, mode: int, ftype: FileType) -> None:
+        await super()._mkdir(path, mode, ftype)
+        self.__invalidate(path)
+
+    @override
+    async def _unlink(self, path: str) -> None:
+        await super()._unlink(path)
+        self.__invalidate(path)
+
+    @override
+    async def _erase(self, path: str) -> None:
+        await super()._erase(path)
+        self.__invalidate(path)
+
+    @override
+    async def _rename(self, src: str, dst: str) -> None:
+        await super()._rename(src, dst)
+        self.__invalidate(src, dst)
+
+    @override
+    async def _chown(self, path: str, owner: str | None, group: str | None) -> None:
+        await super()._chown(path, owner, group)
+        self.__invalidate(path)
+
+    @override
+    async def _chmod(self, path: str, mode: int) -> None:
+        await super()._chmod(path, mode)
+        self.__invalidate(path)