feat(stats): populate receiver wire counters on both receive paths

The single-threaded and -m receivers never populated ReceiverStats.matched_data
or .deleted_files, so --stats always printed 0 for both even when rsync
reported nonzero.  Track the bytes reconstructed from the basis file while
applying a delta, and tally the delete-commit counts (manifest and
per-directory sessions) into the receiver stats.  The -m pipeline now carries
its own stats/would-delete fields and emits the STATUS_STATS frame before the
terminal success, so --threads finally reports the counters and renders
-n --delete  lines.

Also normalize the -n --delete would-delete enumeration's absolute basis
prefixes exactly like the real commit path (fixing an over-report) and fix the
basis_delete_relative off-by-one when the receive root is '/'.  Unit tests
cover the root mapping and the basis protection; integration tests cover
matched/deleted stats for both receivers and the --threads dry-run delete
lines.

src/shared/delete_plan.c

@@ -444,6 +444,10 @@ bool delete_plan_session_limit_reached(const DeletePlanSession* session) {
   return session && session->limit_hit;
 }
 
+size_t delete_plan_session_deleted(const DeletePlanSession* session) {
+  return session ? session->deleted : 0;
+}
+
 /* True for a destination-relative path section entry (non-empty, relative,
  * traversal-free). */
 static bool valid_rel_path(const char* value) {

src/shared/delete_plan.h

@@ -70,5 +70,8 @@ int delete_plan_session_receive(DeletePlanSession* session, const Config* config
 DeleteCommitResult delete_plan_session_commit(DeletePlanSession* session, const Config* config);
 /* True once the shared --max-delete budget stopped part of a deletion. */
 bool delete_plan_session_limit_reached(const DeletePlanSession* session);
+/* Number of destination entries the session's plans removed (or, for
+   --delete-delay, snapshotted for removal), for the end-of-transfer stats. */
+size_t delete_plan_session_deleted(const DeletePlanSession* session);
 
 #endif

src/shared/file_receive.c

@@ -1093,6 +1093,13 @@ static File* receive_delta_file(int fd, const Config* config, const char* check_
       *failed = true;
       return NULL;
     }
+    /* Wire-stats tally: bytes taken straight from the basis file (matched
+       delta blocks).  Computed before the delta is destroyed. */
+    unsigned long long matched = 0;
+    for (uint32_t k = 0; k < delta->instruction_count; k++) {
+      if (delta->instructions[k].type == DELTA_INSTR_BLOCK_MATCH)
+        matched += delta->instructions[k].match.length;
+    }
     void* new_data = delta_apply(old_data, old_size, delta, config->delta_block_size);
     delta_destroy(delta);
 
@@ -1111,6 +1118,7 @@ static File* receive_delta_file(int fd, const Config* config, const char* check_
       *failed = true;
       return NULL;
     }
+    file->matched_bytes = matched;
 
     if (config->use_metadata) {
       int meta_ok = 1;
@@ -3149,8 +3157,9 @@ typedef struct {
    compares paths relative to the receive root, so a relative entry is already
    in the right form; an absolute entry that lies below the root is converted to
    its root-relative form, and one outside the root returns NULL (the walk
-   cannot reach it, and it is not protected data beneath the root). */
-static char* basis_delete_relative(const Config* config, const char* path) {
+   cannot reach it, and it is not protected data beneath the root).  Exposed so
+   tests can exercise the root-of-"/" child mapping directly. */
+char* file_receive_basis_delete_relative(const Config* config, const char* path) {
   if (!path)
     return NULL;
   if (path[0] != '/')
@@ -3163,6 +3172,13 @@ static char* basis_delete_relative(const Config* config, const char* path) {
     root_len--;
   if (strncmp(path, root, root_len) != 0)
     return NULL;
+  if (root_len == 1 && root[0] == '/') {
+    /* The receive root is "/": every absolute path is below it, and the child
+       relative form is everything after the leading '/'. */
+    if (path[1] == '\0')
+      return NULL; /* identical to the root, not a child */
+    return str_dup(path + 1);
+  }
   if (path[root_len] != '/')
     return NULL; /* identical or a sibling sharing a name prefix */
   return str_dup(path + root_len + 1);
@@ -3217,7 +3233,7 @@ static bool delete_extras_budgeted(const Config* config, DeleteManifest* manifes
     for (int i = 0; i < config->basis_count; i++) {
       /* An absolute basis outside the receive root is unreachable by this walk,
          so it contributes no protection prefix (and no slot). */
-      char* prefix = basis_delete_relative(config, config->basis_dirs[i].path);
+      char* prefix = file_receive_basis_delete_relative(config, config->basis_dirs[i].path);
       if (!prefix)
         continue;
       owned_prefixes[i] = prefix;
@@ -3304,7 +3320,7 @@ static bool delete_missing_args_budgeted(const Config* config, DeleteManifest* m
       idx++;
     }
     for (int i = 0; i < config->basis_count; i++) {
-      char* prefix = basis_delete_relative(config, config->basis_dirs[i].path);
+      char* prefix = file_receive_basis_delete_relative(config, config->basis_dirs[i].path);
       if (!prefix)
         continue;
       owned_prefixes[i] = prefix;
@@ -3469,10 +3485,16 @@ bool manifest_would_delete_list(const Config* config, DeleteManifest* manifest,
   int skip_count = (config->delay_updates ? 1 : 0) + config->basis_count +
                    (manifest->protected ? manifest->protected->size : 0);
   DeleteSkipEntry* skips = NULL;
+  char** owned_prefixes = NULL;
+  int used = 0;
   if (skip_count > 0) {
     skips = calloc((size_t)skip_count, sizeof(DeleteSkipEntry));
-    if (!skips)
+    owned_prefixes = calloc((size_t)config->basis_count, sizeof(char*));
+    if (!skips || (config->basis_count > 0 && !owned_prefixes)) {
+      free(skips);
+      free(owned_prefixes);
       return false;
+    }
     int idx = 0;
     if (config->delay_updates) {
       skips[idx].prefix = DELAY_UPDATES_STAGING_DIR;
@@ -3480,7 +3502,14 @@ bool manifest_would_delete_list(const Config* config, DeleteManifest* manifest,
       idx++;
     }
     for (int i = 0; i < config->basis_count; i++) {
-      skips[idx].prefix = config->basis_dirs[i].path;
+      /* Normalize exactly like the real commit path: a relative entry is
+         already root-relative, an absolute one inside the receive root is
+         converted, and one outside contributes no protection prefix. */
+      char* prefix = file_receive_basis_delete_relative(config, config->basis_dirs[i].path);
+      if (!prefix)
+        continue;
+      owned_prefixes[i] = prefix;
+      skips[idx].prefix = prefix;
       skips[idx].top_level_only = false;
       idx++;
     }
@@ -3489,9 +3518,15 @@ bool manifest_would_delete_list(const Config* config, DeleteManifest* manifest,
       skips[idx].top_level_only = false;
       idx++;
     }
+    used = idx;
   }
   bool ok = delete_extras_list(config->receive_root_directory, manifest->keeps, manifest->dirs,
-                               skips, skip_count, out, count_out);
+                               skips, used, out, count_out);
+  if (owned_prefixes) {
+    for (int i = 0; i < config->basis_count; i++)
+      free(owned_prefixes[i]);
+  }
+  free(owned_prefixes);
   free(skips);
   return ok;
 }
@@ -3531,6 +3566,13 @@ bool manifest_delete_missing_args_limited(const Config* config, DeleteManifest*
    Both draw from one --max-delete budget; the result reports a cap-stopped
    (partial) commit distinctly so the client can exit 25 like rsync. */
 DeleteCommitResult manifest_delete_all(const Config* config, DeleteManifest* manifest) {
+  return manifest_delete_all_counted(config, manifest, NULL);
+}
+
+DeleteCommitResult manifest_delete_all_counted(const Config* config, DeleteManifest* manifest,
+                                               size_t* deleted) {
+  if (deleted)
+    *deleted = 0;
   if (!config || !manifest)
     return DELETE_COMMIT_ERROR;
   /* Central no-mutation guard: a dry-run never deletes.  No manifest is sent on
@@ -3551,6 +3593,8 @@ DeleteCommitResult manifest_delete_all(const Config* config, DeleteManifest* man
     return DELETE_COMMIT_ERROR;
   if (config->use_delete && !delete_extras_budgeted(config, manifest, &budget))
     return DELETE_COMMIT_ERROR;
+  if (deleted)
+    *deleted = budget.deleted;
   if (budget.limit_hit) {
     if (user_limited) {
       log_message(LOG_LEVEL_ERROR, "Deletions stopped due to --max-delete limit (%zu skipped)",

src/shared/file_receive.h

@@ -142,6 +142,10 @@ typedef enum {
    do or everything committed, DELETE_COMMIT_LIMIT_REACHED when the budget
    stopped part of the work, or DELETE_COMMIT_ERROR on a genuine failure. */
 DeleteCommitResult manifest_delete_all(const Config* config, DeleteManifest* manifest);
+/* Like manifest_delete_all, but reports how many destination entries the commit
+   removed (for the end-of-transfer wire stats).  `deleted` may be NULL. */
+DeleteCommitResult manifest_delete_all_counted(const Config* config, DeleteManifest* manifest,
+                                               size_t* deleted);
 
 /* -n/--dry-run --delete would-delete reporting: walk the destination exactly as
    the delete pass would and append (strdup'd) destination-relative paths that
@@ -150,6 +154,10 @@ DeleteCommitResult manifest_delete_all(const Config* config, DeleteManifest* man
    clean walk; `*count_out` receives the number of paths appended. */
 bool manifest_would_delete_list(const Config* config, DeleteManifest* manifest, ArrayList* out,
                                 size_t* count_out);
+/* Convert one basis-directory path to the receive-root-relative protection
+   prefix the delete walker uses (NULL when it lies outside the root).  Exposed
+   for unit tests of the root-of-"/" and normalization edge cases. */
+char* file_receive_basis_delete_relative(const Config* config, const char* path);
 
 /* Outcome of a single file_save_to_disk operation.  The receiver needs to
    distinguish "written" from "skipped" so --remove-source-files can be told

src/shared/file_types.h

@@ -93,6 +93,10 @@ typedef struct {
    * no report was requested/received, in which case -i/--out-format treats the
    * entry conservatively as newly created. */
   OutputDestState dest_state;
+  /* Receiver-only wire-stats tally: the number of bytes reconstructed from the
+   * basis file (matched delta blocks) for this entry.  0 when the file was sent
+   * whole.  Accumulated into ReceiverStats.matched_data by the receiver sink. */
+  unsigned long long matched_bytes;
 } File;
 
 /* The path that should be sent on the wire and used for the receiver-side