fix(parity): actual-removal delete-delay counts, rsync-accurate stats/progress/%C

- --delete-delay: count/track only entries actually removed; a directory
  refilled before commit (ENOTEMPTY) no longer inflates Number of deleted
  files or the --max-delete budget (unit + integration + rsync differential).
- --stats: per-type Number of files breakdown; only stored regular files
  count as transferred; transferred/literal byte totals and Total file size
  (symlink target lengths) now match rsync for whole-file transfers.
- --progress: print the leading ./ root line and include the root entry in
  the to-chk denominator (single-file output byte-identical to rsync).
- --out-format %C: use the selected transfer checksum and render every
  algorithm exactly like rsync; checksum_digest_file gains md4/sha1/none.
  Reclassify --out-format to Divergent (protocol-specific %b/delta-%c).
- Docs: RSYNC_COMPAT tally 107/25/24, HANDOFF update. No wire change.

tests/integration/test_delete_timing_parity.py

@@ -313,6 +313,91 @@ class TestDeleteTimingFailure:
                 )
 
 
+class TestDeleteDelayDeletedCount:
+    """The reported deleted count must reflect entries actually removed."""
+
+    def test_refilled_deferred_dir_is_not_counted(self):
+        """A directory snapshotted into a --delete-delay plan that is refilled
+        before the commit survives ENOTEMPTY and must NOT inflate "Number of
+        deleted files" (regression for delete_plan.c counting at snapshot)."""
+        source = os.path.join(TEST_DATA_DIR, "ddc_src")
+        dest = os.path.join(TEST_DATA_DIR, "ddc_dst")
+        clean_dir(source)
+        clean_dir(dest)
+        _write(os.path.join(source, "d", "keep.txt"), b"kept payload\n")
+        _write(os.path.join(source, "d", "big.bin"), b"B" * BIG_BYTES)
+        received = get_dest_received_dir(dest, source)
+        extra_dir = os.path.join(received, "d", "extradir")
+        os.makedirs(extra_dir, exist_ok=True)
+
+        def hook():
+            # Runs while big.bin is in flight, after d's delete plan was processed.
+            _write(os.path.join(extra_dir, "new.txt"), b"created mid-transfer\n")
+
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            proxy = _SlicingProxy(server.port, hook=hook, hook_after=MID_TRANSFER_BYTES,
+                                  throttle=PROXY_THROTTLE, wait_for_reply=True)
+            flags = ["--delete-delay", "--incremental", "--ignore-times", "--stats"]
+            result, _ = run_client(source, dest, flags=flags, port=proxy.port)
+            proxy.finish()
+        assert result.returncode == 0, (result.stderr or result.stdout)[:400]
+        assert proxy.hook_called.is_set(), "hook never fired"
+        assert os.path.exists(os.path.join(extra_dir, "new.txt")), "late file vanished"
+        deleted = None
+        for line in result.stdout.splitlines():
+            if line.startswith("Number of deleted files:"):
+                deleted = int(line.split(":", 1)[1].split()[0])
+        assert deleted == 0, (deleted, result.stdout)
+
+
+class TestDeleteDelayMaxDeleteParity:
+    """--max-delete with --delete-delay: a partial deletion still reports the
+    number of entries actually removed, matching rsync (the exact surviving set
+    can differ; only the count is compared)."""
+
+    @requires_rsync
+    def test_max_delete_count_matches_rsync(self):
+        source = os.path.join(TEST_DATA_DIR, "ddm_src")
+        rsync_dst = os.path.join(TEST_DATA_DIR, "ddm_rsync_dst")
+        clean_dir(source)
+        clean_dir(rsync_dst)
+        _write(os.path.join(source, "d", "keep.txt"), b"keep\n")
+        for i in range(1, 6):
+            _write(os.path.join(rsync_dst, "d", f"e{i}.txt"), f"extra{i}\n".encode())
+
+        rsync_result = _rsync(["-a", "--delete-delay", "--max-delete=2", "--stats",
+                               source + "/", rsync_dst + "/"])
+        # rsync exits 25 ("the --max-delete limit stopped deletions").
+        assert rsync_result.returncode == 25, rsync_result.stderr
+        rsync_count = _deleted_count(rsync_result.stdout)
+        assert rsync_count == 2, rsync_result.stdout
+
+        dest = os.path.join(TEST_DATA_DIR, "ddm_dst")
+        clean_dir(dest)
+        received = get_dest_received_dir(dest, source)
+        for i in range(1, 6):
+            _write(os.path.join(received, "d", f"e{i}.txt"), f"extra{i}\n".encode())
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(
+                source, dest,
+                flags=["--delete-delay", "--max-delete=2", "--stats"],
+                port=server.port,
+            )
+        # A capped --max-delete commit is a successful transfer that both tools
+        # report with exit 25.
+        assert result.returncode == 25, (result.stderr or result.stdout)[:300]
+        assert _deleted_count(result.stdout) == rsync_count, result.stdout
+
+
+def _deleted_count(text):
+    for line in text.splitlines():
+        if line.startswith("Number of deleted files:"):
+            return int(line.split(":", 1)[1].split()[0])
+    return None
+
+
 class TestDeleteDelayVsAfterSnapshot:
     """A destination entry created after its directory's scan survives under
     --delete-delay but is removed by --delete-after's fresh end scan."""

tests/integration/test_output_parity.py

@@ -447,6 +447,43 @@ class TestWireStatsParity:
         assert fast_frames[0] == rsync_frames[0], (rsync_frames[0], fast_frames[0])
         assert "(xfr#1," in fast_frames[-1], fast_frames[-1]
 
+    @requires_rsync
+    @pytest.mark.ci
+    def test_progress_leading_root_line_and_to_chk_match_rsync(self, shared_server):
+        """A single-file transfer: rsync emits the transfer-root `./` name line
+        and a `to-chk=0/2` denominator that counts that root entry.  Both must
+        match FastSync byte-for-byte for the deterministic frames."""
+        source = os.path.join(TEST_DATA_DIR, "wire_pgroot_src")
+        dest = os.path.join(TEST_DATA_DIR, "wire_pgroot_dst")
+        rdst = os.path.join(TEST_DATA_DIR, "wire_pgroot_rdst")
+        _make_one_file(source, "f.bin", 100)
+        clean_dir(dest)
+        # rsync prints the `./` root line only when the transfer root itself is
+        # created, so make the rsync destination absent.  The "created directory"
+        # line it then emits has no FastSync counterpart (different mirror
+        # layout), so only the name/frame lines are compared.
+        shutil.rmtree(rdst, ignore_errors=True)
+        rsync_result = _rsync(["-a", "--progress", source + "/", rdst + "/"])
+        assert rsync_result.returncode == 0, rsync_result.stderr
+        result, _ = run_client(source, dest, flags=["-a", "--progress"],
+                               port=shared_server.port)
+        assert result.returncode == 0, result.stderr[:300]
+
+        # subprocess text mode normalizes \r to \n (universal newlines).
+        def lines_of(text):
+            return [ln for ln in text.splitlines() if ln and not ln.startswith("created directory")]
+
+        rsync_lines = lines_of(rsync_result.stdout)
+        fast_lines = lines_of(result.stdout)
+        rsync_names = [ln for ln in rsync_lines if "%" not in ln]
+        fast_names = [ln for ln in fast_lines if "%" not in ln]
+
+        assert rsync_names == ["sending incremental file list", "./", "f.bin"], rsync_names
+        assert fast_names == rsync_names, (rsync_names, fast_names)
+        # The final frame's to-chk denominator must include the source-root entry.
+        assert "to-chk=0/2" in fast_lines[-1], fast_lines[-1]
+        assert fast_lines[-1] == rsync_lines[-1], (rsync_lines[-1], fast_lines[-1])
+
     @requires_rsync
     @pytest.mark.ci
     @pytest.mark.parametrize("mt", [False, True])
@@ -488,18 +525,12 @@ class TestWireStatsParity:
 
     @requires_rsync
     @pytest.mark.ci
-    def test_stats_file_count_breakdown_residual(self, shared_server):
-        """Residual (row #3): rsync prints the `Number of files` and
-        `Number of created files` lines with a per-type breakdown
-        (`(reg: X, dir: Y, link: Z)`).
-
-        FastSync cannot reproduce it from what the sender currently knows: the
-        scanner does not put directory entries in the transfer list (directories
-        are created implicitly), and without a per-entry destination-probe the
-        sender cannot tell which entries the receiver newly created.  So FastSync
-        prints the bare transferred-entry count.  This test pins the divergence
-        explicitly -- the row must not be marked ✅.
-        """
+    def test_stats_file_count_breakdown_matches_rsync(self, shared_server):
+        """`Number of files` now carries rsync's per-type breakdown: the scanner
+        accounts directory entries (captured for -a/-t/-p) plus reg/link/special
+        from the transfer list.  `Number of created files` still lacks the type
+        breakdown (FastSync cannot tell which entries the receiver newly
+        created), so that residual is pinned separately."""
         source = os.path.join(TEST_DATA_DIR, "wire_stc_src")
         dest = os.path.join(TEST_DATA_DIR, "wire_stc_dst")
         rdst = os.path.join(TEST_DATA_DIR, "wire_stc_rdst")
@@ -523,15 +554,44 @@ class TestWireStatsParity:
         f_files = stats_line(result.stdout, "Number of files")
         f_created = stats_line(result.stdout, "Number of created files")
 
-        # rsync always carries the type breakdown (the source root counts as a
-        # directory; the single regular file as reg).
         assert re.match(r"Number of files: 2 \(reg: 1, dir: 1\)$", r_files), r_files
+        assert r_files == f_files, (r_files, f_files)
+        # rsync always carries the created type breakdown; FastSync prints the
+        # bare transferred-regular count (documented residual).
         assert re.match(r"Number of created files: 1 \(reg: 1\)$", r_created), r_created
-        # FastSync prints only the bare count: no directory accounting and no
-        # per-entry "created" knowledge.
-        assert re.fullmatch(r"Number of files: 1", f_files), f_files
         assert re.fullmatch(r"Number of created files: 1", f_created), f_created
 
+    @requires_rsync
+    @pytest.mark.ci
+    @pytest.mark.parametrize("choice", ["xxh128", "xxh64", "xxh3", "md5", "md4", "sha1", "none"])
+    def test_out_format_C_selected_algorithm_matches_rsync(self, shared_server, choice):
+        """`%C` must use the algorithm selected by --checksum-choice, not always
+        xxh128, and render it exactly like rsync (big-endian for the 64-bit
+        hashes, high-then-low for xxh128, standard hex for md5/md4/sha1)."""
+        source = os.path.join(TEST_DATA_DIR, f"wire_cc_{choice}_src")
+        dest = os.path.join(TEST_DATA_DIR, f"wire_cc_{choice}_dst")
+        rdst = os.path.join(TEST_DATA_DIR, f"wire_cc_{choice}_rdst")
+        _make_one_file(source, "f.bin", 200000)
+        clean_dir(dest)
+        clean_dir(rdst)
+        fmt = "%C %l %n"
+        rsync_result = _rsync(["-a", "--checksum-choice=" + choice,
+                               "--out-format=" + fmt, source + "/", rdst + "/"])
+        assert rsync_result.returncode == 0, rsync_result.stderr
+        result, _ = run_client(source, dest,
+                               flags=["-a", "--checksum-choice=" + choice,
+                                      "--out-format=" + fmt],
+                               port=shared_server.port)
+        assert result.returncode == 0, result.stderr[:300]
+
+        def file_lines(text):
+            return [line for line in text.splitlines()
+                    if line and not line.rsplit(" ", 1)[-1].endswith("/")]
+
+        assert file_lines(result.stdout) == file_lines(rsync_result.stdout), (
+            f"choice={choice}: rsync={rsync_result.stdout!r} fastsync={result.stdout!r}"
+        )
+
     @requires_rsync
     @pytest.mark.ci
     @pytest.mark.parametrize("mt", [False, True])

tests/runner.c

@@ -11,6 +11,7 @@
 #include "test_daemon_conf.h"
 #include "test_daemon_limits.h"
 #include "test_delay_updates.h"
+#include "test_delete_plan.h"
 #include "test_delta.h"
 #include "test_file.h"
 #include "test_file_list.h"
@@ -66,6 +67,7 @@ int main() {
   RUN_TEST(test_scanner);
   RUN_TEST(test_checksum);
   RUN_TEST(test_delta);
+  RUN_TEST(test_delete_plan);
   RUN_TEST(test_data);
   RUN_TEST(test_protocol);
   RUN_TEST(test_protocol_error);

tests/test_change_list.c

@@ -177,6 +177,40 @@ static void test_change_list_enabled() {
   config_delete(config); /* closes config->log_file */
 }
 
+/* %C uses the negotiated TRANSFER checksum's column width (not the pre-transfer
+ * whole-file digest), and `none` renders as a blank 2-char column, matching
+ * rsync.  A not-yet-filled checksum renders as spaces. */
+static void test_format_C_padding_uses_transfer_algo() {
+  ChangeEvent event = sample_event();
+  Config* config = config_create();
+  EXPECT_NOT_NULL(config);
+  /* Deliberately different pre-transfer algorithm: the transfer one must win. */
+  config->checksum_algo = (int)CHECKSUM_ALGO_MD4;
+  struct {
+    int algo;
+    int width;
+  } cases[] = {
+      {CHECKSUM_ALGO_XXH128, 32}, {CHECKSUM_ALGO_XXH64, 16}, {CHECKSUM_ALGO_XXH3, 16},
+      {CHECKSUM_ALGO_MD5, 32},    {CHECKSUM_ALGO_MD4, 32},   {CHECKSUM_ALGO_SHA1, 40},
+      {CHECKSUM_ALGO_NONE, 2},
+  };
+  for (size_t i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
+    config->checksum_transfer_algo = cases[i].algo;
+    char expected[64];
+    size_t n = 0;
+    expected[n++] = '[';
+    for (int j = 0; j < cases[i].width; j++)
+      expected[n++] = ' ';
+    expected[n++] = ']';
+    expected[n] = '\0';
+    char* line = change_render_format("[%C]", config, &event);
+    EXPECT_NOT_NULL(line);
+    EXPECT_EQ_STR(line, expected);
+    free(line);
+  }
+  config_delete(config);
+}
+
 void test_change_list() {
   test_format_tokens();
   test_format_unknown_tokens_preserved();
@@ -188,4 +222,5 @@ void test_change_list() {
   test_render_itemize_up_to_date_is_empty();
   test_render_list_line();
   test_change_list_enabled();
+  test_format_C_padding_uses_transfer_algo();
 }

tests/test_checksum.c

@@ -1,7 +1,9 @@
 #include "test_checksum.h"
 #include "checksum.h"
 #include "test_utils.h"
+#include <stdlib.h>
 #include <string.h>
+#include <unistd.h>
 
 /* Known xxHash64 vector (seed 0) for the empty string and a literal.
  * The md5 vectors are the standard NIST/RFC1321 test strings.  These pin the
@@ -230,6 +232,39 @@ static void test_checksum_null_empty_digest() {
   EXPECT_TRUE(memcmp(a, b, alen) == 0);
 }
 
+/* Every algorithm checksum_digest_file() claims to support must produce the
+ * SAME digest as the in-memory one-shot, including the newly added md4/sha1/
+ * none.  A 200000-byte payload forces several 64 KiB streaming reads. */
+static void test_checksum_digest_file_matches_oneshot(void) {
+  static const ChecksumAlgo algos[] = {
+      CHECKSUM_ALGO_XXH64, CHECKSUM_ALGO_XXH3, CHECKSUM_ALGO_XXH128, CHECKSUM_ALGO_MD5,
+      CHECKSUM_ALGO_MD4,   CHECKSUM_ALGO_SHA1, CHECKSUM_ALGO_NONE,
+  };
+  enum { SIZE = 200000 };
+  uint8_t* data = malloc(SIZE);
+  EXPECT_NOT_NULL(data);
+  for (int i = 0; i < SIZE; i++)
+    data[i] = (uint8_t)((i * 7 + 3) & 0xff);
+  char path[] = "/tmp/fastsync_ck_XXXXXX";
+  int fd = mkstemp(path);
+  EXPECT_TRUE(fd >= 0);
+  ssize_t written = write(fd, data, SIZE);
+  close(fd);
+  EXPECT_EQ_INT((int)written, SIZE);
+  for (size_t a = 0; a < sizeof(algos) / sizeof(algos[0]); a++) {
+    ChecksumAlgo algo = algos[a];
+    uint8_t one[CHECKSUM_MAX_DIGEST_LEN];
+    uint8_t file[CHECKSUM_MAX_DIGEST_LEN];
+    size_t one_len = 0, file_len = 0;
+    EXPECT_TRUE(checksum_digest(algo, 0, data, SIZE, one, sizeof(one), &one_len));
+    EXPECT_TRUE(checksum_digest_file(algo, 0, path, file, sizeof(file), &file_len));
+    EXPECT_EQ_INT((int)file_len, (int)one_len);
+    EXPECT_TRUE(memcmp(one, file, one_len) == 0);
+  }
+  unlink(path);
+  free(data);
+}
+
 void test_checksum(void) {
   test_checksum_xxh64_seed0();
   test_checksum_xxh64_empty();
@@ -245,4 +280,5 @@ void test_checksum(void) {
   test_checksum_xxh3_xxh128();
   test_checksum_truncated_buffer_rejected();
   test_checksum_null_empty_digest();
+  test_checksum_digest_file_matches_oneshot();
 }

tests/test_delete_plan.c

@@ -0,0 +1,156 @@
+#include "test_delete_plan.h"
+#include "charset.h"
+#include "config.h"
+#include "delete_plan.h"
+#include "protocol.h"
+#include "test_utils.h"
+#include "utils.h"
+#include <fcntl.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+/* Send one STATUS_DELETE_PLAN body (the leading status is consumed by the
+ * caller/receiver entry point) describing `dir` with no kept children. */
+static void send_plan_frame(int fd, const char* dir) {
+  EXPECT_TRUE(send_int(fd, 0)); /* has_config */
+  EXPECT_TRUE(send_wire_str(fd, dir));
+  EXPECT_TRUE(send_int(fd, 0)); /* kept child dirs */
+  EXPECT_TRUE(send_int(fd, 0)); /* kept child files */
+}
+
+/* --delete-delay: a directory snapshotted into the plan that is refilled before
+ * the commit must NOT be counted as deleted once its unlink fails ENOTEMPTY.
+ * Regression for delete_plan.c counting at snapshot (defer_add) instead of at
+ * the actual removal. */
+static void test_delete_delay_refilled_dir_not_counted(void) {
+  char root[] = "/tmp/fastsync_dp_refill_XXXXXX";
+  EXPECT_TRUE(mkdtemp(root) != NULL);
+  char extra[1024];
+  snprintf(extra, sizeof(extra), "%s/extra", root);
+  EXPECT_EQ_INT(mkdir(extra, 0700), 0);
+
+  Config* config = config_create();
+  EXPECT_NOT_NULL(config);
+  config->receive_root_directory = str_dup(root);
+  config->use_delete = true;
+  config->delete_delay = true;
+
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+
+  DeletePlanSession* session = delete_plan_session_create(config);
+  EXPECT_NOT_NULL(session);
+  send_plan_frame(p[1], ".");
+  EXPECT_EQ_INT(delete_plan_session_receive(session, config, p[0]), 0);
+  /* The empty extra directory was snapshotted, not removed yet. */
+  EXPECT_EQ_INT((int)delete_plan_session_deleted(session), 0);
+
+  /* Refill the directory while the deferred commit is pending. */
+  char refill[1200];
+  snprintf(refill, sizeof(refill), "%s/new.txt", extra);
+  int fd = open(refill, O_WRONLY | O_CREAT | O_TRUNC, 0600);
+  EXPECT_TRUE(fd >= 0);
+  close(fd);
+
+  EXPECT_EQ_INT(delete_plan_session_commit(session, config), DELETE_COMMIT_OK);
+  /* ENOTEMPTY: the directory survives, so it must not be reported as deleted. */
+  EXPECT_EQ_INT((int)delete_plan_session_deleted(session), 0);
+  struct stat st;
+  EXPECT_EQ_INT(lstat(extra, &st), 0);
+
+  delete_plan_session_destroy(session);
+  close(p[0]);
+  close(p[1]);
+  unlink(refill);
+  rmdir(extra);
+  rmdir(root);
+  config_delete(config);
+}
+
+/* The complement: a deferred regular extra that DOES get removed is counted. */
+static void test_delete_delay_removed_file_counted(void) {
+  char root[] = "/tmp/fastsync_dp_file_XXXXXX";
+  EXPECT_TRUE(mkdtemp(root) != NULL);
+  char extra[1024];
+  snprintf(extra, sizeof(extra), "%s/extra.txt", root);
+  int fd = open(extra, O_WRONLY | O_CREAT | O_TRUNC, 0600);
+  EXPECT_TRUE(fd >= 0);
+  close(fd);
+
+  Config* config = config_create();
+  EXPECT_NOT_NULL(config);
+  config->receive_root_directory = str_dup(root);
+  config->use_delete = true;
+  config->delete_delay = true;
+
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+
+  DeletePlanSession* session = delete_plan_session_create(config);
+  EXPECT_NOT_NULL(session);
+  send_plan_frame(p[1], ".");
+  EXPECT_EQ_INT(delete_plan_session_receive(session, config, p[0]), 0);
+  EXPECT_EQ_INT((int)delete_plan_session_deleted(session), 0);
+  EXPECT_EQ_INT(delete_plan_session_commit(session, config), DELETE_COMMIT_OK);
+  EXPECT_EQ_INT((int)delete_plan_session_deleted(session), 1);
+  EXPECT_TRUE(lstat(extra, &(struct stat){0}) != 0);
+
+  delete_plan_session_destroy(session);
+  close(p[0]);
+  close(p[1]);
+  rmdir(root);
+  config_delete(config);
+}
+
+/* --max-delete still bounds the deferred plan; the actual (removed) count must
+ * not exceed the limit even though more extras existed. */
+static void test_delete_delay_max_delete_bounds_actual(void) {
+  char root[] = "/tmp/fastsync_dp_max_XXXXXX";
+  EXPECT_TRUE(mkdtemp(root) != NULL);
+  for (int i = 0; i < 3; i++) {
+    char path[1024];
+    snprintf(path, sizeof(path), "%s/e%d.txt", root, i);
+    int fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
+    EXPECT_TRUE(fd >= 0);
+    close(fd);
+  }
+
+  Config* config = config_create();
+  EXPECT_NOT_NULL(config);
+  config->receive_root_directory = str_dup(root);
+  config->use_delete = true;
+  config->delete_delay = true;
+  config->max_delete = 1;
+
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+
+  DeletePlanSession* session = delete_plan_session_create(config);
+  EXPECT_NOT_NULL(session);
+  send_plan_frame(p[1], ".");
+  EXPECT_EQ_INT(delete_plan_session_receive(session, config, p[0]), 0);
+  EXPECT_TRUE(delete_plan_session_limit_reached(session));
+  EXPECT_EQ_INT(delete_plan_session_commit(session, config), DELETE_COMMIT_LIMIT_REACHED);
+  EXPECT_EQ_INT((int)delete_plan_session_deleted(session), 1);
+
+  delete_plan_session_destroy(session);
+  close(p[0]);
+  close(p[1]);
+  for (int i = 0; i < 3; i++) {
+    char path[1024];
+    snprintf(path, sizeof(path), "%s/e%d.txt", root, i);
+    unlink(path);
+  }
+  rmdir(root);
+  config_delete(config);
+}
+
+void test_delete_plan(void) {
+  test_delete_delay_refilled_dir_not_counted();
+  test_delete_delay_removed_file_counted();
+  test_delete_delay_max_delete_bounds_actual();
+}

tests/test_delete_plan.h

@@ -0,0 +1,6 @@
+#ifndef TEST_DELETE_PLAN_H
+#define TEST_DELETE_PLAN_H
+
+void test_delete_plan(void);
+
+#endif