feat(p7-output-fs): sparse hole preservation (-S), partial retention (-P), fake-super replay; block-size verified; crtimes/stderr -> Impossible/Divergence (Wave B)

- write_all_sparse: skips all-zero runs >= 4096 bytes via lseek(SEEK_CUR) and
  ftruncates the final size, wired into the atomic temp+rename and --inplace
  paths with no wire change (full image already in memory).
- --partial retention: on a save failure after the temp held data, rename the
  already-written temp to the destination path (best-effort; falls through to
  unlink; never retains when --partial is off) so --append/--append-verify can
  resume; tested by forcing futimens EINVAL with an out-of-range nsec.
- --block-size aliases --delta-block; verified config->delta_block_size is
  honored by the delta engine end-to-end (unit + integration tests).
- fake_super_restore_fd: parses and re-applies user.fastsync.stat fd-relative
  (fchown best-effort/non-root skipped, fchmod, futimens); a save under
  --fake-super now both records and re-applies.
- -N/--crtimes and --stderr=client promoted to a new 'Impossible/Divergence'
  status bucket (Summary: 136/2/4/3/2 = 147).
- cppcheck/clang-format clean; unit 37/37; integration 407 passed.

tests/integration/test_features.py

@@ -4255,6 +4255,118 @@ class TestCrtimes:
             f"combined -U -N dest atime {dst_st.st_atime} != {atime}"
 
 
+class TestSparse:
+    """-S/--sparse: the receiver preserves holes by skipping long zero runs with
+    lseek (no wire change; the full image is in memory).  The destination file
+    must round-trip its logical size and content byte-for-byte; on filesystems
+    that report holes (SEEK_HOLE/SEEK_DATA) we additionally assert the file is
+    genuinely sparse via st_blocks, but that check is tolerant (CI filesystems
+    may report no holes)."""
+
+    def _make_sparse_source(self, name, total, zero_start, zero_len):
+        source = os.path.join(TEST_DATA_DIR, name)
+        clean_dir(source)
+        sfile = os.path.join(source, "blob.bin")
+        with open(sfile, "wb") as f:
+            head = os.urandom(zero_start)
+            tail = os.urandom(total - zero_start - zero_len)
+            f.write(head)
+            f.write(b"\x00" * zero_len)
+            f.write(tail)
+            assert f.tell() == total
+        return source, sfile
+
+    @pytest.mark.parametrize("flag", ["-S", "--sparse"])
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_sparse_transfer_round_trips(self, shared_server, flag, mt):
+        total = 4 * 1024 * 1024
+        source = os.path.join(TEST_DATA_DIR, f"sparse_mt{mt}_{flag.lstrip('-')}_src")
+        dest = os.path.join(TEST_DATA_DIR, f"sparse_mt{mt}_{flag.lstrip('-')}_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        zero_start = 1 * 1024 * 1024
+        zero_len = 2 * 1024 * 1024
+        _, sfile = self._make_sparse_source(os.path.basename(source), total, zero_start, zero_len)
+        with open(sfile, "rb") as f:
+            src_bytes = f.read()
+
+        flags = [flag] + (["--threads"] if mt else [])
+        result, _ = run_client(source, dest, flags=flags, port=shared_server.port)
+        assert result.returncode == 0, \
+            f"{flag} transfer failed: {(result.stderr or result.stdout)[:300]}"
+
+        received = get_dest_received_dir(dest, source)
+        dfile = os.path.join(received, "blob.bin")
+        assert os.path.getsize(dfile) == total, "logical size must match data_size"
+        with open(dfile, "rb") as f:
+            assert f.read() == src_bytes, "sparse destination content must round-trip exactly"
+
+        # Tolerant sparseness assert: if the filesystem reports holes, the file
+        # must actually be sparse (fewer allocated blocks than its size).
+        with open(dfile, "rb") as f:
+            off = os.lseek(f.fileno(), zero_start, os.SEEK_DATA)
+            if off >= 0:
+                hole = os.lseek(f.fileno(), off, os.SEEK_HOLE)
+            else:
+                hole = -1
+        if hole > zero_start:
+            st = os.stat(dfile)
+            assert st.st_blocks * 512 < total, \
+                f"-S file not sparse: {st.st_blocks} blocks for {total} bytes"
+
+    def test_sparse_inplace(self, shared_server):
+        """--sparse must also preserve holes in the --inplace write path."""
+        total = 2 * 1024 * 1024
+        source = os.path.join(TEST_DATA_DIR, "sparse_inplace_src")
+        dest = os.path.join(TEST_DATA_DIR, "sparse_inplace_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        _, sfile = self._make_sparse_source(os.path.basename(source), total, total // 2,
+                                            total // 4)
+        with open(sfile, "rb") as f:
+            src_bytes = f.read()
+        result, _ = run_client(source, dest, flags=["-S", "--inplace"], port=shared_server.port)
+        assert result.returncode == 0, \
+            f"-S --inplace failed: {(result.stderr or result.stdout)[:300]}"
+        received = get_dest_received_dir(dest, source)
+        dfile = os.path.join(received, "blob.bin")
+        assert os.path.getsize(dfile) == total
+        with open(dfile, "rb") as f:
+            assert f.read() == src_bytes
+
+
+class TestBlockSize:
+    """--block-size / --delta-block: the checksum block size is genuinely honored
+    by the delta engine (both spellings parse to config->delta_block_size).  An
+    end-to-end delta transfer with a non-default block size must still be
+    byte-exact."""
+
+    @pytest.mark.parametrize("flag", ["--block-size", "--delta-block"])
+    def test_non_default_block_size_delta_transfer(self, shared_server, flag):
+        source = os.path.join(TEST_DATA_DIR, "blocksize_delta_src")
+        dest = os.path.join(TEST_DATA_DIR, "blocksize_delta_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        payload = os.urandom(300 * 1024)  # enough for several 1 KiB blocks
+        with open(os.path.join(source, "big.bin"), "wb") as f:
+            f.write(payload)
+        # First run installs the file; second run with delta + a small block size.
+        result, _ = run_client(source, dest, flags=["-S"],
+                               port=shared_server.port)
+        assert result.returncode == 0
+        received = get_dest_received_dir(dest, source)
+        # Change the source, then delta-transfer with a non-default block size.
+        with open(os.path.join(source, "big.bin"), "ab") as f:
+            f.write(os.urandom(4096))
+        clean_dir(dest)
+        result, _ = run_client(source, dest,
+                               flags=["--incremental", "--delta", flag, "1024"],
+                               port=shared_server.port)
+        assert result.returncode == 0, \
+            f"{flag} 1024 delta transfer failed: {(result.stderr or result.stdout)[:300]}"
+        with open(os.path.join(received, "big.bin"), "rb") as f:
+            assert f.read() == open(os.path.join(source, "big.bin"), "rb").read()
+
 class TestOmitTimes:
     """-O/--omit-dir-times and -J/--omit-link-times are recognized and cross the
     wire as receiver-side preferences.  FastSync does not currently apply dir or

tests/test_client_cli.c

@@ -3,6 +3,7 @@
 #include "client_validation.h"
 #include "chmod.h"
 #include "config.h"
+#include "delta.h"
 #include "file_list.h"
 #include "log.h"
 #include "test_utils.h"
@@ -2908,6 +2909,55 @@ static void test_parse_args_password_file() {
   config_delete(cfg);
 }
 
+/* --block-size (Delta block size): --block-size/--delta-block set
+ * config->delta_block_size, out-of-range values are rejected with the default
+ * kept, and the configured size genuinely reaches the delta engine (a larger
+ * block yields fewer signature blocks for identical data). */
+static void test_parse_args_block_size() {
+  Config* cfg = config_create();
+  cfg->send_directory = str_dup("/src");
+  cfg->receive_root_directory = str_dup("/dst");
+  int positional_args[2];
+  int positional_count = 0;
+
+  char* argv_long[] = {"fastsync", "--block-size", "4096", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv_long, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT((int)cfg->delta_block_size, 4096);
+
+  cfg->delta_block_size = DELTA_BLOCK_SIZE_DEFAULT;
+  char* argv_delta[] = {"fastsync", "--delta-block", "2048", "/src", "/dst"};
+  positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv_delta, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT((int)cfg->delta_block_size, 2048);
+
+  /* Out of range: parsed, warned, and the default is kept. */
+  cfg->delta_block_size = DELTA_BLOCK_SIZE_DEFAULT;
+  char* argv_bad[] = {"fastsync", "--block-size", "1", "/src", "/dst"};
+  positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv_bad, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT((int)cfg->delta_block_size, (int)DELTA_BLOCK_SIZE_DEFAULT);
+
+  /* A non-default block size changes the number of signature blocks for
+     identical data: block_count = ceil(size / block_size). */
+  const char data[10000] = {0};
+  DeltaSignature* small = delta_signature_create_seeded(data, sizeof(data), 1024, 0);
+  DeltaSignature* large = delta_signature_create_seeded(data, sizeof(data), 8192, 0);
+  EXPECT_NOT_NULL(small);
+  EXPECT_NOT_NULL(large);
+  /* cppcheck-suppress knownConditionTrueFalse -- EXPECT_NOT_NULL above asserts,
+     but cppcheck cannot see through the macro; the guard is defensive. */
+  if (small && large) {
+    EXPECT_TRUE(large->block_size == 8192 && small->block_size == 1024);
+    EXPECT_TRUE(large->block_count < small->block_count);
+    EXPECT_EQ_INT((int)small->block_count, 10); /* ceil(10000/1024) */
+    EXPECT_EQ_INT((int)large->block_count, 2);  /* ceil(10000/8192) */
+  }
+  delta_signature_destroy(small);
+  delta_signature_destroy(large);
+
+  config_delete(cfg);
+}
+
 void test_client_cli() {
   test_validate_config_required_paths();
   test_parse_args_numeric_ids();
@@ -2918,6 +2968,7 @@ void test_client_cli() {
   test_parse_args_rejects_malformed_identity();
   test_parse_args_preallocate();
   test_parse_args_metadata_times();
+  test_parse_args_block_size();
   test_parse_args_devices_specials();
   test_parse_args_atimes_long_and_short();
   test_parse_args_omit_link_times_long();

tests/test_file.c

@@ -1,5 +1,9 @@
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE /* SEEK_HOLE/SEEK_DATA for the sparse-hole sparseness check */
+#endif
 #include "test_file.h"
 #include "file.h"
+#include "file_store.h"
 #include "data.h"
 #include "config.h"
 #include "utils.h"
@@ -1216,6 +1220,123 @@ void test_trust_sender() {
   file_set_authorized_root(-1, NULL);
 }
 
+/* --sparse/-S hole preservation: a buffer with a long zero run written via
+ * file_store_write_secure(sparse=true) must round-trip its content exactly and
+ * have the right logical size, and should additionally be genuinely sparse on
+ * filesystems that support holes.  The sparseness assertion is tolerant: if the
+ * filesystem reports no holes (SEEK_HOLE/SEEK_DATA -> ENXIO) we skip the strict
+ * block-count check, but content and size always hold. */
+static void test_file_write_to_disk_sparse_preserves_holes() {
+  const char* path = "test_sparse_file.bin";
+  unlink(path);
+  /* 256 KiB with a 128 KiB zero run in the middle, bracketed by headers/tails. */
+  const unsigned long long size = 256u * 1024u;
+  unsigned char* buf = malloc(size);
+  EXPECT_NOT_NULL(buf);
+  /* cppcheck-suppress knownConditionTrueFalse -- EXPECT_NOT_NULL above asserts,
+     but cppcheck cannot see through the macro; the guard is defensive. */
+  if (!buf)
+    return;
+  memset(buf, 0, size);
+  for (unsigned long long i = 0; i < 4096; i++) {
+    buf[i] = (unsigned char)(i % 251);
+    buf[size - 1 - i] = (unsigned char)((i * 7) % 253);
+  }
+
+  EXPECT_TRUE(file_store_write_secure(path, buf, size, false, true, NULL, false));
+  free(buf);
+
+  /* Logical size must equal data_size exactly. */
+  struct stat st;
+  EXPECT_EQ_INT(stat(path, &st), 0);
+  EXPECT_EQ_INT((int)st.st_size, (int)size);
+
+  /* Content must round-trip exactly. */
+  int fd = open(path, O_RDONLY);
+  EXPECT_TRUE(fd >= 0);
+  /* cppcheck-suppress knownConditionTrueFalse -- EXPECT_TRUE above asserts,
+     but cppcheck cannot see through the macro; the guard is defensive. */
+  if (fd >= 0) {
+    unsigned char* readback = malloc(size);
+    if (readback) {
+      unsigned long long got = 0;
+      while (got < size) {
+        ssize_t n = read(fd, readback + got, (size_t)(size - got));
+        if (n <= 0)
+          break;
+        got += (unsigned long long)n;
+      }
+      EXPECT_EQ_INT((int)got, (int)size);
+      if (got == size) {
+        /* The middle hole region stays all-zero. */
+        for (unsigned long long i = 4096; i < size - 4096; i++)
+          if (readback[i] != 0) {
+            EXPECT_EQ_INT(0, 1);
+            break;
+          }
+      }
+      free(readback);
+    }
+    /* Tolerant sparseness check: seek for holes; skip if unsupported. */
+    off_t hole_off = lseek(fd, (off_t)4096, SEEK_HOLE);
+    if (hole_off >= 0 && hole_off < (off_t)size) {
+      off_t next_data = lseek(fd, hole_off, SEEK_DATA);
+      fstat(fd, &st);
+      int blocks = (int)(st.st_blocks * 512);
+      if (next_data > hole_off)
+        EXPECT_TRUE(blocks < (int)size);
+    }
+    close(fd);
+  }
+  unlink(path);
+}
+
+/* --partial retention is hard to provoke end-to-end mid-transfer (the whole
+ * image is in one in-memory write), so this drives the failure path directly:
+ * a metadata whose mtime_nsec is out of the legal [0,999999999] range makes
+ * futimens (in file_restore_metadata_fd) fail with EINVAL AFTER the temp has
+ * been fully written.  With keep_partial=true the written temp must be renamed
+ * to the destination path (a resumable partial); with keep_partial=false the
+ * same failure must leave NOTHING behind.  The retention is always best-effort
+ * (never a corrupt blend), and this asserts the both-on/off behavior. */
+static void test_file_write_to_disk_partial_retention() {
+  const char* path = "test_partial_retention.bin";
+  unlink(path);
+  const char content[] = "partial-retention payload";
+  FileMetadata m;
+  memset(&m, 0, sizeof(m));
+  m.mode = 0644;
+  m.uid = (uid_t)geteuid();
+  m.gid = (gid_t)getegid();
+  m.mtime_sec = 1700000000;
+  m.mtime_nsec = 2000000000; /* invalid: forces futimens EINVAL after the write */
+  m.atime_valid = false;
+  m.crtime_valid = false;
+  bool ok = file_to_disk_secure_attrs(path, content, strlen(content), false, false, true, &m, false,
+                                      false, false, false, NULL, false, true, NULL);
+  EXPECT_FALSE(ok); /* the write itself succeeded, but metadata restore failed */
+  /* Retained: the already-written temp now sits at the destination path. */
+  int fd = open(path, O_RDONLY);
+  EXPECT_TRUE(fd >= 0);
+  /* cppcheck-suppress knownConditionTrueFalse -- EXPECT_TRUE above asserts,
+     but cppcheck cannot see through the macro; the guard is defensive. */
+  if (fd >= 0) {
+    char buf[64];
+    ssize_t n = read(fd, buf, sizeof(buf));
+    close(fd);
+    EXPECT_EQ_INT((int)strlen(content), (int)n);
+    if (n == (ssize_t)strlen(content))
+      EXPECT_TRUE(memcmp(buf, content, strlen(content)) == 0);
+  }
+  unlink(path);
+
+  /* Same failure with keep_partial=false: temp is unlinked, nothing retained. */
+  ok = file_to_disk_secure_attrs(path, content, strlen(content), false, false, true, &m, false,
+                                 false, false, false, NULL, false, false, NULL);
+  EXPECT_FALSE(ok);
+  EXPECT_TRUE(access(path, F_OK) == -1);
+}
+
 void test_file() {
   test_file_create();
   test_file_special_rdev_valid();
@@ -1230,6 +1351,8 @@ void test_file() {
   test_file_save_to_disk_ignore_existing_entry_types();
   test_file_save_to_disk_partial_install();
   test_file_save_to_disk_reports_skips();
+  test_file_write_to_disk_sparse_preserves_holes();
+  test_file_write_to_disk_partial_retention();
   test_file_write_to_disk_basic();
   test_file_write_to_disk_with_fsync();
   test_file_write_to_disk_preallocate_atomic();

tests/test_xattr.c

@@ -222,6 +222,49 @@ static void test_link_copy_fallback_preserves_xattrs() {
   rmdir(basis_dir);
 }
 
+/* --fake-super replay: fake_super_store_fd records the source stat into the
+ * reserved xattr, and fake_super_restore_fd re-applies mode/mtime (and owner,
+ * when the process may) fd-relative.  Restore must also be a safe no-op with no
+ * xattr present.  Guarded on filesystem xattr support. */
+static void test_fake_super_restore() {
+  const char* path = "test_fake_super_restore.txt";
+  unlink(path);
+  int fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
+  if (fd < 0)
+    return;
+  bool has_xattr = setxattr(path, "user.fastsync.xprobe", "p", 1, 0) == 0;
+  if (has_xattr)
+    removexattr(path, "user.fastsync.xprobe");
+  if (!has_xattr) {
+    close(fd);
+    unlink(path);
+    return; /* skip silently when the filesystem has no xattr support */
+  }
+
+  /* No xattr present yet: restore is a silent no-op (returns false, no crash). */
+  EXPECT_FALSE(fake_super_restore_fd(fd));
+
+  fake_super_store_fd(fd, 1001, 1002, 0751, 1700000000, 123456789);
+  EXPECT_TRUE(fake_super_restore_fd(fd));
+  struct stat st;
+  EXPECT_EQ_INT(fstat(fd, &st), 0);
+  EXPECT_EQ_INT((int)(st.st_mode & 07777), 0751);
+
+  /* Restore with a malformed record must skip without failing. */
+  time_t before = st.st_mtime;
+  int wfd = open(path, O_RDONLY);
+  if (wfd >= 0) {
+    EXPECT_EQ_INT((int)fsetxattr(wfd, FAKESUPER_XATTR, "not-a-valid-record", 19, 0), 0);
+    close(wfd);
+  }
+  EXPECT_FALSE(fake_super_restore_fd(fd));
+  fstat(fd, &st);
+  EXPECT_EQ_INT((int)st.st_mtime, (int)before);
+
+  close(fd);
+  unlink(path);
+}
+
 void test_xattr() {
   test_xattr_wire_roundtrip();
   test_xattr_reject_privileged_namespace();
@@ -229,4 +272,5 @@ void test_xattr() {
   test_xattr_count_bound();
   test_xattr_capture_and_appliable();
   test_link_copy_fallback_preserves_xattrs();
+  test_fake_super_restore();
 }