Handle special filesystem entries explicitly (#11)

* Handle special filesystem entries explicitly

- Added fs.entry_kind(): classifies via lstat() mode bits alone
  ("file", "dir", "symlink", "fifo", "socket", "block_device",
  "char_device", "unknown") - never opens the entry, so classifying a
  FIFO can't block even without a writer.
- rsync() now classifies every source entry before touching it and
  yields a new Actions.EXCLUDE for anything outside file/dir/symlink,
  instead of falling through to copy_direntry()'s file branch, which
  would either block forever opening a writerless FIFO for read, or
  raise an OSError that initiate_backup() treats as fatal (aborting the
  whole backup for one socket).
- initiate_backup() records exclusions in the completion manifest
  (keyed by snapshot-relative path, valued with the excluded kind), and
  no longer nest_hardlinks or counts a recurring exclusion as a change
  worth keeping a new snapshot for.
- Chose the "recorded exclusion" policy the plan allows, rather than
  fail-fast: a FIFO or socket living in a backed-up tree is not itself
  a backup failure, and previously turned into one.

Note: rsync_ext() (external rsync) is unaffected by this change and
keeps whatever its own --archive flags do with special files; the P1
item on Python/external implementation parity covers reconciling that.
The pre-existing test_src_is_socket failure (AF_UNIX path length on
macOS tmp_path) is untouched here - it's separately tracked under P1
test fixes - but this PR adds working socket/FIFO/char-device coverage
via a short-path tempdir.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Address Copilot review: fix stale exclude, exclusion-only diffs, test portability

- fs.rsync(): remove the existing dst_entry before yielding EXCLUDE for
  a src entry with an unsupported kind, so an entry inherited via
  hardlink from a prior snapshot doesn't linger in the new one.
- backup.initiate_backup(): compare this run's excluded set against
  the previous snapshot's manifest; a change in exclusions alone now
  counts as backup_changed, so the updated manifest gets written.
- tests: skip mkfifo/AF_UNIX-dependent tests on platforms without
  support, and use TemporaryDirectory/context managers instead of
  manual mkdtemp/unlink for more robust cleanup.

* Remove unused shutil import from test_rsync.py

Left over after switching the socket/fifo tests to
tempfile.TemporaryDirectory(), per Copilot's suppressed comment.

* Fix rm_direntry() no-op on special dst entries

rm_direntry() only handled file/symlink/dir - a FIFO, socket, or
device node at dst (e.g. inherited from an external-rsync snapshot)
silently survived removal, since is_file/is_dir/is_symlink all
return False for it. Any non-directory entry is unlinkable the same
way, so invert the branching: recurse for directories, unlink
everything else.

---------

Co-authored-by: Maxim Shulga <maxim.shulga@alludo.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

IMPROVEMENTS.md

@@ -106,11 +106,11 @@ Acceptance criteria:
 
 ### Handle special filesystem entries explicitly
 
-- [ ] Distinguish regular files, directories, symlinks, sockets, FIFOs, block
+- [x] Distinguish regular files, directories, symlinks, sockets, FIFOs, block
   devices, and character devices.
-- [ ] Never open an unsupported entry as a regular file.
-- [ ] Define a fail or recorded-exclusion policy for every unsupported type.
-- [ ] Record exclusions in the snapshot manifest.
+- [x] Never open an unsupported entry as a regular file.
+- [x] Define a fail or recorded-exclusion policy for every unsupported type.
+- [x] Record exclusions in the snapshot manifest.
 
 Acceptance criteria:
 

curateipsum/backup.py

@@ -159,12 +159,14 @@ def _ensure_repo_id(backups_dir: str) -> str:
 
 
 def _write_manifest(marker_path: str, backups_dir: str, snapshot_name: str,
-                    checksums: Optional[Dict[str, str]] = None):
+                    checksums: Optional[Dict[str, str]] = None,
+                    excluded: Optional[Dict[str, str]] = None):
     """Write and fsync the completion manifest for a finished snapshot."""
     manifest = json.dumps({
         "repo_id": _ensure_repo_id(backups_dir),
         "snapshot": snapshot_name,
         "checksums": checksums or {},
+        "excluded": excluded or {},
     })
     _write_small_file_no_follow(marker_path, manifest)
 
@@ -719,7 +721,8 @@ def process_backed_entry(backup_dir: str,
     - if DirEntry was not deleted, hardlink it to DELTA_DIR.
     """
     _lg.debug("%s %s %s", action, entry_relpath, msg)
-    if action not in (fs.Actions.ERROR, fs.Actions.DELETE):
+    if action not in (fs.Actions.ERROR, fs.Actions.DELETE,
+                      fs.Actions.EXCLUDE):
         fs.nest_hardlink(src_dir=backup_dir, src_relpath=entry_relpath,
                          dst_dir=os.path.join(backup_dir, DELTA_DIR))
 
@@ -802,6 +805,11 @@ def initiate_backup(sources,
     # keyed by its path relative to the snapshot root. Recorded in the
     # manifest so a completed snapshot's integrity can be verified later.
     checksums: Dict[str, str] = {}
+    # Entries this run deliberately never backed up (sockets, FIFOs,
+    # devices, ...), keyed by path relative to the snapshot root, valued
+    # with the excluded kind. Recorded in the manifest so an exclusion is
+    # a visible, auditable result rather than a silent gap.
+    excluded: Dict[str, str] = {}
     try:
         for src in sources:
             src_abs = os.path.abspath(src)
@@ -817,6 +825,8 @@ def initiate_backup(sources,
                         "Failed to copy %s: %s" % (entry_relpath, msg)
                     )
                 full_relpath = os.path.join(src_name, entry_relpath)
+                if action == fs.Actions.EXCLUDE:
+                    excluded[full_relpath] = msg
                 if action in (fs.Actions.CREATE, fs.Actions.REWRITE):
                     if msg:
                         # fs.rsync() already hashed the file while
@@ -832,14 +842,17 @@ def initiate_backup(sources,
                                 and not os.path.islink(full_path)):
                             checksums[full_relpath] = _sha256_file(full_path)
                 # TODO maybe should be run if first backup too?
-                if latest_backup is not None:
+                if latest_backup is not None and action != fs.Actions.EXCLUDE:
                     process_backed_entry(
                         backup_dir=staging.path,
                         entry_relpath=full_relpath,
                         action=action,
                         msg=msg,
                     )
-                # raise flag if something was changed since last backup
+                # raise flag if something was changed since last backup.
+                # A recurring exclusion (e.g. a FIFO that's always there)
+                # is not itself a change worth keeping a new snapshot for.
+                if action != fs.Actions.EXCLUDE:
                     backup_changed = True
     except fs.BackupCreationError as err:
         _lg.error("Error during backup creation: %s", err)
@@ -858,7 +871,18 @@ def initiate_backup(sources,
         _lg.info("Dry-run, removing staging directory: %s", snapshot_name)
         shutil.rmtree(staging.path, ignore_errors=True)
         return None
-    # do not create backup if no change from previous one
+    # do not create backup if no change from previous one - but a change
+    # in *which* entries are excluded is itself a change worth keeping,
+    # or the new exclusions would never make it into a written manifest.
+    if latest_backup is not None and not backup_changed:
+        prev_manifest = _read_manifest(_get_backup_marker(latest_backup).path)
+        prev_excluded = (
+            prev_manifest.get("excluded", {})
+            if isinstance(prev_manifest, dict) else {}
+        )
+        if excluded != prev_excluded:
+            backup_changed = True
+
     if latest_backup is not None and not backup_changed:
         _lg.info("Staged backup is the same as previous one %s, removing",
                  latest_backup.name)
@@ -869,7 +893,8 @@ def initiate_backup(sources,
     # then atomically rename the staging dir into its final snapshot name
     marker_name = "%s_%s" % (BACKUP_MARKER, snapshot_name)
     _write_manifest(os.path.join(staging.path, marker_name),
-                    backups_dir, snapshot_name, checksums=checksums)
+                    backups_dir, snapshot_name, checksums=checksums,
+                    excluded=excluded)
     _fsync_dir(staging.path)
 
     final_path = os.path.join(backups_dir, snapshot_name)

curateipsum/fs.py

@@ -7,6 +7,7 @@ import glob
 import hashlib
 import logging
 import os
+import stat
 import subprocess
 import sys
 import uuid
@@ -28,6 +29,40 @@ class Actions(enum.Enum):
     UPDATE_OWNER = enum.auto()
     CREATE = enum.auto()
     ERROR = enum.auto()
+    EXCLUDE = enum.auto()
+
+
+# Entry kinds copy_direntry() knows how to back up. Everything else
+# (FIFOs, sockets, block/char devices, and anything lstat() doesn't
+# recognize) is classified and excluded without ever being opened - a
+# FIFO with no writer would otherwise block open() for read forever, and
+# reading a device node can return unbounded or meaningless data.
+_SUPPORTED_KINDS = frozenset({"file", "dir", "symlink"})
+
+
+def entry_kind(entry: Union[os.DirEntry, "PseudoDirEntry"]) -> str:
+    """
+    Classify a filesystem entry via lstat() mode bits alone - never
+    opens it, so classifying a FIFO can't block even without a writer.
+    :return: one of "file", "dir", "symlink", "fifo", "socket",
+        "block_device", "char_device", "unknown".
+    """
+    mode = entry.stat(follow_symlinks=False).st_mode
+    if stat.S_ISLNK(mode):
+        return "symlink"
+    if stat.S_ISDIR(mode):
+        return "dir"
+    if stat.S_ISREG(mode):
+        return "file"
+    if stat.S_ISFIFO(mode):
+        return "fifo"
+    if stat.S_ISSOCK(mode):
+        return "socket"
+    if stat.S_ISBLK(mode):
+        return "block_device"
+    if stat.S_ISCHR(mode):
+        return "char_device"
+    return "unknown"
 
 
 class PseudoDirEntry:
@@ -202,15 +237,19 @@ def scantree(path, dir_first=True) -> Iterable[os.DirEntry]:
 
 
 def rm_direntry(entry: Union[os.DirEntry, PseudoDirEntry]):
-    """ Recursively delete DirEntry (dir/file/symlink). """
-    if entry.is_file(follow_symlinks=False) or entry.is_symlink():
-        os.unlink(entry.path)
-    elif entry.is_dir(follow_symlinks=False):
+    """
+    Recursively delete DirEntry. Directories are removed recursively;
+    anything else (file, symlink, or a special entry like a FIFO,
+    socket, or device node) is unlinked directly.
+    """
+    if entry.is_dir(follow_symlinks=False):
         with os.scandir(entry.path) as it:
             child_entry: os.DirEntry
             for child_entry in it:
                 rm_direntry(child_entry)
         os.rmdir(entry.path)
+    else:
+        os.unlink(entry.path)
 
 
 try:
@@ -429,6 +468,21 @@ def rsync(src_dir,
         del src_files_map[rel_path]
 
         src_entry: os.DirEntry
+        src_kind = entry_kind(src_entry)
+        if src_kind not in _SUPPORTED_KINDS:
+            _lg.warning("Rsync, excluding unsupported entry (%s): %s",
+                       src_kind, rel_path)
+            # dst_entry may exist here (e.g. inherited via hardlink from
+            # the previous snapshot); an excluded entry must not linger.
+            if not dry_run:
+                try:
+                    rm_direntry(dst_entry)
+                except OSError as exc:
+                    yield rel_path, Actions.ERROR, str(exc)
+                    continue
+            yield rel_path, Actions.EXCLUDE, src_kind
+            continue
+
         # rewrite dst if it has different type from src
         if src_entry.is_file(follow_symlinks=False):
             if not dst_entry.is_file(follow_symlinks=False):
@@ -543,6 +597,13 @@ def rsync(src_dir,
 
     # process remained source entries (new files/dirs/symlinks)
     for rel_path, src_entry in src_files_map.items():
+        src_kind = entry_kind(src_entry)
+        if src_kind not in _SUPPORTED_KINDS:
+            _lg.warning("Rsync, excluding unsupported entry (%s): %s",
+                       src_kind, rel_path)
+            yield rel_path, Actions.EXCLUDE, src_kind
+            continue
+
         dst_path = os.path.join(dst_root_abs, rel_path)
         _lg.debug("Rsync, creating: %s", rel_path)
         try:

tests/test_integration.py

@@ -1081,3 +1081,35 @@ def test_external_rsync_snapshot_still_gets_checksums(integration_dirs):
     manifest = bk._read_manifest(bk._get_backup_marker(snapshot).path)
     assert manifest["checksums"]
     assert bk.verify_snapshot(snapshot) == []
+
+
+@pytest.mark.skipif(not hasattr(os, "mkfifo"),
+                    reason="os.mkfifo not available on this platform")
+def test_fifo_excluded_and_recorded_without_blocking(integration_dirs):
+    """
+    A FIFO in the source must not block the backup (no writer is ever
+    attached), must not abort it, and must be recorded in the manifest
+    as an exclusion rather than silently disappearing.
+    """
+    backups_dir, source_dir = integration_dirs
+    (source_dir / "file1.txt").write_text("content1")
+    os.mkfifo(str(source_dir / "a_fifo"))
+
+    snapshot = bk.initiate_backup(
+        sources=[str(source_dir)],
+        backups_dir=str(backups_dir),
+        dry_run=False,
+    )
+
+    assert snapshot is not None
+    source_name = os.path.basename(str(source_dir))
+    manifest = bk._read_manifest(bk._get_backup_marker(snapshot).path)
+    assert manifest["excluded"] == {
+        os.path.join(source_name, "a_fifo"): "fifo"
+    }
+    assert not os.path.lexists(
+        os.path.join(snapshot.path, source_name, "a_fifo")
+    )
+    assert os.path.exists(
+        os.path.join(snapshot.path, source_name, "file1.txt")
+    )

tests/test_rsync.py

@@ -709,3 +709,103 @@ class TestRsyncHardlinkImmutability:
 
         assert os.lstat(str(external_target)).st_mode == external_mode_before
         assert (external_target / "canary").read_text() == "do-not-touch"
+
+
+class TestEntryKind:
+    """Tests for fs.entry_kind() classification."""
+
+    def test_regular_file(self, tmp_path):
+        path = os.path.join(str(tmp_path), "f")
+        with open(path, "w") as f:
+            f.write("x")
+        with os.scandir(str(tmp_path)) as it:
+            entry = next(iter(it))
+        assert fs.entry_kind(entry) == "file"
+
+    def test_directory(self, tmp_path):
+        path = os.path.join(str(tmp_path), "d")
+        os.mkdir(path)
+        with os.scandir(str(tmp_path)) as it:
+            entry = next(iter(it))
+        assert fs.entry_kind(entry) == "dir"
+
+    def test_symlink(self, tmp_path):
+        target = os.path.join(str(tmp_path), "target")
+        link = os.path.join(str(tmp_path), "link")
+        with open(target, "w") as f:
+            f.write("x")
+        os.symlink(target, link)
+        with os.scandir(str(tmp_path)) as it:
+            entries = {e.name: e for e in it}
+        assert fs.entry_kind(entries["link"]) == "symlink"
+
+    @pytest.mark.skipif(not hasattr(os, "mkfifo"),
+                        reason="os.mkfifo not available on this platform")
+    def test_fifo(self, tmp_path):
+        path = os.path.join(str(tmp_path), "fifo")
+        os.mkfifo(path)
+        with os.scandir(str(tmp_path)) as it:
+            entry = next(iter(it))
+        assert fs.entry_kind(entry) == "fifo"
+
+    @pytest.mark.skipif(not os.path.exists("/dev/null"),
+                        reason="no /dev/null on this platform")
+    def test_char_device(self):
+        entry = fs.PseudoDirEntry("/dev/null")
+        assert fs.entry_kind(entry) == "char_device"
+
+    @pytest.mark.skipif(not hasattr(socket, "AF_UNIX"),
+                        reason="AF_UNIX not available on this platform")
+    def test_socket(self):
+        import tempfile
+        # AF_UNIX paths are capped at ~104-108 bytes; pytest's tmp_path
+        # nests too deep on macOS, so use a short-path tempdir directly.
+        with tempfile.TemporaryDirectory() as d:
+            sock_path = os.path.join(d, "s")
+            with socket.socket(socket.AF_UNIX) as sock:
+                sock.bind(sock_path)
+                with os.scandir(d) as it:
+                    entry = next(iter(it))
+                assert fs.entry_kind(entry) == "socket"
+
+
+class TestRsyncExcludesUnsupportedEntries:
+    """
+    Regression tests: rsync() must classify FIFOs/sockets/devices via
+    lstat() alone (never opening them) and exclude them with a visible
+    EXCLUDE action, instead of either blocking forever (a FIFO with no
+    writer) or treating them as a fatal Actions.ERROR that aborts the
+    whole backup.
+    """
+
+    @pytest.mark.skipif(not hasattr(os, "mkfifo"),
+                        reason="os.mkfifo not available on this platform")
+    def test_fifo_without_writer_does_not_block(self, common_fs_dirs):
+        src_dir, dst_dir = common_fs_dirs
+        fifo_path = os.path.join(str(src_dir), "a_fifo")
+        os.mkfifo(fifo_path)
+
+        # No writer is ever opened on the other end. If rsync() opened
+        # this for reading, this test would hang instead of failing.
+        results = list(fs.rsync(str(src_dir), str(dst_dir)))
+
+        assert ("a_fifo", fs.Actions.EXCLUDE, "fifo") in results
+        assert not os.path.lexists(os.path.join(str(dst_dir), "a_fifo"))
+
+    @pytest.mark.skipif(not hasattr(socket, "AF_UNIX"),
+                        reason="AF_UNIX not available on this platform")
+    def test_socket_is_excluded_not_fatal(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as d:
+            src_dir = os.path.join(d, "src")
+            dst_dir = os.path.join(d, "dst")
+            os.mkdir(src_dir)
+            os.mkdir(dst_dir)
+            sock_path = os.path.join(src_dir, "a_socket")
+            with socket.socket(socket.AF_UNIX) as sock:
+                sock.bind(sock_path)
+                results = list(fs.rsync(src_dir, dst_dir))
+
+            assert ("a_socket", fs.Actions.EXCLUDE, "socket") in results
+            assert fs.Actions.ERROR not in [r[1] for r in results]
+            assert not os.path.lexists(os.path.join(dst_dir, "a_socket"))