Merge feat/p3-delete-policy: delete policy (--delete-excluded/--max-delete/--ignore-errors/--force/--prune-empty-dirs)

RSYNC_COMPAT.md

@@ -103,16 +103,16 @@ This document maps rsync's full feature set to FastSync's current implementation
 
 | Flag | Rsync Description | FastSync Status | Notes |
 |------|-------------------|-----------------|-------|
-| `--delete` | Delete extraneous files from dest | ✅ Implemented | `use_delete` config field. Deletion is always derived from the transmitted keep-set manifest of the paths the sender sent/keeps (never from unchecked input), runs through the symlink-safe walker bounded by `MAX_SERVER_DELETE_COUNT`, and skips the `.fastsync-stage` staging dir under `--delay-updates`. FastSync's default timing when no timing flag is given is **delete-after** (extras are removed only once the whole transfer succeeded) — intentionally NOT rsync's `--del`/delete-during default, to preserve FastSync's commit-style safety |
+| `--delete` | Delete extraneous files from dest | ✅ Implemented | `use_delete` config field. Deletion is always derived from the transmitted keep-set manifest of the paths the sender sent/keeps (never from unchecked input), runs through the symlink-safe walker bounded by `MAX_SERVER_DELETE_COUNT`, and skips the `.fastsync-stage` staging dir under `--delay-updates`. FastSync's default timing when no timing flag is given is **delete-after** (extras are removed only once the whole transfer succeeded) — intentionally NOT rsync's `--del`/delete-during default, to preserve FastSync's commit-style safety. By default the destination mirror of a path the source scan pruned (filter/exclude/size rules) is **protected** from deletion — matching rsync, which does not delete excluded files under `--delete`; `--delete-excluded` opts back into deleting them (see below). The bounded deletion is **all-or-nothing**: if the destination holds more extras than the effective bound (a client `--max-delete=NUM` or the 100000-entry server bound) nothing is deleted and the run fails with a distinct error instead of silently truncating |
-| `--delete-before` | Delete before transfer | ✅ Implemented | Implies `--delete`. The sender runs a full source pre-scan (paths only) and transmits the keep-set manifest BEFORE any file data; the receiver validates it, removes every destination entry not listed (bounded walk, staging-dir skip), then acks `STATUS_OK`. The sender only starts streaming after the deletion committed, or aborts if the receiver reported a deletion error. By definition the deletions already happened when a later transfer phase fails — rsync's delete-before is destructive the same way; a subsequent failure does not restore the removed files. Divergence: the keep-set is the pre-scan snapshot, so a file that appears on the source between the pre-scan and the data pass is still transferred but was not protected from deletion |
+| `--delete-before` | Delete before transfer | ✅ Implemented | Implies `--delete`. The sender runs a full source pre-scan (paths only) and transmits the keep-set manifest BEFORE any file data; the receiver validates it, removes every destination entry not listed (all-or-nothing bounded walk, staging-dir skip, protected prefixes honored), then acks `STATUS_OK`. The sender only starts streaming after the deletion committed, or aborts if the receiver reported a deletion error. By definition the deletions already happened when a later transfer phase fails — rsync's delete-before is destructive the same way; a subsequent failure does not restore the removed files. Divergence: the keep-set is the pre-scan snapshot, so a file that appears on the source between the pre-scan and the data pass is still transferred but was not protected from deletion |
 | `--del`, `--delete-during` | Delete during transfer | ✅ Implemented | Both spellings accepted; imply `--delete`. FastSync streams the source in a single directory scan and has no per-directory generator pass, so deletions cannot be interleaved per-directory the way rsync's delete-during does. `--delete-during` therefore selects the same early engine mode as `--delete-before` (manifest transmitted before any data, extras removed and acknowledged before data is applied); observable success/failure behaviour equals `--delete-before`. That is the documented divergence from rsync, where `--del` is the default meaning of `--delete` |
 | `--delete-delay` | Find deletions during, delete after | ✅ Implemented | Implies `--delete`. Commit-mode timing: extras are removed only after the whole transfer succeeded. rsync's delete-delay records the deletion list during its scan and applies it at the end; FastSync never snapshots the destination while data flows (the keep-set is the transmitted manifest and the destination is listed only at deletion time), so `--delete-delay` is implemented as the same end-of-transfer commit as `--delete-after` with identical safety. That is the documented divergence |
 | `--delete-after` | Delete after transfer | ✅ Implemented | Implies `--delete`. The delete-after timing is also what plain `--delete` does: the keep-set manifest closes the data stream and the receiver commits the bounded deletion only after the terminal `STATUS_FINISHED` proves the whole transfer (every data frame received and stored) succeeded. A failed or aborted transfer removes nothing |
-| `--delete-excluded` | Also delete excluded files | ❌ Not Implemented | Removed because it had no effect |
+| `--delete-excluded` | Also delete excluded files | ✅ Implemented | `delete_excluded` config field. Under `--delete` FastSync now protects (rsync's default) the destination mirror of paths the sender's source scan pruned by user-selection rules — the `--filter`/`-F`/`-C` layer, the legacy `--exclude`/`--include` layer, and `--max-size`/`--min-size`. The sender transmits those concrete pruned paths as **protected prefixes** in the delete-manifest frame (see the Phase-3 notes below); the walker never descends into or removes them. `--delete-excluded` opts back in: the sender sends an empty protected list, so the excluded destination mirrors become ordinary extras and are removed. Divergences (documented): protection is derived only from what the source scan actually pruned — a stray destination-only file that happens to match an exclude rule is not protected (FastSync never re-applies rules to the destination, keeping deletion sender-derived), and `--files-from` subset pruning stays keep-set-only (an unlisted source path is treated as absent and its mirror is deletable, matching the `--files-from` delete note below). The two are orthogonal: `--delete-excluded` removes filter-excluded mirrors; it does not make `--files-from` prune things |
-| `--max-delete=NUM` | Max files to delete | ❌ Not Implemented | Removed because it had no effect |
+| `--max-delete=NUM` | Max files to delete | ✅ Implemented | `max_delete` config field (default -1 = no client limit; 0 = delete nothing). NUM bounds a `--delete` run with rsync's all-or-nothing semantics: the receiver rehearses the deletion first and, if the destination holds more than NUM extras, deletes NOTHING and fails the transfer with a distinct `--max-delete` error. A run at or below NUM deletes exactly the extras. NUM only applies together with `--delete` (it is inert otherwise, matching rsync). The hard server bound `MAX_SERVER_DELETE_COUNT` (100000) still caps the walk; a NUM above it never raises that cap, and exceeding the server bound is its own all-or-nothing error. Directories count toward the limit (each removed empty directory is one deletion), like rsync |
-| `--ignore-errors` | Delete even with I/O errors | ❌ Not Implemented | |
+| `--ignore-errors` | Delete even with I/O errors | ✅ Implemented | Sender-side, client-only config field. rsync suppresses `--delete` when the transfer had I/O errors; FastSync's equivalent is a source-scan I/O error (an unreadable directory, e.g. EACCES): by default the scan aborts the run so no deletion happens. With `--ignore-errors` the scan continues past the unreadable directory, the readable tree is transferred and the deletion still runs (the mirror of the unreadable directory is treated as an extra). The run still exits non-zero (the error is reported, matching rsync's error status). Divergence: without the flag FastSync aborts the whole run on the scan error, whereas rsync transfers the rest of the tree and merely skips the deletion; both leave the deletion undone |
-| `--force` | Force deletion of non-empty dirs | ❌ Not Implemented | |
+| `--force` | Force deletion of non-empty dirs | ✅ Implemented | `force_delete` receiver config field (crosses the wire). rsync's `--force` lets an incoming non-directory replace a destination directory; FastSync implements exactly that: when a regular file is written to a path that is currently a (possibly non-empty) destination directory, `--force` removes that directory tree first — confined to the receive root and symlink-safe (O_NOFOLLOW fd walk, symlinks removed by name, never followed) — so the atomic install can place the file. Without `--force` such a write fails and the run aborts. Divergence: `--force` acts on the immediate-install path only; under `--delay-updates` a blocking directory is not cleared (publication renames over regular files) |
-| `--prune-empty-dirs` | Prune empty dir chains | ❌ Not Implemented | Removed because it had no effect |
+| `--prune-empty-dirs` | Prune empty dir chains | ✅ Implemented | Long-only: FastSync's `-m` is already multithreading (recorded divergence — rsync's `-m` short form is not reassigned). FastSync's recursive transfer never emits directory entries, so empty directories are inherently never transferred (which is rsync's `-m` behavior) and truly-empty destination directory chains are removed by `--delete` regardless of this flag. The flag's additional real effect is on the `--dirs` explicit directory-entry generator: a plain `-d <empty-dir>` run omits the empty source directory's entry, so nothing is created at the destination (no `STATUS_MKDIR`, no `-i`/`--out-format` change line, and an existing empty mirror becomes an extra that `--delete` prunes). Explicitly `--files-from`-listed directories always pass through (documented `--files-from` behavior). A directory that still holds an excluded-but-protected file survives, matching the `--delete-excluded` default |
 
 **Deletion-timing implementation notes (Phase 3):** the delete flags above are
 real. Two new config booleans (`delete_during`, `delete_delay`) join the already
@@ -129,18 +129,52 @@ manifest ack. `--delete-delay` and `--delete-during` are each implemented as
 the closest safe approximation their engine mode allows; the divergences are
 noted in the rows above.
 
-Manifest size: the sender's keep-set collection (streaming or early pre-scan)
+**Deletion-policy notes (Phase 3, delete-policy wave):** this wave made the
-is unbounded, but the receiver rejects any manifest beyond `MAX_MANIFEST_ENTRIES`
+deletion family real — `--delete-excluded`, `--max-delete`, `--ignore-errors`,
-(1 048 576 entries) / `MAX_MANIFEST_BYTES` (16 MB of paths) as a hard protocol
+`--force`, `--prune-empty-dirs` — and, to support them, the `STATUS_MANIFEST`
-error. In the commit modes this only means the deletion is refused after the
+frame now carries **two sections**: the keep-set paths followed by a list of
-data already arrived; in the NEW early modes (`--delete-before`/`--delete-during`)
+**protected prefixes** (destination-relative paths the source scan pruned by
-the manifest is the first frame, so an oversized keep-set now aborts the whole
+user-selection rules, which the walker must never delete unless
-transfer BEFORE any data is sent (previously all data transferred and only the
+`--delete-excluded` opted out). Two config booleans were added for the wave:
-deletion step failed). Keep the source tree small enough for the receiver's
+`force_delete` (crosses the wire; the receiver clears a directory that blocks an
+incoming file) and `ignore_errors` (client-only; the sender's scan continues
+past an unreadable directory). `max_delete`'s default became -1 ("no client
+limit"). These wire/layout changes bumped `PROTOCOL_VERSION` **2.8.0 → 2.9.0**
+(peers must match). All four wire additions — `force_delete`,
+`delete_excluded`, `prune_empty_dirs`, `max_delete` — round-trip unchanged and
+are validated on receive.
+
+The deletion walker is now **all-or-nothing**: before any unlink it rehearses
+the deletion (an fd-relative walk identical to the delete pass, counting every
+regular file it would unlink and every directory it would remove) and refuses to
+start when the extras exceed the effective bound — a client `--max-delete=NUM`
+below the hard bound, or the hard `MAX_SERVER_DELETE_COUNT` (100000) bound
+itself. Previously the walker removed up to `MAX_SERVER_DELETE_COUNT` extras and
+then reported an error (a truncated deletion); it now removes nothing and fails
+with an error naming the bound. Directories count toward the bound. A directory
+that still holds entries the walker leaves in place (a protected excluded file,
+a kept manifest entry, a symlink) is left behind rather than failing the run —
+matching rsync's "cannot delete non-empty directory" behaviour. The
+all-or-nothing guarantee holds only while the destination is not concurrently
+modified: rehearsal and delete are two separate walks, so a concurrent change
+between them (another process adding or removing destination entries) can make
+the actual deletion diverge from the counted set.
+
+Manifest size: the sender's keep-set and protected-prefix collections (streaming
+or early pre-scan) are unbounded, but the receiver rejects a manifest beyond
+`MAX_MANIFEST_ENTRIES` (1 048 576 entries, applied to EACH section — a frame can
+therefore total up to 2 097 152 entries) / `MAX_MANIFEST_BYTES` (16 MB of paths,
+counted across BOTH sections) as a hard protocol error. A heavily filtered
+source whose exclusion list grows large thus fails the run cleanly on the
+receiver (STATUS_ERROR) instead of being silently truncated. In the commit
+modes this only means the deletion is refused after the data already arrived; in
+the early modes (`--delete-before`/`--delete-during`) the manifest is the first
+frame, so an oversized keep-set or protected list aborts the whole transfer
+BEFORE any data is sent. Keep the source tree small enough for the receiver's
 manifest caps when using the early timing.
 
 Early-delete ACK wait: after committing a large deletion (up to
-`MAX_SERVER_DELETE_COUNT` unlinks) the receiver's `STATUS_OK`/`STATUS_ERROR`
+`MAX_SERVER_DELETE_COUNT` removals) the receiver's `STATUS_OK`/`STATUS_ERROR`
 reply can legitimately take much longer than a normal round trip, so the sender
 waits for that single ACK with an extended explicit deadline (1 hour) instead
 of the default 60 s per-message receive window. A receiver that is genuinely
@@ -151,7 +185,9 @@ Flag-conflict policy: unlike rsync's last-one-wins behaviour, every deletion
 timing flag implies `--delete`, and combining a timing flag with `--no-delete`
 (in either argument order) — or more than one timing flag — is rejected as a
 configuration error rather than silently resolved. Note the check is
-order-independent because it runs over the fully parsed config.
+order-independent because it runs over the fully parsed config. The deletion
+POLICY flags (`--delete-excluded`, `--max-delete`, `--ignore-errors`, `--force`)
+do NOT imply `--delete`; without `--delete` they are inert (matching rsync).
 
 ## 8. Metadata Preservation
 
@@ -210,9 +246,9 @@ order-independent because it runs over the fully parsed config.
 |------|-------------------|-----------------|-------|
 | `--checksum` | Skip based on checksum | ✅ Implemented | With `--incremental`, compares xxHash64 content checksums; `-c` remains compression |
 | `--checksum-choice=STR` | Choose checksum algorithm | ❌ Not Implemented | xxHash used internally |
-| `--compare-dest=DIR` | Compare dest files relative to DIR | ✅ Implemented | DIR is a receiver-side basis relative to the destination root (confined below it; absolute/`..`/`.` rejected, `//` collapsed and trailing `/` dropped). On the receiver's per-file check (implies `--incremental`) an exact match = same size + mtime (unless `--size-only`; `-I` disables matching) **and** equal xxHash64 of the sender's file; a match suppresses the data transfer. compare-dest never copies: it only skips a file the destination does **not** already hold (sparse destination, rsync parity), and is consulted before the normal delta/full paths. Repeatable; searched in command-line order, first match wins. Divergences: when the destination already holds a *different* version rsync deletes it but FastSync instead transfers the data (keeps the mirror complete; never deletes without `--delete`); attribute-only differences on a match are not re-applied (data is skipped so the sender never sends metadata); content is verified by xxHash64, stricter than rsync's default quick check. Sizing: FastSync's whole-file payload limit is 256 MiB on **every** transfer path (not basis-specific); rsync applies basis dirs to arbitrary sizes, so FastSync refuses a basis run whose source contains a larger file up front with a clear error before any transfer. Wire: a basis-count field is always present on the config frame (protocol bumped to 2.8.0, so clients and servers must both be 2.8.0) |
+| `--compare-dest=DIR` | Compare dest files relative to DIR | ✅ Implemented | DIR is a receiver-side basis relative to the destination root (confined below it; absolute/`..`/`.` rejected, `//` collapsed and trailing `/` dropped). On the receiver's per-file check (implies `--incremental`) an exact match = same size + mtime (unless `--size-only`; `-I` disables matching) **and** equal xxHash64 of the sender's file; a match suppresses the data transfer. compare-dest never copies: it only skips a file the destination does **not** already hold (sparse destination, rsync parity), and is consulted before the normal delta/full paths. Repeatable; searched in command-line order, first match wins. Divergences: when the destination already holds a *different* version rsync deletes it but FastSync instead transfers the data (keeps the mirror complete; never deletes without `--delete`); attribute-only differences on a match are not re-applied (data is skipped so the sender never sends metadata); content is verified by xxHash64, stricter than rsync's default quick check. Sizing: FastSync's whole-file payload limit is 256 MiB on **every** transfer path (not basis-specific); rsync applies basis dirs to arbitrary sizes, so FastSync refuses a basis run whose source contains a larger file up front with a clear error before any transfer. Wire: a basis-count field is always present on the config frame (protocol bumped to 2.8.0, so clients and servers must both be 2.8.0 (current wire version is 2.9.0)) |
-| `--copy-dest=DIR` | Include copies of unchanged files | ✅ Implemented | Same basis rules as `--compare-dest`, but an exact match materializes a **local copy** of the DIR file into the destination (via the normal atomic temp+rename store path, so `--existing`/`--ignore-existing`/`--update`/`--backup`/`--delay-updates` all still apply) instead of transferring data. Repeatable; command-line order = priority. Content is xxHash64-verified before the copy. Divergences: a basis-hit destination keeps the basis file's own mode/uid/gid and mtime (the sender sends no metadata on a skip), so with `--size-only` its mtime can differ from the source and attribute-only differences are copied with the basis attributes rather than rsync's "copy + fix attributes". Requires `--incremental` (implied); incompatible with `-s`. Wire: protocol 2.8.0 |
+| `--copy-dest=DIR` | Include copies of unchanged files | ✅ Implemented | Same basis rules as `--compare-dest`, but an exact match materializes a **local copy** of the DIR file into the destination (via the normal atomic temp+rename store path, so `--existing`/`--ignore-existing`/`--update`/`--backup`/`--delay-updates` all still apply) instead of transferring data. Repeatable; command-line order = priority. Content is xxHash64-verified before the copy. Divergences: a basis-hit destination keeps the basis file's own mode/uid/gid and mtime (the sender sends no metadata on a skip), so with `--size-only` its mtime can differ from the source and attribute-only differences are copied with the basis attributes rather than rsync's "copy + fix attributes". Requires `--incremental` (implied); incompatible with `-s`. Wire: protocol 2.9.0 |
-| `--link-dest=DIR` | Hardlink to files when unchanged | ✅ Implemented | Same basis rules as `--copy-dest`, but an exact match installs an atomic **hard link** to the DIR file (temp hard link + rename) so no data or disk space is used; where the link is impossible (basis on another filesystem, filesystem refuses links) it falls back cleanly to a byte-identical local copy, never a corrupt/partial file. `--delay-updates` stages the link and publishes by rename, so the final entry stays a real hard link. Repeatable (searched in command-line order, first match wins). Content is xxHash64-verified before linking. Divergences and caveats: an already up-to-date destination file is not re-linked to a basis file (only files that would otherwise be written are linked); a link keeps the basis inode's own mode/uid/gid and mtime — metadata is never written through the shared inode (that would mutate the basis file), so a later `--inplace` run that rewrites such a destination path **will mutate the basis snapshot** through the shared inode (use `--copy-dest` when the destination must stay independently writable); with `--size-only` the linked mtime can differ from the source; a `--remove-source-files` source satisfied by a basis dir is treated as skipped and therefore **retained** (never removed); basis dirs are excluded from `--delete`. Requires `--incremental` (implied); incompatible with `-s`. Wire: protocol 2.8.0 |
+| `--link-dest=DIR` | Hardlink to files when unchanged | ✅ Implemented | Same basis rules as `--copy-dest`, but an exact match installs an atomic **hard link** to the DIR file (temp hard link + rename) so no data or disk space is used; where the link is impossible (basis on another filesystem, filesystem refuses links) it falls back cleanly to a byte-identical local copy, never a corrupt/partial file. `--delay-updates` stages the link and publishes by rename, so the final entry stays a real hard link. Repeatable (searched in command-line order, first match wins). Content is xxHash64-verified before linking. Divergences and caveats: an already up-to-date destination file is not re-linked to a basis file (only files that would otherwise be written are linked); a link keeps the basis inode's own mode/uid/gid and mtime — metadata is never written through the shared inode (that would mutate the basis file), so a later `--inplace` run that rewrites such a destination path **will mutate the basis snapshot** through the shared inode (use `--copy-dest` when the destination must stay independently writable); with `--size-only` the linked mtime can differ from the source; a `--remove-source-files` source satisfied by a basis dir is treated as skipped and therefore **retained** (never removed); basis dirs are excluded from `--delete`. Requires `--incremental` (implied); incompatible with `-s`. Wire: protocol 2.9.0 |
 | `--fuzzy`, `--no-fuzzy` | Find similar file for basis | ❌ Not Implemented | |
 
 ## 12. Compression

src/client/client_cli.c

@@ -454,6 +454,11 @@ static const OptionEntry OPTION_TABLE[] = {
     {"--delete-during", "--del", OPT_FLAG, offsetof(Config, delete_during)},
     {"--delete-delay", NULL, OPT_FLAG, offsetof(Config, delete_delay)},
     {"--delete-after", NULL, OPT_FLAG, offsetof(Config, delete_after)},
+    {"--delete-excluded", NULL, OPT_FLAG, offsetof(Config, delete_excluded)},
+    {"--max-delete", NULL, OPT_NONNEG_INT, offsetof(Config, max_delete)},
+    {"--ignore-errors", NULL, OPT_FLAG, offsetof(Config, ignore_errors)},
+    {"--force", NULL, OPT_FLAG, offsetof(Config, force_delete)},
+    {"--prune-empty-dirs", NULL, OPT_FLAG, offsetof(Config, prune_empty_dirs)},
 
     {"--source-dir", NULL, OPT_STRING, offsetof(Config, send_directory)},
     {"--dest-dir", NULL, OPT_STRING, offsetof(Config, receive_root_directory)},

src/client/client_send.c

@@ -102,6 +102,10 @@ static bool prepare_scanner(const Config* config, int num_threads, PreparedScann
   options->per_dir_filters = config->per_dir_filter;
   options->dirs = config->dirs;
   options->relative = config->relative;
+  options->prune_empty_dirs = config->prune_empty_dirs;
+  options->ignore_io_errors = config->ignore_errors;
+  options->excluded_paths = NULL;
+  options->excluded_mutex = NULL;
   return true;
 }
 
@@ -255,7 +259,7 @@ static bool basis_oversize_preflight(const Config* config) {
     if (!ok)
       break;
   }
-  if (directory_scanner_failed(scanner))
+  if (directory_scanner_failed(scanner) || directory_scanner_had_io_error(scanner))
     ok = false;
   directory_scanner_destroy(scanner);
   return ok;
@@ -596,7 +600,7 @@ static int send_list_only(const Config* config) {
     if (oom)
       break;
   }
-  bool failed = oom || directory_scanner_failed(scanner);
+  bool failed = oom || directory_scanner_failed(scanner) || directory_scanner_had_io_error(scanner);
   directory_scanner_destroy(scanner);
   prepared_scanner_destroy(&prepared);
   if (failed) {
@@ -621,8 +625,16 @@ static int send_list_only(const Config* config) {
   return 0;
 }
 
-/* Send the delete manifest (list of files) to the server. Returns 0 on success, -1 on failure. */
+/* Send the delete manifest (keep-set paths plus the protected excluded
-static int send_delete_manifest(int fd, ArrayList* manifest) {
+   prefixes) to the server. Returns 0 on success, -1 on failure.  When
+   --delete-excluded is given `protected` is empty: excluded destination
+   mirrors are then ordinary extras and are removed.  Both sections are
+   unbounded on the sender; the receiver enforces MAX_MANIFEST_ENTRIES per
+   section and a single MAX_MANIFEST_BYTES budget shared across the two
+   sections, rejecting (with STATUS_ERROR) an over-budget frame.  A heavily
+   filtered source whose exclusion list is large therefore fails the run
+   cleanly on the receiver rather than being truncated. */
+static int send_delete_manifest(int fd, ArrayList* manifest, ArrayList* protected_prefixes) {
   if (!manifest)
     return -1;
   if (!send_status(fd, STATUS_MANIFEST))
@@ -633,6 +645,13 @@ static int send_delete_manifest(int fd, ArrayList* manifest) {
     if (!send_str(fd, (char*)manifest->items[i]))
       return -1;
   }
+  int protected_count = protected_prefixes ? protected_prefixes->size : 0;
+  if (!send_int(fd, protected_count))
+    return -1;
+  for (int i = 0; i < protected_count; i++) {
+    if (!send_str(fd, (char*)protected_prefixes->items[i]))
+      return -1;
+  }
   return 0;
 }
 
@@ -647,10 +666,11 @@ static int send_delete_manifest(int fd, ArrayList* manifest) {
    instead of the default 60 s receive window. */
 #define DELETE_ACK_TIMEOUT_SEC 3600
 
-static bool send_delete_manifest_early(Client* client, ArrayList* manifest) {
+static bool send_delete_manifest_early(Client* client, ArrayList* manifest,
+                                       ArrayList* protected_prefixes) {
   if (!client || !manifest)
     return false;
-  if (send_delete_manifest(client->file_descriptor, manifest) != 0)
+  if (send_delete_manifest(client->file_descriptor, manifest, protected_prefixes) != 0)
     return false;
   Status ack;
   if (!receive_status_timed(client->file_descriptor, &ack, DELETE_ACK_TIMEOUT_SEC))
@@ -666,9 +686,14 @@ static bool send_delete_manifest_early(Client* client, ArrayList* manifest) {
    paths, loading and sending nothing.  --delete-before/--delete-during need the
    complete keep-set manifest before the first data byte, so it is built by a
    dedicated pre-scan pass and transmitted early; the data pass then re-scans
-   with a fresh scanner. */
+   with a fresh scanner.  A source I/O error is fatal unless the options carry
+   --ignore-errors, in which case the scan continues past the unreadable
+   directory and *io_error_out reports it (the caller still performs the
+   deletion but reports the run as errored). */
 static bool scan_paths_only(const Config* config, const ScannerOptions* options,
-                            ArrayList* manifest) {
+                            ArrayList* manifest, bool* io_error_out) {
+  if (io_error_out)
+    *io_error_out = false;
   DirectoryScanner* scanner =
       directory_scanner_create_with_options(config->send_directory, options);
   if (!scanner)
@@ -685,6 +710,8 @@ static bool scan_paths_only(const Config* config, const ScannerOptions* options,
   }
   if (ok && directory_scanner_failed(scanner))
     ok = false;
+  if (io_error_out)
+    *io_error_out = directory_scanner_had_io_error(scanner);
   directory_scanner_destroy(scanner);
   return ok;
 }
@@ -1004,7 +1031,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
   if (context->early_delete) {
     /* The keep-set manifest was prebuilt by a path-only pre-scan.  Transmit it
        and wait for the receiver to delete extras before streaming any data. */
-    if (!send_delete_manifest_early(client, context->manifest)) {
+    if (!send_delete_manifest_early(client, context->manifest, context->excluded_paths)) {
       pipeline_cancel(context);
       disconnect_transfer_client(client);
       mark_sender_done(context);
@@ -1026,10 +1053,27 @@ static int send_chunks_multithreaded(void* pipeline_context) {
         return thrd_error;
       }
       if (context->config->use_delete && !context->early_delete) {
-        if (send_delete_manifest(client->file_descriptor, context->manifest) != 0)
+        /* Empty keep-set + scan I/O error must not delete the whole destination
+           (the source may not be genuinely empty -- see send_files). */
+        bool empty_io;
+        mtx_lock(&context->mutex_scanner);
+        empty_io = context->scan_had_io_error && context->manifest && context->manifest->size == 0;
+        mtx_unlock(&context->mutex_scanner);
+        if (empty_io) {
+          log_message(LOG_LEVEL_ERROR,
+                      "source scan hit an I/O error before finding any file; refusing to delete "
+                      "with an empty keep-set (--delete)");
+          goto send_fail;
+        }
+        if (send_delete_manifest(client->file_descriptor, context->manifest,
+                                 context->excluded_paths) != 0)
           goto send_fail;
       }
       bool ok = finalize_transfer(client, context->config, context->remove_source_files);
+      if (!ok && context->config->use_delete)
+        log_message(LOG_LEVEL_ERROR,
+                    "server reported a deletion failure (--delete); see the server log for the "
+                    "reason (a --max-delete limit that the run would exceed deletes nothing)");
       if (ok)
         remove_transferred_sources(context->config, context->remove_source_files);
       mtx_lock(&context->mutex_progress);
@@ -1091,6 +1135,12 @@ static int scan_directory_multithreaded(void* pipeline_context) {
     protocol_session_unbind();
     return thrd_error;
   }
+  /* The keep-set manifest for the late modes is built from this data pass, so
+     the parallel scanner records the protected excluded prefixes here.  The
+     early modes already transmitted the pre-scan keep-set and its protected
+     list, so the data pass must not append to it again. */
+  if (!context->early_delete)
+    prepared.options.excluded_paths = context->excluded_paths;
   bool dirs_mode = prepared.options.dirs;
   DirectoryScanner* dscanner = NULL;
   ParallelScanner* scanner = NULL;
@@ -1138,6 +1188,11 @@ static int scan_directory_multithreaded(void* pipeline_context) {
       break;
     }
   }
+  /* Capture the scanner results BEFORE destroying the scanner objects (the
+     io_error flag lives on the scanner, so reading it after destroy would be a
+     use-after-free). */
+  bool had_io =
+      dirs_mode ? directory_scanner_had_io_error(dscanner) : parallel_scanner_had_io_error(scanner);
   if (dirs_mode)
     directory_scanner_destroy(dscanner);
   else
@@ -1153,6 +1208,13 @@ static int scan_directory_multithreaded(void* pipeline_context) {
     protocol_session_unbind();
     return thrd_error;
   }
+  /* --ignore-errors: an unreadable subdirectory was skipped (workers recorded
+     io_error, not failure); the deletion still runs but the run reports it. */
+  if (had_io) {
+    mtx_lock(&context->mutex_scanner);
+    context->scan_had_io_error = true;
+    mtx_unlock(&context->mutex_scanner);
+  }
   mtx_lock(&context->mutex_scanner);
   context->scanner_done = true;
   cnd_signal(&context->condition_not_empty_scanner);
@@ -1280,8 +1342,11 @@ int send_files(Config* config) {
   DirectoryScanner* scanner = NULL;
   ArrayList* manifest = NULL;
   ArrayList* remove_sources = NULL;
+  /* Protected excluded prefixes (delete-excluded default protection). */
+  ArrayList* excluded = NULL;
   bool delete_early = config->use_delete && config_delete_timing_early(config);
   bool send_failed = false;
+  bool had_scan_io = false;
   PreparedScanner prepared;
   memset(&prepared, 0, sizeof(prepared));
   if (!config_send(client->file_descriptor, config))
@@ -1292,6 +1357,16 @@ int send_files(Config* config) {
     remove_sources = array_list_create(source_file_destroy);
   if (config->remove_source_files && !remove_sources)
     goto send_fail;
+  /* Unless --delete-excluded opts out, collect the paths the source scan prunes
+     by user-selection rules so the receiver protects their destination mirrors
+     from --delete (rsync's default).  Only scans that build the keep-set get the
+     sink attached (prescan for early timing, the streaming data pass otherwise). */
+  if (config->use_delete && !config->delete_excluded) {
+    excluded = array_list_create(free);
+    if (!excluded)
+      goto send_fail;
+    prepared.options.excluded_paths = excluded;
+  }
   /* The late-timing modes (plain --delete / --delete-after / --delete-delay)
      build the manifest while streaming and send it after the last data frame.
      The early modes (--delete-before/--delete-during) send it up front from a
@@ -1303,13 +1378,28 @@ int send_files(Config* config) {
     ArrayList* early_manifest = array_list_create(free);
     if (!early_manifest)
       goto send_fail;
-    if (!scan_paths_only(config, &prepared.options, early_manifest)) {
+    bool prescan_ok = scan_paths_only(config, &prepared.options, early_manifest, &had_scan_io);
-      array_list_delete(early_manifest);
+    bool early_ok = false;
-      goto send_fail;
+    if (prescan_ok) {
+      /* A scan that hit an I/O error and produced NO keep entries is ambiguous
+         (the source may not be genuinely empty -- part of it was unreadable),
+         and an empty keep-set would delete the whole destination.  Refuse to
+         delete; the genuine-empty-source case has no io_error and still sends
+         its (empty) keep-set. */
+      if (had_scan_io && early_manifest->size == 0) {
+        log_message(LOG_LEVEL_ERROR,
+                    "source scan hit an I/O error before finding any file; refusing to delete "
+                    "with an empty keep-set (--delete)");
+        prescan_ok = false;
+      } else {
+        early_ok = send_delete_manifest_early(client, early_manifest, excluded);
+      }
     }
-    bool early_ok = send_delete_manifest_early(client, early_manifest);
     array_list_delete(early_manifest);
-    if (!early_ok)
+    /* The keep-set (and its protected prefixes) are already on the wire; the
+       data pass must not append to the exclusion list again. */
+    prepared.options.excluded_paths = NULL;
+    if (!prescan_ok || !early_ok)
       goto send_fail;
   } else if (config->use_delete) {
     manifest = array_list_create(free);
@@ -1377,10 +1467,21 @@ int send_files(Config* config) {
   }
   if (directory_scanner_failed(scanner))
     goto send_fail;
+  if (directory_scanner_had_io_error(scanner))
+    had_scan_io = true;
+  if (had_scan_io && manifest && manifest->size == 0) {
+    /* A scan that hit an I/O error and produced no keep entries is ambiguous;
+       an empty keep-set would delete the whole destination.  Refuse to delete
+       (see the early-timing comment above). */
+    log_message(LOG_LEVEL_ERROR,
+                "source scan hit an I/O error before finding any file; refusing to delete with "
+                "an empty keep-set (--delete)");
+    goto send_fail;
+  }
   if (manifest) {
     /* Late (commit) ordering: all file data is out; transmit the keep-set
        manifest so the receiver deletes only after the transfer succeeds. */
-    if (send_delete_manifest(client->file_descriptor, manifest) != 0) {
+    if (send_delete_manifest(client->file_descriptor, manifest, excluded) != 0) {
       array_list_delete(manifest);
       manifest = NULL;
       goto send_fail;
@@ -1389,6 +1490,10 @@ int send_files(Config* config) {
     manifest = NULL;
   }
   bool ok = finalize_transfer(client, config, remove_sources);
+  if (!ok && config->use_delete)
+    log_message(LOG_LEVEL_ERROR,
+                "server reported a deletion failure (--delete); see the server log for the "
+                "reason (a --max-delete limit that the run would exceed deletes nothing)");
   if (ok)
     remove_transferred_sources(config, remove_sources);
   if (config->show_progress && !config->quiet)
@@ -1410,13 +1515,17 @@ int send_files(Config* config) {
   }
   log_info_message(LOG_INFO_STATS, "Transfer summary: %d files, %.1f MB", total_files,
                    total_bytes / 1048576.0);
-  ret = ok ? 0 : 1;
+  /* --ignore-errors: an unreadable source directory was skipped but the run
+     still completed (and deleted); report the run as errored like rsync does. */
+  ret = (ok && !had_scan_io) ? 0 : 1;
 
 send_fail:
   /* Single cleanup path for all exits. The manifest is intentionally deleted
      here even on success without --delete, fixing a pre-existing leak. */
   if (manifest)
     array_list_delete(manifest);
+  if (excluded)
+    array_list_delete(excluded);
   if (remove_sources)
     array_list_delete(remove_sources);
   if (scanner)
@@ -1468,21 +1577,41 @@ int send_files_multithreaded(Config** config_ptr) {
     return 1;
   }
   *config_ptr = NULL; /* context now owns config through all remaining paths */
+  bool collect_excluded = config->use_delete && !config->delete_excluded;
   if (config->use_delete) {
     context->manifest = array_list_create(free);
     if (!context->manifest) {
       pipeline_context_sender_destroy(context);
       return 1;
     }
+    if (collect_excluded) {
+      context->excluded_paths = array_list_create(free);
+      if (!context->excluded_paths) {
+        pipeline_context_sender_destroy(context);
+        return 1;
+      }
+    }
     if (config_delete_timing_early(config)) {
       /* --delete-before/--delete-during: build the complete keep-set manifest
          (paths only, nothing loaded or sent) up front so the sender thread can
-         transmit it before the first data byte. */
+         transmit it before the first data byte.  The path-only pre-scan also
+         fills the protected excluded prefixes. */
       PreparedScanner prepared;
       memset(&prepared, 0, sizeof(prepared));
-      bool prebuilt = prepare_scanner(config, 4, &prepared) &&
+      bool prepared_ok = prepare_scanner(config, 4, &prepared);
-                      scan_paths_only(config, &prepared.options, context->manifest);
+      if (prepared_ok && context->excluded_paths)
+        prepared.options.excluded_paths = context->excluded_paths;
+      bool prebuilt = prepared_ok && scan_paths_only(config, &prepared.options, context->manifest,
+                                                     &context->scan_had_io_error);
       prepared_scanner_destroy(&prepared);
+      if (prebuilt && context->scan_had_io_error && context->manifest->size == 0) {
+        /* Empty keep-set + scan I/O error: refusing an empty keep-set manifest
+           would have deleted the whole destination (see send_files). */
+        log_message(LOG_LEVEL_ERROR,
+                    "source scan hit an I/O error before finding any file; refusing to delete "
+                    "with an empty keep-set (--delete)");
+        prebuilt = false;
+      }
       if (!prebuilt) {
         pipeline_context_sender_destroy(context);
         return 1;
@@ -1548,6 +1677,13 @@ int send_files_multithreaded(Config** config_ptr) {
     thrd_join(progress, NULL);
   }
 
+  bool scan_io;
+  mtx_lock(&context->mutex_scanner);
+  scan_io = context->scan_had_io_error;
+  mtx_unlock(&context->mutex_scanner);
+  bool sender_ok = sender_result == thrd_success;
+  /* --ignore-errors: the run completed (and deleted) past an unreadable source
+     directory; report it as errored like rsync does. */
   pipeline_context_sender_destroy(context);
-  return sender_result == thrd_success ? 0 : 1;
+  return sender_ok && !scan_io ? 0 : 1;
 }

src/client/scanner.c

@@ -112,6 +112,10 @@ typedef struct {
   char* path;
   struct stat stats;
   bool is_directory;
+  /* True when the entry was pruned by a user selection rule (--filter/-C/per-dir
+     rules, the --exclude/--include layer, or --max-size/--min-size) rather than
+     skipped for another reason (unreadable, symlink policy, not applicable). */
+  bool excluded;
 } ScannerEntry;
 
 /* --one-file-system (-x) decision. Only directories can carry a different
@@ -161,6 +165,38 @@ static bool entry_passes_selection(const FileListSet* file_list, const FilterRul
   return true;
 }
 
+/* Append `rel` to the caller's exclusion sink, taking `mtx` when shared across
+   parallel worker threads.  Returns false on allocation failure (list left
+   unchanged). */
+static bool excluded_sink_append(ArrayList* list, mtx_t* mtx, const char* rel) {
+  if (!list)
+    return true;
+  char* dup = str_dup(rel);
+  if (!dup)
+    return false;
+  if (mtx)
+    mtx_lock(mtx);
+  bool ok = array_list_add(list, dup);
+  if (mtx)
+    mtx_unlock(mtx);
+  if (!ok)
+    free(dup);
+  return ok;
+}
+
+/* Record one pruned-by-user-selection filesystem path in the scanner's
+   exclusion sink (see ScannerOptions.excluded_paths).  The stored form is the
+   entry's wire/destination-relative path (a single leading '/' removed, exactly
+   how manifest keep entries are stored), so the receiver's walker prefixes
+   match the destination layout.  An allocation failure is a fatal scan error. */
+static void scanner_record_excluded(DirectoryScanner* scanner, const char* fs_path) {
+  if (!scanner->excluded_paths || !fs_path)
+    return;
+  const char* rel = *fs_path == '/' ? fs_path + 1 : fs_path;
+  if (!excluded_sink_append(scanner->excluded_paths, scanner->excluded_mutex, rel))
+    scanner->failed = true;
+}
+
 /* Merge the open directory's own .rsync-filter rules into the inherited
  * context, returning the context used for this directory's entries. On a parse
  * error the scanner is marked failed. Returns 0 on success, -1 on failure. */
@@ -198,6 +234,7 @@ static int open_directory_filter_context(DirectoryScanner* scanner, const Filter
 static int scanner_inspect_entry(const ScannerOptions* options, const char* source_root,
                                  const char* containing_dir, const char* name,
                                  ScannerEntry* entry) {
+  entry->excluded = false;
   entry->path = path_cat(containing_dir, name);
   if (!entry->path)
     return -1;
@@ -241,19 +278,25 @@ static int scanner_inspect_entry(const ScannerOptions* options, const char* sour
   if (entry->is_directory)
     return 1;
   for (int i = 0; i < options->exclude_count; i++)
-    if (glob_match(options->exclude_patterns[i], name))
+    if (glob_match(options->exclude_patterns[i], name)) {
+      entry->excluded = true;
       goto skip;
+    }
   if (options->include_count > 0) {
     bool included = false;
     for (int i = 0; i < options->include_count; i++)
       if (glob_match(options->include_patterns[i], name))
         included = true;
-    if (!included)
+    if (!included) {
+      entry->excluded = true;
       goto skip;
     }
+  }
   if ((options->max_size > 0 && (unsigned long long)entry->stats.st_size > options->max_size) ||
-      (options->min_size > 0 && (unsigned long long)entry->stats.st_size < options->min_size))
+      (options->min_size > 0 && (unsigned long long)entry->stats.st_size < options->min_size)) {
+    entry->excluded = true;
     goto skip;
+  }
   return 1;
 
 skip:
@@ -306,8 +349,13 @@ DirectoryScanner* directory_scanner_create_with_options(const char* root_directo
   scanner->file_list = options->file_list;
   scanner->base_filters = options->base_filters;
   scanner->per_dir_filters = options->per_dir_filters;
+  scanner->excluded_paths = options->excluded_paths;
+  scanner->excluded_mutex = options->excluded_mutex;
+  scanner->ignore_io_errors = options->ignore_io_errors;
+  scanner->io_error = false;
   scanner->dirs_mode = options->dirs;
   scanner->relative_mode = options->relative && options->file_list != NULL;
+  scanner->prune_empty_dirs = options->prune_empty_dirs;
   scanner->dirs_root_emitted = false;
   scanner->list_index = 0;
   scanner->dirs_batch = NULL;
@@ -360,27 +408,29 @@ DirectoryScanner* directory_scanner_create(const char* root_directory, bool use_
                                            unsigned long long min_size, int max_depth,
                                            bool follow_symlinks, bool copy_links, bool safe_links,
                                            bool copy_unsafe_links, bool checksum) {
-  ScannerOptions options = {use_metadata,
+  ScannerOptions options = {
-                            chunk_size,
+      .use_metadata = use_metadata,
-                            exclude_patterns,
+      .chunk_size = chunk_size,
-                            exclude_count,
+      .exclude_patterns = exclude_patterns,
-                            include_patterns,
+      .exclude_count = exclude_count,
-                            include_count,
+      .include_patterns = include_patterns,
-                            max_size,
+      .include_count = include_count,
-                            min_size,
+      .max_size = max_size,
-                            max_depth,
+      .min_size = min_size,
-                            0,
+      .max_depth = max_depth,
-                            follow_symlinks,
+      .num_threads = 0,
-                            copy_links,
+      .follow_symlinks = follow_symlinks,
-                            safe_links,
+      .copy_links = copy_links,
-                            copy_unsafe_links,
+      .safe_links = safe_links,
-                            checksum,
+      .copy_unsafe_links = copy_unsafe_links,
-                            false,
+      .checksum = checksum,
-                            NULL,
+      .one_file_system = false,
-                            NULL,
+      .file_list = NULL,
-                            false,
+      .base_filters = NULL,
-                            false,
+      .per_dir_filters = false,
-                            false};
+      .dirs = false,
+      .relative = false,
+  };
   return directory_scanner_create_with_options(root_directory, &options);
 }
 
@@ -413,16 +463,21 @@ static Chunk* chunk_data_to_chunk(ArrayList* chunk_data) {
   return chunk;
 }
 
+/* Open the next queued directory and set up its filter context.  Returns 1 when
+   a directory is open, 0 when the queue is exhausted, and -1 on a fatal error.
+   A directory that cannot be opened is an I/O error: it is recorded on the
+   scanner and, when --ignore-errors is active, skipped so the rest of the tree
+   is still scanned (the caller decides whether to treat the recorded error as
+   fatal). */
 static int open_next_directory(DirectoryScanner* scanner) {
   if (scanner->current_dir) {
     closedir(scanner->current_dir);
     scanner->current_dir = NULL;
   }
   free(scanner->current_path);
+  scanner->current_path = NULL;
 
-  if (queue_is_empty(scanner->directories))
+  while (!queue_is_empty(scanner->directories)) {
-    return 0;
-
     DirEntry* de = (DirEntry*)queue_dequeue(scanner->directories);
     scanner->current_path = de->path;
     scanner->current_depth = de->depth;
@@ -438,24 +493,46 @@ static int open_next_directory(DirectoryScanner* scanner) {
     if (!scanner->current_rel) {
       log_message(LOG_LEVEL_ERROR, "Could not compute relative path under %s", scanner->root_path);
       scanner->failed = true;
+      free(scanner->current_path);
+      scanner->current_path = NULL;
       return -1;
     }
 
     scanner->current_dir = opendir(scanner->current_path);
     if (scanner->current_dir == NULL) {
+      scanner->io_error = true;
       log_perror("Could not open directory");
+      /* The transfer ROOT (a sequential scanner's seed directory) must be
+         readable even under --ignore-errors: an unreadable root would produce
+         an empty scan whose keep-set would delete the whole destination.  Only
+         subdirectories discovered during an otherwise-successful root scan are
+         skippable.  (The parallel scanner never reaches this for the root: its
+         root open failure aborts scanner creation; worker seeds are assigned
+         subdirectories with a non-empty relative path and stay skippable.) */
+      bool is_root_seed = scanner->current_rel != NULL && scanner->current_rel[0] == '\0' &&
+                          scanner->current_depth == 0;
+      free(scanner->current_rel);
+      scanner->current_rel = NULL;
       free(scanner->current_path);
       scanner->current_path = NULL;
+      if (!scanner->ignore_io_errors || is_root_seed) {
         scanner->failed = true;
         return -1;
       }
+      /* --ignore-errors: record the I/O error and keep scanning the rest. */
+      continue;
+    }
     if (open_directory_filter_context(scanner, inherited) != 0) {
       closedir(scanner->current_dir);
       scanner->current_dir = NULL;
+      free(scanner->current_path);
+      scanner->current_path = NULL;
       return -1;
     }
     return 1;
   }
+  return 0;
+}
 
 /* ---- --dirs mode ----
    With -d the scanner transfers directory entries and never recurses into
@@ -563,12 +640,35 @@ static File* dirs_file_for_entry(DirectoryScanner* scanner, const char* entry) {
   return file;
 }
 
+/* True when the directory contains no entries at all (ignoring "." and "..").
+   An unreadable directory is reported as non-empty so the regular (erroring)
+   root-entry path runs instead of silently transferring nothing. */
+static bool dirs_source_dir_is_empty(const char* path) {
+  DIR* dir = opendir(path);
+  if (!dir)
+    return false;
+  bool empty = true;
+  const struct dirent* entry;
+  while ((entry = readdir(dir)) != NULL) {
+    if (strcmp(entry->d_name, ".") != 0 && strcmp(entry->d_name, "..") != 0) {
+      empty = false;
+      break;
+    }
+  }
+  closedir(dir);
+  return empty;
+}
+
 /* The next File from the --dirs generator, or NULL when exhausted. */
 static File* dirs_next_file(DirectoryScanner* scanner) {
   if (!scanner->file_list) {
     if (scanner->dirs_root_emitted)
       return NULL;
     scanner->dirs_root_emitted = true;
+    /* --prune-empty-dirs: a physically empty source directory's explicit entry
+       would only create an empty destination directory, so it is omitted. */
+    if (scanner->prune_empty_dirs && dirs_source_dir_is_empty(scanner->root_path))
+      return NULL;
     return dirs_root_dir_file(scanner);
   }
   while (scanner->list_index < scanner->file_list->count) {
@@ -663,27 +763,29 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) {
     if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
       continue;
 
-    ScannerOptions options = {scanner->use_metadata,
+    ScannerOptions options = {
-                              scanner->chunk_size,
+        .use_metadata = scanner->use_metadata,
-                              scanner->exclude_patterns,
+        .chunk_size = scanner->chunk_size,
-                              scanner->exclude_count,
+        .exclude_patterns = scanner->exclude_patterns,
-                              scanner->include_patterns,
+        .exclude_count = scanner->exclude_count,
-                              scanner->include_count,
+        .include_patterns = scanner->include_patterns,
-                              scanner->max_size,
+        .include_count = scanner->include_count,
-                              scanner->min_size,
+        .max_size = scanner->max_size,
-                              scanner->max_depth,
+        .min_size = scanner->min_size,
-                              0,
+        .max_depth = scanner->max_depth,
-                              scanner->follow_symlinks,
+        .num_threads = 0,
-                              scanner->copy_links,
+        .follow_symlinks = scanner->follow_symlinks,
-                              scanner->safe_links,
+        .copy_links = scanner->copy_links,
-                              scanner->copy_unsafe_links,
+        .safe_links = scanner->safe_links,
-                              scanner->checksum,
+        .copy_unsafe_links = scanner->copy_unsafe_links,
-                              scanner->one_file_system,
+        .checksum = scanner->checksum,
-                              scanner->file_list,
+        .one_file_system = scanner->one_file_system,
-                              scanner->base_filters,
+        .file_list = scanner->file_list,
-                              scanner->per_dir_filters,
+        .base_filters = scanner->base_filters,
-                              false,
+        .per_dir_filters = scanner->per_dir_filters,
-                              false};
+        .dirs = false,
+        .relative = false,
+    };
     ScannerEntry inspected;
     int inspection = scanner_inspect_entry(&options, scanner->current_path, scanner->current_path,
                                            entry->d_name, &inspected);
@@ -691,8 +793,21 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) {
       scanner->failed = true;
       break;
     }
-    if (inspection == 0)
+    if (inspection == 0) {
+      /* The entry was pruned by a user selection rule (exclude/include/size) or
+         skipped for another reason; only the user-selection prunes protect the
+         corresponding destination mirror from --delete. */
+      if (inspected.excluded) {
+        char* abs_path = path_cat(scanner->current_path, entry->d_name);
+        if (!abs_path) {
+          scanner->failed = true;
+          break;
+        }
+        scanner_record_excluded(scanner, abs_path);
+        free(abs_path);
+      }
       continue;
+    }
     char* cur_path = inspected.path;
     struct stat stats = inspected.stats;
 
@@ -708,6 +823,16 @@ Chunk* directory_scanner_next(DirectoryScanner* scanner) {
     bool passes_selection =
         entry_passes_selection(scanner->file_list, scanner->base_filters, scanner->current_node,
                                rel, entry->d_name, is_dir, scanner->per_dir_filters);
+    if (!passes_selection) {
+      /* --files-from subset pruning is not a filter exclusion: its delete
+         semantics stay keep-set-only (an unlisted source path is treated as
+         absent, so its destination mirror is a deletable extra).  A rule-based
+         exclusion is recorded as a protected prefix.  -R + --files-from bare
+         wire paths are never recorded (see ScannerOptions.excluded_paths). */
+      bool files_from_prune = scanner->file_list && !file_list_affects(scanner->file_list, rel);
+      if (!files_from_prune && !scanner->relative_mode)
+        scanner_record_excluded(scanner, cur_path);
+    }
     /* With -R + --files-from the wire/destination path is the entry's bare
        relative path; keep `rel` alive to attach it to a transferred file. */
     char* rel_copy = scanner->relative_mode ? str_dup(rel) : NULL;
@@ -796,6 +921,10 @@ bool directory_scanner_failed(const DirectoryScanner* scanner) {
   return scanner == NULL || scanner->failed;
 }
 
+bool directory_scanner_had_io_error(const DirectoryScanner* scanner) {
+  return scanner != NULL && scanner->io_error;
+}
+
 typedef struct {
   ParallelScanner* ps;
   char** dirs;
@@ -826,10 +955,12 @@ static int parallel_worker_thread(void* arg) {
     /* Root .rsync-filter rules (parsed by the parallel scanner) apply to the
      * contents of every assigned subdirectory. Relative paths (used by the
      * allow-set and per-directory rules) are computed against the transfer
-     * root, not the subdirectory the worker is seeded with. */
+     * root, not the subdirectory the worker is seeded with.  Exclusion
+     * recording shares one caller-owned list across the workers. */
     free(ds->root_path);
     ds->root_path = str_dup(wa->root_dir);
     ds->seed_node = wa->ps->root_filter_node;
+    ds->excluded_mutex = &wa->ps->result_mutex;
     Chunk* chunk;
     while ((chunk = directory_scanner_next(ds)) != NULL) {
       if (!queue_enqueue_multithreaded_cancel(wa->ps->result_queue, chunk, &wa->ps->result_mutex,
@@ -846,6 +977,11 @@ static int parallel_worker_thread(void* arg) {
       cnd_broadcast(&wa->ps->result_not_empty);
       cnd_broadcast(&wa->ps->result_not_full);
       mtx_unlock(&wa->ps->result_mutex);
+    } else if (directory_scanner_had_io_error(ds)) {
+      /* --ignore-errors path: an unreadable directory was skipped, not fatal. */
+      mtx_lock(&wa->ps->result_mutex);
+      wa->ps->io_error = true;
+      mtx_unlock(&wa->ps->result_mutex);
     }
     directory_scanner_destroy(ds);
     free(wa->dirs[i]);
@@ -993,8 +1129,23 @@ static void scan_root_entry(const ScannerOptions* options, const FilterNode* roo
     ps->failed = true;
     return;
   }
-  if (inspection == 0)
+  if (inspection == 0) {
+    if (inspected.excluded && options->excluded_paths) {
+      /* A root-level user-selection prune protects the destination mirror of
+         the same-named wire path (at the root the bare name is the wire path in
+         every layout). */
+      char* abs_path = path_cat(root_directory, entry->d_name);
+      if (!abs_path) {
+        ps->failed = true;
+      } else {
+        const char* rel = *abs_path == '/' ? abs_path + 1 : abs_path;
+        if (!excluded_sink_append(options->excluded_paths, options->excluded_mutex, rel))
+          ps->failed = true;
+        free(abs_path);
+      }
+    }
     return;
+  }
   char* cur_path = inspected.path;
   struct stat st = inspected.stats;
   bool is_dir = inspected.is_directory;
@@ -1009,6 +1160,14 @@ static void scan_root_entry(const ScannerOptions* options, const FilterNode* roo
   /* -R + --files-from: root-level files keep their bare relative send path. */
   bool use_rel = options->relative && options->file_list != NULL;
   if (!passes) {
+    /* --files-from subset pruning is not a filter exclusion; -R bare-wire-path
+       exclusions are never recorded (see ScannerOptions.excluded_paths). */
+    bool files_from_prune = options->file_list && !file_list_affects(options->file_list, rel);
+    if (!files_from_prune && !use_rel && options->excluded_paths) {
+      const char* rel_path = *cur_path == '/' ? cur_path + 1 : cur_path;
+      if (!excluded_sink_append(options->excluded_paths, options->excluded_mutex, rel_path))
+        ps->failed = true;
+    }
     free(rel);
     free(cur_path);
     return;
@@ -1258,6 +1417,10 @@ bool parallel_scanner_failed(const ParallelScanner* ps) {
   return ps == NULL || ps->failed;
 }
 
+bool parallel_scanner_had_io_error(const ParallelScanner* ps) {
+  return ps != NULL && ps->io_error;
+}
+
 void parallel_scanner_destroy(ParallelScanner* ps) {
   if (!ps)
     return;

src/client/scanner.h

@@ -37,6 +37,28 @@ typedef struct {
   bool per_dir_filters;               /* -F: read .rsync-filter per directory */
   bool dirs;                          /* -d/--dirs: transfer dir entries, no recursion */
   bool relative;                      /* -R/--relative (dest rel paths, with --files-from) */
+  /* --prune-empty-dirs (long only): in --dirs mode an empty source directory's
+     explicit entry is omitted from the transfer file list (so nothing is
+     created at the destination and it can be pruned by --delete); explicitly
+     --files-from-listed directories always pass through.  Recursive transfers
+     never emit empty directories, so the flag has no additional effect there. */
+  bool prune_empty_dirs;
+  /* Delete-excluded protection sink (optional): when non-NULL the scanner
+   * appends the destination-relative path of every entry it prunes because a
+   * USER SELECTION rule excluded it (--filter/-C/per-dir rules, the legacy
+   * --exclude/--include layer, and --max-size/--min-size).  The sender turns
+   * this list into the manifest's protected prefixes so `--delete` leaves the
+   * destination mirror of excluded source paths alone (rsync's default), and
+   * empties it when --delete-excluded opts back into deleting them.  NOT
+   * recorded for --files-from subset pruning (whose delete semantics stay
+   * keep-set-only) or for -R/--files-from relative wire paths.  When
+   * `excluded_mutex` is non-NULL it is taken around every append (the parallel
+   * scanner shares one list across its worker threads). */
+  ArrayList* excluded_paths;
+  mtx_t* excluded_mutex;
+  /* --ignore-errors: an unreadable directory during the scan is recorded as an
+   * I/O error and skipped instead of aborting the scan.  Client-only. */
+  bool ignore_io_errors;
 } ScannerOptions;
 
 /* Internal per-scanner filter state. FilterNode chains represent the ordered
@@ -79,10 +101,21 @@ typedef struct {
      the recursive scan). */
   bool dirs_mode;
   bool relative_mode; /* file_list && relative: send bare relative wire paths */
+  bool prune_empty_dirs;
   bool dirs_root_emitted;
   int list_index;
   ArrayList* dirs_batch; /* owned when non-NULL */
   unsigned long long dirs_batch_size;
+  /* Excluded-path sink (see ScannerOptions). `excluded_mutex` is shared across
+     parallel worker threads. */
+  ArrayList* excluded_paths;
+  mtx_t* excluded_mutex;
+  /* --ignore-errors: continue past unreadable directories (records io_error). */
+  bool ignore_io_errors;
+  /* A directory could not be opened (I/O error, e.g. EACCES).  With
+     --ignore-errors the scan continues past it and the caller decides what to
+     do; `failed` is reserved for fatal errors that always abort the scan. */
+  bool io_error;
 } DirectoryScanner;
 
 typedef struct {
@@ -96,6 +129,8 @@ typedef struct {
   thrd_t* threads;
   bool done;
   bool failed;
+  /* A worker skipped an unreadable directory under --ignore-errors (non-fatal). */
+  bool io_error;
   atomic_bool cancelled;
   int completed;
   Chunk* initial_chunk;
@@ -131,6 +166,11 @@ ParallelScanner* parallel_scanner_create_with_options(const char* root_directory
                                                       ProtocolSession* allocation_session);
 Chunk* parallel_scanner_next(ParallelScanner* scanner);
 bool parallel_scanner_failed(const ParallelScanner* scanner);
+bool parallel_scanner_had_io_error(const ParallelScanner* scanner);
 void parallel_scanner_destroy(ParallelScanner* scanner);
 
+/* True when a directory could not be opened during the scan (an I/O error,
+   recorded even when --ignore-errors keeps the scan going past it). */
+bool directory_scanner_had_io_error(const DirectoryScanner* scanner);
+
 #endif

src/client/usage.c

@@ -35,6 +35,20 @@ void print_usage(void) {
   printf("                      (implies --delete)\n");
   printf("  --delete-after      Delete only after the whole transfer succeeded\n");
   printf("                      (the default --delete timing; implies --delete)\n");
+  printf("  --delete-excluded   Also delete destination files that were excluded on\n");
+  printf("                      the source (default protects them, matching rsync)\n");
+  printf("  --max-delete=NUM    Never delete more than NUM destination entries per run;\n");
+  printf("                      if the extras would exceed NUM, nothing is deleted and\n");
+  printf("                      the run fails with a clear error (implies --delete only\n");
+  printf("                      when used with it)\n");
+  printf("  --ignore-errors     Continue (and still delete) when a source directory is\n");
+  printf("                      unreadable during the scan, instead of aborting with no\n");
+  printf("                      deletion\n");
+  printf("  --force             A file may replace a destination directory by removing\n");
+  printf("                      that (non-empty) directory first\n");
+  printf("  --prune-empty-dirs  Do not transfer empty directory entries (--dirs mode);\n");
+  printf("                      recursive transfers never send empty dirs.  rsync's -m\n");
+  printf("                      short form stays FastSync multithreading\n");
   printf("  Note: each timing flag implies --delete.  Combining a timing flag with\n");
   printf("  --no-delete (in either order) is rejected as a config error.\n");
   printf("  --ignore-existing  Skip files that already exist on receiver\n");

src/server/receiver.c

@@ -143,7 +143,7 @@ int receiver_process(Config* config, int file_descriptor, const ReceiverSink* si
    the whole transfer succeeded.  See receiver_process_pending() for how the -m
    receiver defers that commit until its disk writer has drained. */
 int receiver_process_pending(Config* config, int file_descriptor, const ReceiverSink* sink,
-                             ArrayList** pending_manifest) {
+                             DeleteManifest** pending_manifest) {
   Status status;
   if (!receive_status(file_descriptor, &status))
     return -1;
@@ -151,7 +151,7 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
   /* Parked keep-set for the late/commit timing.  Every exit path below frees it
      exactly once; the only exception is the successful FINISHED handoff, which
      transfers ownership to *pending_manifest (used by the -m receiver). */
-  ArrayList* deferred_manifest = NULL;
+  DeleteManifest* deferred_manifest = NULL;
   while (status == STATUS_NEXT || status == STATUS_CHUNK || status == STATUS_CHECK ||
          status == STATUS_KEEPALIVE || status == STATUS_ABORT || status == STATUS_CHECK_BATCH ||
          status == STATUS_MKDIR || status == STATUS_MANIFEST) {
@@ -182,7 +182,7 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
       if (!dir || !sink->store_file(dir, sink->context))
         goto receive_error;
     } else if (status == STATUS_MANIFEST) {
-      ArrayList* manifest = receive_manifest_entries(file_descriptor);
+      DeleteManifest* manifest = receive_manifest_entries(file_descriptor);
       if (!manifest)
         goto fail; /* receive_manifest_entries already sent STATUS_ERROR */
       if (early_delete) {
@@ -192,7 +192,7 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
            failed).  This is the rsync delete-before/delete-during window: a
            later transfer failure does not restore these deletions. */
         bool deletion_ok = config->use_delete ? manifest_delete_extras(config, manifest) : true;
-        array_list_delete(manifest);
+        delete_manifest_free(manifest);
         if (!deletion_ok) {
           send_status(file_descriptor, STATUS_ERROR);
           goto fail;
@@ -204,15 +204,15 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
            and commit the deletion only after STATUS_FINISHED. */
         if (deferred_manifest) {
           log_message(LOG_LEVEL_ERROR, "Received a second delete manifest");
-          array_list_delete(deferred_manifest);
+          delete_manifest_free(deferred_manifest);
           deferred_manifest = NULL;
-          array_list_delete(manifest);
+          delete_manifest_free(manifest);
           send_status(file_descriptor, STATUS_ERROR);
           goto fail;
         }
         deferred_manifest = manifest;
       } else {
-        array_list_delete(manifest);
+        delete_manifest_free(manifest);
       }
       goto next_status;
     } else {
@@ -247,7 +247,7 @@ int receiver_process_pending(Config* config, int file_descriptor, const Receiver
       deferred_manifest = NULL;
     } else {
       bool deletion_ok = manifest_delete_extras(config, deferred_manifest);
-      array_list_delete(deferred_manifest);
+      delete_manifest_free(deferred_manifest);
       deferred_manifest = NULL;
       if (!deletion_ok) {
         send_status(file_descriptor, STATUS_ERROR);
@@ -269,14 +269,14 @@ fail:
   /* Failure exits that must not (or already did) report a STATUS_ERROR.  The
      parked keep-set is dropped: never commit a deletion for a failed stream. */
   if (deferred_manifest) {
-    array_list_delete(deferred_manifest);
+    delete_manifest_free(deferred_manifest);
     deferred_manifest = NULL;
   }
   return -1;
 
 receive_error:
   if (deferred_manifest) {
-    array_list_delete(deferred_manifest);
+    delete_manifest_free(deferred_manifest);
     deferred_manifest = NULL;
   }
   if (sink->send_error)

src/server/receiver.h

@@ -42,7 +42,7 @@ int receiver_process(Config* config, int file_descriptor, const ReceiverSink* si
    commit the deletion only after its disk writer has fully drained.  Pass NULL
    to keep the default behaviour (delete before the success frame). */
 int receiver_process_pending(Config* config, int file_descriptor, const ReceiverSink* sink,
-                             ArrayList** pending_manifest);
+                             DeleteManifest** pending_manifest);
 int receiver_receive_files(Config* config, int file_descriptor);
 
 #endif

src/server/server.c

@@ -265,7 +265,7 @@ void handler(int file_descriptor) {
         if (!manifest_delete_extras(config, context->deferred_manifest)) {
           transfer_ok = false;
         }
-        array_list_delete(context->deferred_manifest);
+        delete_manifest_free(context->deferred_manifest);
         context->deferred_manifest = NULL;
       }
     }

src/shared/config.c

@@ -92,7 +92,9 @@ static void config_set_defaults(Config* config) {
   config->append_verify = false;
   config->delete_excluded = false;
   config->delete_after = false;
-  config->max_delete = 0;
+  config->max_delete = -1;
+  config->ignore_errors = false;
+  config->force_delete = false;
   config->filters = NULL;
   config->files_from = NULL;
   config->files_from_set = NULL;
@@ -161,11 +163,11 @@ static bool validate_received_config(const Config* config) {
          valid_wire_bool(config->existing) && valid_wire_bool(config->update) &&
          valid_wire_bool(config->inplace) && valid_wire_bool(config->append) &&
          valid_wire_bool(config->use_fsync) && valid_wire_bool(config->append_verify) &&
-         valid_wire_bool(config->delete_excluded) && valid_wire_bool(config->delete_after) &&
+         valid_wire_bool(config->delete_excluded) && valid_wire_bool(config->force_delete) &&
-         valid_wire_bool(config->delete_delay) && valid_wire_bool(config->delete_during) &&
+         valid_wire_bool(config->delete_after) && valid_wire_bool(config->delete_delay) &&
-         valid_wire_bool(config->relative) && valid_wire_bool(config->prune_empty_dirs) &&
+         valid_wire_bool(config->delete_during) && valid_wire_bool(config->relative) &&
-         valid_wire_bool(config->delay_updates) && valid_wire_bool(config->mkpath) &&
+         valid_wire_bool(config->prune_empty_dirs) && valid_wire_bool(config->delay_updates) &&
-         !(config->delay_updates && config->inplace) &&
+         valid_wire_bool(config->mkpath) && !(config->delay_updates && config->inplace) &&
          !(config->delay_updates && delay_updates_staging_name_conflict(config->backup_dir)) &&
          valid_wire_bool(config->partial) && valid_wire_bool(config->delete_before) &&
          valid_wire_bool(config->checksum) && valid_wire_bool(config->eight_bit_output) &&
@@ -177,7 +179,7 @@ static bool validate_received_config(const Config* config) {
          config->delta_block_size >= DELTA_BLOCK_SIZE_MIN &&
          config->delta_block_size <= DELTA_BLOCK_SIZE_MAX &&
          config->delta_max_file_size <= DELTA_MAX_FILE_SIZE && config->modify_window >= 0 &&
-         config->max_delete >= 0 && config->skip_compress_count >= 0 &&
+         config->max_delete >= -1 && config->skip_compress_count >= 0 &&
          config->skip_compress_count <= 10000 && config->max_alloc > 0 &&
          (!config->chmod_spec || !*config->chmod_spec ||
           chmod_apply(0, config->chmod_spec, &(mode_t){0}));
@@ -417,10 +419,10 @@ static bool send_selection_options(int fd, const Config* c) {
   return send_int(fd, c->ignore_existing) && send_int(fd, c->existing) && send_int(fd, c->update) &&
          send_int(fd, c->inplace) && send_int(fd, c->delay_updates) && send_int(fd, c->append) &&
          send_int(fd, c->use_fsync) && send_int(fd, c->append_verify) &&
-         send_int(fd, c->delete_excluded) && send_int(fd, c->delete_after) &&
+         send_int(fd, c->delete_excluded) && send_int(fd, c->force_delete) &&
-         send_n_data(fd, &c->max_delete, sizeof(c->max_delete)) && send_int(fd, c->relative) &&
+         send_int(fd, c->delete_after) && send_n_data(fd, &c->max_delete, sizeof(c->max_delete)) &&
-         send_int(fd, c->prune_empty_dirs) && send_int(fd, c->mkpath) &&
+         send_int(fd, c->relative) && send_int(fd, c->prune_empty_dirs) &&
-         send_int(fd, c->delete_during) && send_int(fd, c->delete_delay);
+         send_int(fd, c->mkpath) && send_int(fd, c->delete_during) && send_int(fd, c->delete_delay);
 }
 
 static bool send_skip_compress_options(int fd, const Config* c) {
@@ -525,7 +527,7 @@ static bool receive_file_options(int fd, Config* c) {
 static bool receive_selection_options(int fd, Config* c) {
   bool* flags[] = {&c->ignore_existing, &c->existing,     &c->update,      &c->inplace,
                    &c->delay_updates,   &c->append,       &c->use_fsync,   &c->append_verify,
-                   &c->delete_excluded, &c->delete_after};
+                   &c->delete_excluded, &c->force_delete, &c->delete_after};
   for (size_t i = 0; i < sizeof(flags) / sizeof(flags[0]); i++) {
     if (!receive_wire_bool(fd, flags[i]))
       return false;

src/shared/config.h

@@ -117,9 +117,26 @@ typedef struct Config {
   bool append_verify;
 
   // Issue #128: Extended delete options
+  /* --delete-excluded: also delete destination entries that were excluded on
+   * the source.  Default (off) matches rsync: excluded paths are protected from
+   * deletion.  Crosses the wire (the sender encodes the choice by whether it
+   * transmits a protected-prefix list with the keep-set manifest). */
   bool delete_excluded;
   bool delete_after;
+  /* --max-delete=NUM: the receiver refuses to delete more than NUM entries per
+   * run (all-or-nothing: when the extras would exceed NUM nothing is removed and
+   * the transfer fails with a distinct error).  -1 == no client limit (the
+   * server hard bound MAX_SERVER_DELETE_COUNT still applies). */
   int max_delete;
+  /* --ignore-errors (client-only, never serialized): a sender-side source I/O
+   * error (an unreadable directory during the scan) normally aborts the run so
+   * no deletion happens; with --ignore-errors the scan continues and the
+   * (partial) keep-set is still transmitted so the deletion runs. */
+  bool ignore_errors;
+  /* --force (receiver-side): a regular file may replace a destination
+   * directory by removing that (possibly non-empty, symlink-safe) directory
+   * tree first, instead of failing the write.  Crosses the wire. */
+  bool force_delete;
 
   // Issue #129: Advanced file selection. These fields are CLIENT-ONLY: they are
   // never serialized to the wire (the receiver must not learn them).
@@ -206,7 +223,7 @@ typedef struct Config {
   DelayUpdatesContext* delay_context;
 } Config;
 
-#define PROTOCOL_VERSION "2.8.0"
+#define PROTOCOL_VERSION "2.9.0"
 #define DEFAULT_CHUNK_SIZE (10 * 1024 * 1024)
 /* Upper bound on total basis-dir entries (rsync caps --link-dest at 20). */
 #define MAX_BASIS_DIRS 64

src/shared/file.c

@@ -1,4 +1,5 @@
 #include <errno.h>
+#include <dirent.h>
 #include <fcntl.h>
 #include <libgen.h>
 #include <limits.h>
@@ -410,6 +411,79 @@ bool file_rename_secure(const char* old_path, const char* new_path) {
   return ok;
 }
 
+/* Recursively delete every entry inside an open directory, never following a
+   symlink (an O_NOFOLLOW fd walk, so a symlink planted inside the tree can
+   never redirect removal outside of it).  The directory itself is left in
+   place; returns false on any failure. */
+static bool wipe_dir_fd(int dirfd) {
+  int scanfd = dup(dirfd);
+  if (scanfd < 0)
+    return false;
+  DIR* dir = fdopendir(scanfd);
+  if (!dir) {
+    close(scanfd);
+    return false;
+  }
+  bool operation_ok = true;
+  const struct dirent* entry;
+  while ((entry = readdir(dir)) != NULL) {
+    if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
+      continue;
+    struct stat st;
+    if (fstatat(dirfd, entry->d_name, &st, AT_SYMLINK_NOFOLLOW) != 0) {
+      if (errno != ENOENT)
+        operation_ok = false;
+      continue;
+    }
+    if (S_ISDIR(st.st_mode)) {
+      int childfd = openat(dirfd, entry->d_name, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+      bool child_removed = false;
+      if (childfd >= 0) {
+        child_removed = wipe_dir_fd(childfd);
+        close(childfd);
+      } else if (errno != ENOENT) {
+        operation_ok = false;
+      }
+      if (child_removed && unlinkat(dirfd, entry->d_name, AT_REMOVEDIR) != 0 && errno != ENOENT)
+        operation_ok = false;
+    } else {
+      /* Files and symlinks alike are removed by name, never followed. */
+      if (unlinkat(dirfd, entry->d_name, 0) != 0 && errno != ENOENT)
+        operation_ok = false;
+    }
+  }
+  closedir(dir);
+  return operation_ok;
+}
+
+/* Remove the whole directory tree at `path` (confined below the authorized
+   root, symlink-safe).  --force uses this to clear a non-empty destination
+   directory that blocks an incoming regular file.  Returns true when the path
+   no longer exists as a directory (a missing path or a non-directory at the
+   final component is a no-op success; the normal write path replaces files). */
+bool file_remove_tree_secure(const char* path) {
+  if (!path)
+    return false;
+  char* leaf = NULL;
+  int parent_fd = file_open_secure_parent(path, &leaf, false);
+  if (parent_fd < 0)
+    return false;
+  int dirfd = openat(parent_fd, leaf, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+  if (dirfd < 0) {
+    bool absent = errno == ENOENT || errno == ENOTDIR || errno == ELOOP;
+    close(parent_fd);
+    free(leaf);
+    return absent;
+  }
+  bool ok = wipe_dir_fd(dirfd);
+  close(dirfd);
+  if (ok && unlinkat(parent_fd, leaf, AT_REMOVEDIR) != 0 && errno != ENOENT)
+    ok = false;
+  close(parent_fd);
+  free(leaf);
+  return ok;
+}
+
 /* Open a private staging/scratch directory, creating it (and any missing path
    components) on demand.  dir_path is expected to already be confined below
    the authorized root by the caller; file_open_secure_parent re-checks that

src/shared/file.h

@@ -32,6 +32,10 @@ int file_open_secure_parent(const char* path, char** leaf_out, bool create_dirs)
 bool file_ensure_directory_secure(const char* path);
 bool file_directory_exists_secure(const char* path);
 bool file_rename_secure(const char* old_path, const char* new_path);
+/* Remove the whole directory tree at `path` (confined, symlink-safe).  Used by
+   --force to clear a non-empty destination directory that blocks an incoming
+   regular file.  See the .c for the exact success semantics. */
+bool file_remove_tree_secure(const char* path);
 /* Open a private 0700 directory (creating it on demand) that must live below
    the authorized root.  Used for the --temp-dir scratch directory and the
    --delay-updates staging directory. */

src/shared/file_receive.c

@@ -218,6 +218,20 @@ FileSaveResult file_save_to_disk_full(const char* root_directory, const File* fi
     return FILE_SAVE_SKIPPED;
   }
 
+  /* --force (rsync semantics): an incoming regular file may replace a
+     destination DIRECTORY by removing that (possibly non-empty, symlink-safe)
+     tree first, so the atomic temp+rename below can install the file.  Only the
+     immediate-install path does this: a --delay-updates run stages into its own
+     tree and is unaffected here (its publication renames over regular files
+     only).  The blocking directory is removed only after the --update /
+     --existing / --ignore-existing decisions above, which see it as an existing
+     destination entry. */
+  if (config && config->force_delete && !file->is_dir &&
+      file_directory_exists_secure(destination_path)) {
+    if (!file_remove_tree_secure(destination_path))
+      goto fail;
+  }
+
   if (backup_enabled) {
     /* Back up the entry that the incoming write will replace.  When writing
        through a partial dir the pre-existing destination file is the one to
@@ -1021,63 +1035,93 @@ File* file_receive_directory(int file_descriptor) {
 }
 
 /* Read a delete-manifest frame (the STATUS_MANIFEST leading code has already
-   been consumed): an entry count followed by that many destination-relative
+   been consumed): a keep-set entry count followed by that many
-   paths.  The frame is self-delimiting (the count is authoritative), so the
+   destination-relative paths, then a protected-prefix count followed by that
-   caller decides what to do next and continues reading the following STATUS_*
+   many destination-relative prefixes.  The frame is self-delimiting (the counts
-   frame.  Returns an owned ArrayList of validated path strings, or NULL after
+   are authoritative), so the caller decides what to do next and continues
-   sending STATUS_ERROR when the frame is malformed (bad count, empty/absolute
+   reading the following STATUS_* frame.  Returns an owned DeleteManifest, or
-   path, path traversal, or an aggregate size beyond MAX_MANIFEST_BYTES). */
+   NULL after sending STATUS_ERROR when the frame is malformed (bad count,
-ArrayList* receive_manifest_entries(int fd) {
+   empty/absolute path, path traversal, or an aggregate size beyond
+   MAX_MANIFEST_BYTES). */
+static bool receive_manifest_section(int fd, ArrayList* list, size_t* manifest_bytes) {
   int count;
   if (!receive_int(fd, &count)) {
     send_status(fd, STATUS_ERROR);
-    return NULL;
+    return false;
   }
   if (count < 0 || count > MAX_MANIFEST_ENTRIES) {
     send_status(fd, STATUS_ERROR);
-    return NULL;
+    return false;
   }
-  ArrayList* manifest = array_list_create(free);
-  if (!manifest) {
-    send_status(fd, STATUS_ERROR);
-    return NULL;
-  }
-  size_t manifest_bytes = 0;
   for (int i = 0; i < count; i++) {
     char* s = receive_str(fd);
     size_t entry_size = s ? strlen(s) : 0;
     if (!s || s[0] == '\0' || s[0] == '/' || has_path_traversal(s) ||
-        entry_size > MAX_MANIFEST_BYTES - manifest_bytes ||
+        entry_size > MAX_MANIFEST_BYTES - *manifest_bytes ||
-        (manifest_bytes += entry_size) > MAX_MANIFEST_BYTES || !array_list_add(manifest, s)) {
+        (*manifest_bytes += entry_size) > MAX_MANIFEST_BYTES || !array_list_add(list, s)) {
       free(s);
-      array_list_delete(manifest);
+      send_status(fd, STATUS_ERROR);
+      return false;
+    }
+  }
+  return true;
+}
+
+DeleteManifest* receive_manifest_entries(int fd) {
+  DeleteManifest* manifest = calloc(1, sizeof(DeleteManifest));
+  if (!manifest) {
     send_status(fd, STATUS_ERROR);
     return NULL;
   }
+  manifest->keeps = array_list_create(free);
+  manifest->protected = array_list_create(free);
+  if (!manifest->keeps || !manifest->protected) {
+    delete_manifest_free(manifest);
+    send_status(fd, STATUS_ERROR);
+    return NULL;
+  }
+  size_t manifest_bytes = 0;
+  if (!receive_manifest_section(fd, manifest->keeps, &manifest_bytes) ||
+      !receive_manifest_section(fd, manifest->protected, &manifest_bytes)) {
+    delete_manifest_free(manifest);
+    return NULL;
   }
   return manifest;
 }
 
-/* Remove every destination entry under the receive root that is not listed in
+void delete_manifest_free(DeleteManifest* manifest) {
-   `manifest`, bounded by MAX_SERVER_DELETE_COUNT, using the symlink-safe
+  if (!manifest)
-   delete walker.  With --delay-updates the not-yet-published staging directory
+    return;
-   is a direct child of the receive root and must not be treated as a set of
+  array_list_delete(manifest->keeps);
-   extras.  Prints a notice and returns true on success. */
+  array_list_delete(manifest->protected);
-bool manifest_delete_extras(const Config* config, ArrayList* manifest) {
+  free(manifest);
-  if (!config || !manifest)
+}
+
+/* Remove every destination entry under the receive root that is not in the
+   keep-set, bounded by MAX_SERVER_DELETE_COUNT (or a smaller client
+   --max-delete=NUM, which is all-or-nothing), using the symlink-safe delete
+   walker.  With --delay-updates the not-yet-published staging directory is a
+   direct child of the receive root and must not be treated as a set of extras;
+   the manifest's protected prefixes (paths excluded on the source) and the
+   alternate basis directories are never destination content and are skipped at
+   any depth.  Prints a notice and returns true on success. */
+bool manifest_delete_extras(const Config* config, DeleteManifest* manifest) {
+  if (!config || !manifest || !manifest->keeps)
     return false;
   fprintf(stderr, "Deleting files not in manifest...\n");
-  /* With --delay-updates the staged (not yet published) files live directly
+  /* Protected entries:
-     under the receive root in the staging directory; the delete walker must
+     - the --delay-updates staging name, protected only as a DIRECT child of the
-     not treat them as extras or it would remove every staged file before it
+       receive root (a nested destination directory that happens to be named
-     can be published.  That staging name is protected only as a DIRECT child
+       .fastsync-stage is ordinary content);
-     of the receive root so a nested destination directory that happens to be
+     - alternate basis directories (--compare-dest / --copy-dest / --link-dest)
-     named .fastsync-stage is still ordinary content.  Alternate basis
+       at any depth: they are extra comparison snapshots the user pointed at,
-     directories (--compare-dest / --copy-dest / --link-dest) are excluded at
+       not destination content, and deleting them would destroy the very files a
-     any depth: they are extra comparison snapshots the user pointed at, not
+       --link-dest run just linked into place;
-     destination content, and deleting them would destroy the very files a
+     - the sender-side protected prefixes (source paths excluded by filters), at
-     --link-dest run just linked into place. */
+       any depth, so an excluded destination mirror survives --delete unless
-  int skip_count = (config->delay_updates ? 1 : 0) + config->basis_count;
+       --delete-excluded opts back into removing it. */
+  int skip_count = (config->delay_updates ? 1 : 0) + config->basis_count +
+                   (manifest->protected ? manifest->protected->size : 0);
   DeleteSkipEntry* skips = NULL;
   if (skip_count > 0) {
     skips = calloc((size_t)skip_count, sizeof(DeleteSkipEntry));
@@ -1094,9 +1138,40 @@ bool manifest_delete_extras(const Config* config, ArrayList* manifest) {
       skips[idx].top_level_only = false;
       idx++;
     }
+    for (int i = 0; i < manifest->protected->size; i++) {
+      skips[idx].prefix = (const char*)manifest->protected->items[i];
+      skips[idx].top_level_only = false;
+      idx++;
     }
-  bool deletion_ok = delete_extras_limited(config->receive_root_directory, manifest,
+  }
-                                           MAX_SERVER_DELETE_COUNT, skips, skip_count);
+  /* A client --max-delete=NUM smaller than the server's hard bound replaces it
+     for this run; both still bound the walk.  The walker is all-or-nothing, so
+     a run that would delete more than the bound removes nothing and fails with
+     an error that names the bound that was hit. */
+  bool user_limited =
+      config->max_delete >= 0 && (size_t)config->max_delete < MAX_SERVER_DELETE_COUNT;
+  size_t cap = user_limited ? (size_t)config->max_delete : MAX_SERVER_DELETE_COUNT;
+  size_t deleted_count = 0;
+  DeleteWalkResult result = delete_extras_limited(config->receive_root_directory, manifest->keeps,
+                                                  cap, skips, skip_count, &deleted_count);
   free(skips);
-  return deletion_ok;
+  if (result == DELETE_WALK_LIMIT_EXCEEDED) {
+    if (user_limited) {
+      log_message(LOG_LEVEL_ERROR,
+                  "deletion stopped: the destination holds more than --max-delete=%d extraneous "
+                  "entries; no files were deleted",
+                  config->max_delete);
+    } else {
+      log_message(LOG_LEVEL_ERROR,
+                  "deletion stopped: the destination holds more than %u extraneous entries "
+                  "(server deletion limit); no files were deleted",
+                  (unsigned)MAX_SERVER_DELETE_COUNT);
+    }
+    return false;
+  }
+  if (result != DELETE_WALK_OK) {
+    log_message(LOG_LEVEL_ERROR, "deletion failed while removing extraneous files");
+    return false;
+  }
+  return true;
 }

src/shared/file_receive.h

@@ -10,14 +10,30 @@
 File* file_receive(const Config* config, int file_descriptor);
 File* file_receive_directory(int file_descriptor);
 File* receive_incremental_check(int fd, const Config* config, bool* skipped);
-/* Read a delete-manifest frame: entry count then paths (self-delimiting; the
+
-   leading STATUS_MANIFEST code has been consumed).  Returns an owned path
+/* A received delete-manifest frame: the keep-set (`keeps`, destination-relative
-   ArrayList, or NULL after signalling STATUS_ERROR on a malformed frame. */
+   paths the sender transferred/keeps) plus `protected`, destination-relative
-ArrayList* receive_manifest_entries(int fd);
+   prefixes the sender asks the receiver never to delete (paths excluded on the
+   source, protected at any depth).  When --delete-excluded is given the sender
+   transmits an empty protected list so excluded destination mirrors are treated
+   as ordinary extras. */
+typedef struct DeleteManifest {
+  ArrayList* keeps;
+  ArrayList* protected;
+} DeleteManifest;
+
+void delete_manifest_free(DeleteManifest* manifest);
+/* Read a delete-manifest frame: keep count + keeps, then protected count +
+   protected prefixes (self-delimiting; the leading STATUS_MANIFEST code has been
+   consumed).  Returns an owned DeleteManifest, or NULL after signalling
+   STATUS_ERROR on a malformed frame. */
+DeleteManifest* receive_manifest_entries(int fd);
 /* Remove destination entries under config->receive_root_directory that are not
-   in `manifest` (bounded walk, staging-dir skip).  The caller decides WHEN to
+   in `manifest` (bounded, all-or-nothing walk; staging-dir, basis-dir and
-   run it based on the negotiated delete timing. */
+   protected-prefix skips).  `--max-delete` and `--force` are honored here.  The
-bool manifest_delete_extras(const Config* config, ArrayList* manifest);
+   caller decides WHEN to run it based on the negotiated delete timing.  Returns
+   false (and the transfer fails) when the deletion cannot be committed. */
+bool manifest_delete_extras(const Config* config, DeleteManifest* manifest);
 
 /* 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/multiprocessing.c

@@ -26,6 +26,8 @@ PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* que
   context->scanner_done = false;
   context->loader_done = false;
   context->manifest = NULL;
+  context->excluded_paths = NULL;
+  context->scan_had_io_error = false;
   context->remove_source_files = NULL;
   context->early_delete = false;
   context->total_files = 0;
@@ -82,6 +84,8 @@ void pipeline_context_sender_destroy(PipelineContextSender* context) {
   if (context->manifest) {
     array_list_delete(context->manifest);
   }
+  if (context->excluded_paths)
+    array_list_delete(context->excluded_paths);
   if (context->remove_source_files)
     array_list_delete(context->remove_source_files);
   config_delete(context->config);
@@ -144,7 +148,7 @@ fail:
 void pipeline_context_receiver_destroy(PipelineContextReceiver* context) {
   config_delete(context->config);
   if (context->deferred_manifest)
-    array_list_delete(context->deferred_manifest);
+    delete_manifest_free(context->deferred_manifest);
   queue_destroy(context->queue);
   receiver_outcomes_destroy(&context->outcomes);
   mtx_destroy(&context->mutex);

src/shared/multiprocessing.h

@@ -25,6 +25,18 @@ typedef struct {
   cnd_t condition_not_empty_loader;
   bool loader_done;
   ArrayList* manifest;
+  /* Protected prefixes (paths the source scan excluded by user rules) sent
+     with the keep-set manifest so --delete leaves them alone unless
+     --delete-excluded is set.  NULL when not collecting.  Populated by the
+     scanner thread (parallel workers append under mutex_scanner via the
+     scanner's exclusion sink) or, in the early modes, by the path-only pre-scan
+     on the calling thread before the pipeline starts. */
+  ArrayList* excluded_paths;
+  /* A source I/O error (unreadable directory) was recorded during the scan.
+     Set by the pre-scan (before the threads start) or by the scanner thread
+     under mutex_scanner; the caller turns it into a non-zero exit when
+     --ignore-errors kept the run going. */
+  bool scan_had_io_error;
   ArrayList* remove_source_files;
   /* True when --delete-before/--delete-during require the keep-set manifest to
      be transmitted before any file data: context->manifest is then prebuilt by
@@ -67,7 +79,7 @@ typedef struct PipelineContextReceiver {
      deletion after both threads have joined, so no extra is removed unless the
       transfer truly succeeded.  NULL in the early delete modes (which delete at
       the manifest). */
-  ArrayList* deferred_manifest;
+  DeleteManifest* deferred_manifest;
 } PipelineContextReceiver;
 
 PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,

src/shared/utils.c

@@ -221,10 +221,117 @@ static bool path_under_skip_prefix(const char* child_rel, bool at_root,
   return false;
 }
 
+/* All-or-nothing max-delete needs to know BEFORE any unlink whether the run
+   would delete more than max_delete entries.  This rehearsal pass walks the
+   destination with the same decisions as the delete pass but never touches the
+   filesystem: it counts every regular file the delete pass would unlink and
+   every directory it would rmdir (a directory is removed only once every entry
+   below it has been removed and nothing the walker leaves in place survives).
+   Entries the walker never removes (symlinks, manifest-listed files, protected
+   prefixes) mark the enclosing directory as surviving, exactly as they would
+   make a real rmdir fail with ENOTEMPTY.  Stops early once *count reaches the
+   cap (sets *exceeds).  Returns false on a traversal error. */
+static bool count_extras_fd(int dirfd, const char* rel_path, ArrayList* manifest, size_t cap,
+                            size_t* count, bool* exceeds, const DeleteSkipEntry* skips,
+                            int skip_count, bool* survives) {
+  /* openat(dirfd, ".") opens an independent file description: a dup() would
+     share dirfd's file offset, and a prior rehearsal pass must not have drained
+     this directory's stream before the delete pass reads it again. */
+  int scanfd = openat(dirfd, ".", O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+  if (scanfd < 0)
+    return false;
+  DIR* dir = fdopendir(scanfd);
+  if (!dir) {
+    close(scanfd);
+    return false;
+  }
+  bool operation_ok = true;
+  bool local_survives = false;
+  bool at_root = rel_path[0] == '\0';
+  const struct dirent* entry;
+  while ((entry = readdir(dir)) != NULL) {
+    if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
+      continue;
+    if (*exceeds)
+      break;
+    char* child_rel = path_cat((char*)rel_path, entry->d_name);
+    if (!child_rel) {
+      operation_ok = false;
+      continue;
+    }
+    if (path_under_skip_prefix(child_rel, at_root, skips, skip_count)) {
+      local_survives = true;
+      free(child_rel);
+      continue;
+    }
+    struct stat st;
+    if (fstatat(dirfd, entry->d_name, &st, AT_SYMLINK_NOFOLLOW) != 0) {
+      if (errno != ENOENT)
+        operation_ok = false;
+      free(child_rel);
+      continue;
+    }
+    if (S_ISLNK(st.st_mode)) {
+      local_survives = true;
+      free(child_rel);
+      continue;
+    }
+    if (S_ISDIR(st.st_mode)) {
+      int childfd = openat(dirfd, entry->d_name, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+      bool child_ok = true;
+      bool child_survives = true;
+      if (childfd >= 0) {
+        child_ok = count_extras_fd(childfd, child_rel, manifest, cap, count, exceeds, skips,
+                                   skip_count, &child_survives);
+        close(childfd);
+      } else if (errno != ENOENT) {
+        operation_ok = false;
+      }
+      if (!child_ok)
+        operation_ok = false;
+      if (is_dir_in_manifest(child_rel, manifest)) {
+        /* A directory with kept content below it is never removed. */
+        local_survives = true;
+      } else if (child_survives) {
+        /* The directory still holds entries the walker leaves in place, so an
+           rmdir would fail with ENOTEMPTY; the delete pass leaves it behind
+           rather than reporting an error (matching rsync). */
+        local_survives = true;
+      } else {
+        if (*count >= cap) {
+          *exceeds = true;
+        } else {
+          (*count)++;
+        }
+      }
+    } else {
+      bool found = false;
+      for (int i = 0; i < manifest->size; i++) {
+        if (strcmp((char*)manifest->items[i], child_rel) == 0) {
+          found = true;
+          break;
+        }
+      }
+      if (!found) {
+        if (*count >= cap) {
+          *exceeds = true;
+        } else {
+          (*count)++;
+        }
+      }
+    }
+    free(child_rel);
+  }
+  closedir(dir);
+  *survives = local_survives;
+  return operation_ok;
+}
+
 static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifest,
                              size_t max_delete, size_t* deleted_count, const DeleteSkipEntry* skips,
                              int skip_count) {
-  int scanfd = dup(dirfd);
+  /* Independent file description (see count_extras_fd). */
+  int scanfd = openat(dirfd, ".", O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
   if (scanfd < 0)
     return false;
   DIR* dir = fdopendir(scanfd);
@@ -282,7 +389,12 @@ static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes
           operation_ok = false;
         } else {
           if (unlinkat(dirfd, entry->d_name, AT_REMOVEDIR) != 0) {
-            if (errno != ENOENT)
+            /* ENOENT: already gone (fine).  ENOTEMPTY/EEXIST: the directory
+               still holds entries the walker leaves in place (a protected
+               excluded prefix, a kept file the manifest protects, a symlink);
+               rsync leaves such a directory behind, so this is not an error.
+               Only genuine I/O failures abort the deletion. */
+            if (errno != ENOENT && errno != ENOTEMPTY && errno != EEXIST)
               operation_ok = false;
           } else {
             (*deleted_count)++;
@@ -321,10 +433,13 @@ static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes
   return operation_ok;
 }
 
-bool delete_extras_limited(const char* dest_root, ArrayList* manifest, size_t max_delete,
+DeleteWalkResult delete_extras_limited(const char* dest_root, ArrayList* manifest,
-                           const DeleteSkipEntry* skips, int skip_count) {
+                                       size_t max_delete, const DeleteSkipEntry* skips,
+                                       int skip_count, size_t* deleted_out) {
+  if (deleted_out)
+    *deleted_out = 0;
   if (!manifest)
-    return false;
+    return DELETE_WALK_ERROR;
   int rootfd;
   if (authorized_root_fd >= 0) {
     if (authorized_root_path)
@@ -337,16 +452,35 @@ bool delete_extras_limited(const char* dest_root, ArrayList* manifest, size_t ma
     rootfd = open(dest_root, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
   }
   if (rootfd < 0)
-    return false;
+    return DELETE_WALK_ERROR;
+  if (max_delete != SIZE_MAX) {
+    /* Rehearse the deletion first so a run that would exceed the cap removes
+       nothing (rsync's all-or-nothing --max-delete contract). */
+    size_t count = 0;
+    bool exceeds = false;
+    bool survives = false;
+    bool counted_ok = count_extras_fd(rootfd, "", manifest, max_delete, &count, &exceeds, skips,
+                                      skip_count, &survives);
+    if (!counted_ok) {
+      close(rootfd);
+      return DELETE_WALK_ERROR;
+    }
+    if (exceeds) {
+      close(rootfd);
+      return DELETE_WALK_LIMIT_EXCEEDED;
+    }
+  }
   size_t deleted_count = 0;
   bool ok = delete_extras_fd(rootfd, "", manifest, max_delete, &deleted_count, skips, skip_count);
   if (close(rootfd) != 0)
     ok = false;
-  return ok;
+  if (deleted_out)
+    *deleted_out = deleted_count;
+  return ok ? DELETE_WALK_OK : DELETE_WALK_ERROR;
 }
 
 bool delete_extras(const char* dest_root, ArrayList* manifest) {
-  return delete_extras_limited(dest_root, manifest, SIZE_MAX, NULL, 0);
+  return delete_extras_limited(dest_root, manifest, SIZE_MAX, NULL, 0, NULL) == DELETE_WALK_OK;
 }
 
 bool has_path_traversal(const char* path) {

src/shared/utils.h

@@ -9,22 +9,43 @@ char* str_dup(const char* string);
 char* output_escape(const char* string, bool eight_bit_output);
 char* path_cat(const char* path1, const char* path2);
 bool glob_match(const char* pattern, const char* str);
-bool delete_extras(const char* dest_root, ArrayList* manifest);
+/* Result of a bounded extra-file deletion run. */
+typedef enum {
+  /* Every extra entry was removed (or there were none). */
+  DELETE_WALK_OK = 0,
+  /* The destination holds more extras than the numeric cap for this run.  With
+     the all-or-nothing max-delete semantics NOTHING was removed (the walker
+     counts first and refuses to start when the run would exceed the limit). */
+  DELETE_WALK_LIMIT_EXCEEDED,
+  /* A traversal or unlink failure aborted the deletion (partial removal is
+     possible, mirroring the delete pass). */
+  DELETE_WALK_ERROR
+} DeleteWalkResult;
 /* One protected entry for the delete walker.  When top_level_only is true the
    prefix is skipped only as a DIRECT child of dest_root (the --delay-updates
    staging directory, which must not hide genuine extras inside a nested
    destination directory that happens to share the staging name); otherwise the
    prefix is skipped at any depth (the --compare-dest/--copy-dest/--link-dest
-   basis trees, which the transfer links from and are never destination
+   basis trees, and the sender-side protected filter-excluded prefixes, which
-   content). */
+   are never destination content). */
 typedef struct {
   const char* prefix;
   bool top_level_only;
 } DeleteSkipEntry;
 /* Remove files/dirs under dest_root that are not listed in manifest without
-   ever descending into a protected prefix (see DeleteSkipEntry). */
+   ever descending into a protected prefix (see DeleteSkipEntry).  When
-bool delete_extras_limited(const char* dest_root, ArrayList* manifest, size_t max_delete,
+   max_delete is not SIZE_MAX the run is all-or-nothing: extras are counted
-                           const DeleteSkipEntry* skips, int skip_count);
+   first and DELETE_WALK_LIMIT_EXCEEDED is returned (with nothing removed) when
+   the count would exceed the cap.  `deleted_out` optionally receives the number
+   of entries actually removed.  The all-or-nothing guarantee holds only while
+   the destination tree is not being concurrently modified: the rehearsal pass
+   and the delete pass are two separate walks, so a concurrent change between
+   them (another process adding/removing entries) can make the second pass
+   delete a different set than the first one counted. */
+DeleteWalkResult delete_extras_limited(const char* dest_root, ArrayList* manifest,
+                                       size_t max_delete, const DeleteSkipEntry* skips,
+                                       int skip_count, size_t* deleted_out);
+bool delete_extras(const char* dest_root, ArrayList* manifest);
 bool utils_set_authorized_root(int fd, const char* canonical_path);
 /* The fd-only compatibility form is fail-closed for path-based operations;
  * callers should use utils_set_authorized_root with the canonical identity. */

tests/integration/test_features.py

@@ -2107,6 +2107,437 @@ class TestDeleteTiming:
         assert os.path.exists(extra), "unauthorized delete removed an extra file"
 
 
+def _seed_delete_tree(tag, entries, dest):
+    """Create a source tree and seed a full mirror at `dest`, returning
+    (source, received_mirror)."""
+    source = os.path.join(TEST_DATA_DIR, f"delpol_{tag}_src")
+    clean_dir(source)
+    for rel, content in entries.items():
+        full = os.path.join(source, rel)
+        os.makedirs(os.path.dirname(full), exist_ok=True)
+        with open(full, "wb") as fh:
+            fh.write(content)
+    clean_dir(dest)
+    with ServerManager() as server:
+        server.start(extra_args=["--allow-delete"])
+        result, _ = run_client(source, dest, port=server.port)
+        assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+    received = get_dest_received_dir(dest, source)
+    return source, received
+
+
+class TestDeletePolicy:
+    """Deletion-policy family: --delete-excluded, --max-delete, --force,
+    --ignore-errors and --prune-empty-dirs."""
+
+    def _write(self, path, content):
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(content)
+
+    @pytest.mark.parametrize("mt", [False, True])
+    @pytest.mark.parametrize("timing",
+                             ["--delete", "--delete-before", "--delete-after", "--delete-delay"])
+    def test_delete_protects_excluded_by_default_and_delete_excluded_removes(self, mt, timing):
+        """rsync parity: with a --delete timing the destination mirror path whose
+        source was excluded survives (protected by default); --delete-excluded
+        opts back into deleting it.  Verified single-threaded and -m across every
+        timing (commit and early)."""
+        source = os.path.join(TEST_DATA_DIR, f"delexcl_{timing.strip('-')}_{mt}_src")
+        clean_dir(source)
+        entries = {
+            "keep.txt": b"kept\n",
+            "secret.log": b"secret\n",
+            "sub/nested.log": b"nested secret\n",
+        }
+        for rel, content in entries.items():
+            self._write(os.path.join(source, rel), content)
+        dest = os.path.join(TEST_DATA_DIR, f"delexcl_{timing.strip('-')}_{mt}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            self._write(os.path.join(received, "extra.txt"), b"extra\n")
+
+            # Default: the excluded mirrors survive --delete, genuine extras die.
+            flags = ["--exclude", "*.log", timing] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"default delete sync failed: {(result.stderr or result.stdout)[:300]}"
+            assert os.path.exists(os.path.join(received, "secret.log")), \
+                "excluded dest file was deleted under plain --delete (rsync protects it)"
+            assert os.path.exists(os.path.join(received, "sub", "nested.log")), \
+                "nested excluded dest file was deleted under plain --delete"
+            assert not os.path.exists(os.path.join(received, "extra.txt")), \
+                "genuine extra was not deleted"
+
+            # --delete-excluded: excluded mirrors are extras again and die.
+            self._write(os.path.join(received, "extra.txt"), b"extra\n")
+            flags = ["--exclude", "*.log", timing, "--delete-excluded"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"--delete-excluded sync failed: {(result.stderr or result.stdout)[:300]}"
+            assert not os.path.exists(os.path.join(received, "secret.log")), \
+                "--delete-excluded did not remove the excluded dest file"
+            assert not os.path.exists(os.path.join(received, "sub", "nested.log")), \
+                "--delete-excluded did not remove the nested excluded dest file"
+            assert not os.path.exists(os.path.join(received, "extra.txt")), \
+                "genuine extra survived --delete-excluded"
+            assert _read_file(os.path.join(received, "keep.txt")) == b"kept\n"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_delete_excluded_excluded_directory_subtree(self, mt):
+        """A whole source directory excluded by a filter rule protects its whole
+        destination mirror by default; --delete-excluded removes the subtree."""
+        source = os.path.join(TEST_DATA_DIR, f"delexcldir_{mt}_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "keep.txt"), b"kept\n")
+        self._write(os.path.join(source, "skipdir", "a.log"), b"a\n")
+        self._write(os.path.join(source, "skipdir", "deep", "b.log"), b"b\n")
+        dest = os.path.join(TEST_DATA_DIR, f"delexcldir_{mt}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+
+            flags = ["--filter=- skipdir/", "--delete"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"default delete sync failed: {(result.stderr or result.stdout)[:300]}"
+            assert os.path.exists(os.path.join(received, "skipdir", "a.log")), \
+                "excluded dir subtree was deleted under plain --delete"
+            assert os.path.exists(os.path.join(received, "skipdir", "deep", "b.log")), \
+                "nested excluded dir content was deleted under plain --delete"
+
+            flags = ["--filter=- skipdir/", "--delete", "--delete-excluded"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"--delete-excluded sync failed: {(result.stderr or result.stdout)[:300]}"
+            assert not os.path.exists(os.path.join(received, "skipdir")), \
+                "--delete-excluded did not remove the excluded dir subtree"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    @pytest.mark.parametrize("timing", ["--delete", "--delete-before"])
+    def test_max_delete_exceeded_fails_without_deleting(self, mt, timing):
+        """A run that would exceed --max-delete deletes nothing and fails."""
+        source = os.path.join(TEST_DATA_DIR, f"maxdel_{timing.strip('-')}_{mt}_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "keep.txt"), b"kept\n")
+        dest = os.path.join(TEST_DATA_DIR, f"maxdel_{timing.strip('-')}_{mt}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            extras = []
+            for i in range(4):
+                name = f"e{i}.txt"
+                self._write(os.path.join(received, name), b"extra\n")
+                extras.append(os.path.join(received, name))
+
+            flags = ["--max-delete=2", timing] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode != 0, \
+                f"--max-delete=2 with 4 extras unexpectedly succeeded: {result.stderr[:300]}"
+            for path in extras:
+                assert os.path.exists(path), \
+                    "--max-delete overrun deleted files (must be all-or-nothing)"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_max_delete_not_exceeded_deletes_exactly(self, mt):
+        """When the extras are at or below --max-delete the run succeeds and
+        removes exactly the extras."""
+        source = os.path.join(TEST_DATA_DIR, f"maxdelok_{mt}_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "keep.txt"), b"kept\n")
+        dest = os.path.join(TEST_DATA_DIR, f"maxdelok_{mt}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            for i in range(3):
+                self._write(os.path.join(received, f"e{i}.txt"), b"extra\n")
+            flags = ["--max-delete=3", "--delete"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"--max-delete=3 with 3 extras failed: {(result.stderr or result.stdout)[:300]}"
+            for i in range(3):
+                assert not os.path.exists(os.path.join(received, f"e{i}.txt")), \
+                    f"extra e{i}.txt not deleted under --max-delete=3"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_force_replaces_nonempty_dir_with_file(self, mt):
+        """--force lets an incoming regular file replace a non-empty destination
+        directory; without it the write (and the run) fails."""
+        source = os.path.join(TEST_DATA_DIR, f"force_{mt}_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "sub", "old.txt"), b"old\n")
+        self._write(os.path.join(source, "keep.txt"), b"kept\n")
+        dest = os.path.join(TEST_DATA_DIR, f"force_{mt}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+
+            # The source path `sub` becomes a regular file (the dir is gone).
+            os.unlink(os.path.join(source, "sub", "old.txt"))
+            os.rmdir(os.path.join(source, "sub"))
+            self._write(os.path.join(source, "sub"), b"now a file\n")
+
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode != 0, \
+                "a file over a non-empty directory must fail without --force"
+            assert os.path.isdir(os.path.join(received, "sub")), \
+                "directory was destroyed although the run failed without --force"
+            assert os.path.exists(os.path.join(received, "sub", "old.txt")), \
+                "non-empty dir content was lost although the run failed without --force"
+
+            flags = ["--force"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"--force run failed: {(result.stderr or result.stdout)[:300]}"
+            assert os.path.isfile(os.path.join(received, "sub")), \
+                "--force did not replace the directory with the file"
+            assert _read_file(os.path.join(received, "sub")) == b"now a file\n"
+
+    def test_force_inert_under_delay_updates(self):
+        """Documented divergence: --force acts on the immediate-install path; a
+        --delay-updates run stages into its own tree and its publication renames
+        over regular files only, so a blocking directory is not cleared and the
+        run fails."""
+        source = os.path.join(TEST_DATA_DIR, "force_delay_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "sub", "old.txt"), b"old\n")
+        self._write(os.path.join(source, "keep.txt"), b"kept\n")
+        dest = os.path.join(TEST_DATA_DIR, "force_delay_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            os.unlink(os.path.join(source, "sub", "old.txt"))
+            os.rmdir(os.path.join(source, "sub"))
+            self._write(os.path.join(source, "sub"), b"now a file\n")
+            result, _ = run_client(source, dest, flags=["--force", "--delay-updates"],
+                                   port=server.port)
+            assert result.returncode != 0, \
+                "--force --delay-updates unexpectedly replaced the blocking directory"
+            assert os.path.isdir(os.path.join(received, "sub")), \
+                "blocking directory was cleared although --delay-updates should keep --force inert"
+            assert os.path.exists(os.path.join(received, "sub", "old.txt")), \
+                "blocking directory content was lost"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_prune_empty_dirs_dirs_mode(self, mt):
+        """--prune-empty-dirs omits an empty source directory's explicit entry in
+        --dirs mode (nothing is created, and an existing empty mirror is removed
+        by --delete).  Recursive transfers never emit empty dirs, so the flag is
+        a no-op there (documented rsync -m parity)."""
+        source = os.path.join(TEST_DATA_DIR, f"prune_{mt}_src")
+        clean_dir(source)
+        os.makedirs(source, exist_ok=True)  # physically empty source dir
+
+        dest = os.path.join(TEST_DATA_DIR, f"prune_{mt}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, flags=["--dirs"], port=server.port)
+            assert result.returncode == 0, f"-d seed failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            assert os.path.isdir(received), "-d should create the empty mirror dir"
+            assert os.listdir(received) == []
+
+            # prune-empty-dirs: the empty mirror is pruned by --delete.
+            flags = ["--dirs", "--prune-empty-dirs", "--delete"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"--dirs --prune-empty-dirs --delete failed: {(result.stderr or result.stdout)[:300]}"
+            assert not os.path.exists(received), \
+                "--prune-empty-dirs did not prune the empty dir (--delete left it)"
+
+        # A fresh destination: prune-empty-dirs means the empty dir is never sent.
+        dest2 = os.path.join(TEST_DATA_DIR, f"prune2_{mt}_dst")
+        clean_dir(dest2)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            flags = ["--dirs", "--prune-empty-dirs", "-i"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest2, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"--dirs --prune-empty-dirs failed: {(result.stderr or result.stdout)[:300]}"
+            received2 = get_dest_received_dir(dest2, source)
+            assert not os.path.exists(received2), \
+                "--prune-empty-dirs transferred the empty directory"
+            assert result.stdout == "", \
+                f"--prune-empty-dirs leaked an itemize line: {result.stdout[:200]}"
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_prune_empty_dirs_recursion_inherent(self, mt):
+        """In recursive mode FastSync never transfers empty directories (rsync
+        -m parity): a truly-empty destination directory chain is removed by
+        --delete whether or not --prune-empty-dirs is given (the flag has no
+        additional effect there), while directories holding kept files survive.
+        A filter-excluded file's mirror is protected, so a directory that still
+        holds one is left intact (rsync default delete-excluded semantics)."""
+        source = os.path.join(TEST_DATA_DIR, f"prunerec_{mt}_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "keep.txt"), b"kept\n")
+        self._write(os.path.join(source, "a", "keep.log"), b"a log\n")
+        self._write(os.path.join(source, "b", "deep", "kept.txt"), b"deep kept\n")
+        dest = os.path.join(TEST_DATA_DIR, f"prunerec_{mt}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            # A stray empty chain (FastSync recursion never creates such dirs, so
+            # this models one left by an external tool / an earlier --dirs run).
+            os.makedirs(os.path.join(received, "empty", "chain"))
+
+            for prune in ([], ["--prune-empty-dirs"]):
+                flags = prune + ["--delete"] + (["-m"] if mt else [])
+                result, _ = run_client(source, dest, flags=flags, port=server.port)
+                assert result.returncode == 0, \
+                    f"prune recursive sync failed: {(result.stderr or result.stdout)[:300]}"
+                assert not os.path.exists(os.path.join(received, "empty")), \
+                    "truly-empty dir chain was not removed by --delete"
+                assert os.path.exists(os.path.join(received, "b", "deep", "kept.txt")), \
+                    "non-empty dir subtree was wrongly removed"
+                assert _read_file(os.path.join(received, "keep.txt")) == b"kept\n"
+
+            # An excluded file's mirror is protected: the dir that holds it stays.
+            flags = ["--exclude", "*.log", "--delete", "--prune-empty-dirs"] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"prune recursive sync failed: {(result.stderr or result.stdout)[:300]}"
+            assert os.path.exists(os.path.join(received, "a", "keep.log")), \
+                "excluded file mirror was deleted under --delete (rsync protects it)"
+
+    def _run_client_as_nobody(self, source, dest, port, flags):
+        cmd = CLIENT_CMD + ["--source-dir", source, "--dest-dir", dest,
+                            "--save-to-disk", "--server-port", str(port)] + flags
+        return subprocess.run(["setpriv", "--reuid=65534", "--regid=65534",
+                               "--clear-groups"] + cmd, text=True, capture_output=True)
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_ignore_errors_keeps_deletion_active_on_scan_error(self, mt):
+        """A source I/O error (unreadable subdirectory) aborts the run so no
+        deletion happens by default; --ignore-errors continues, still transfers
+        the readable tree and still deletes, single-threaded and under -m.  Run
+        as an unprivileged user so the mode-000 directory is genuinely
+        unreadable."""
+        if os.geteuid() != 0 or shutil.which("setpriv") is None:
+            pytest.skip("requires root + setpriv to drop privileges for the client")
+        tag = f"ioerr_{os.getpid()}_{mt}"
+        source = os.path.join(TEST_DATA_DIR, f"{tag}_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "top.txt"), b"top\n")
+        self._write(os.path.join(source, "ok", "inside.txt"), b"inside\n")
+        self._write(os.path.join(source, "locked", "blocked.txt"), b"blocked\n")
+        dest = os.path.join(TEST_DATA_DIR, f"{tag}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            # Seed as root (server is root too).
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            try:
+                os.chmod(os.path.join(source, "locked"), 0)
+
+                # Default: scan error aborts the run; nothing is deleted.
+                self._write(os.path.join(received, "extra.txt"), b"extra\n")
+                flags = ["--delete"] + (["-m"] if mt else [])
+                result = self._run_client_as_nobody(source, dest, server.port, flags)
+                assert result.returncode != 0, "unreadable source dir did not fail the run"
+                assert os.path.exists(os.path.join(received, "extra.txt")), \
+                    "default run deleted although the scan hit an I/O error"
+
+                # --ignore-errors: the readable tree transfers, deletion still runs.
+                self._write(os.path.join(received, "extra.txt"), b"extra\n")
+                flags = ["--delete", "--ignore-errors"] + (["-m"] if mt else [])
+                result = self._run_client_as_nobody(source, dest, server.port, flags)
+                assert not os.path.exists(os.path.join(received, "extra.txt")), \
+                    f"--ignore-errors did not keep deletion active: {result.stderr[:300]}"
+                assert not os.path.exists(os.path.join(received, "locked")), \
+                    "mirror of the unreadable dir was left behind (should be an extra)"
+            finally:
+                os.chmod(os.path.join(source, "locked"), 0o755)
+
+    @pytest.mark.parametrize("mt", [False, True])
+    @pytest.mark.parametrize("timing", ["--delete", "--delete-before"])
+    def test_ignore_errors_unreadable_root_never_deletes(self, mt, timing):
+        """An unreadable SOURCE ROOT must never be treated as a skippable scan
+        error: with --ignore-errors the sequential scanner treats the root as
+        fatal (matching the -m path, which cannot even create its scanner), so
+        no empty keep-set manifest is sent and the destination is never wiped.
+        Run as an unprivileged user so the mode-000 root is genuinely
+        unreadable."""
+        if os.geteuid() != 0 or shutil.which("setpriv") is None:
+            pytest.skip("requires root + setpriv to drop privileges for the client")
+        tag = f"rootio_{os.getpid()}_{mt}_{timing.strip('-')}"
+        source = os.path.join(TEST_DATA_DIR, f"{tag}_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "file.txt"), b"content\n")
+        dest = os.path.join(TEST_DATA_DIR, f"{tag}_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            try:
+                os.chmod(source, 0)
+                self._write(os.path.join(received, "extra.txt"), b"extra\n")
+                flags = [timing, "--ignore-errors"] + (["-m"] if mt else [])
+                result = self._run_client_as_nobody(source, dest, server.port, flags)
+                assert result.returncode != 0, \
+                    f"unreadable source root with {timing} (mt={mt}) unexpectedly succeeded"
+                assert os.path.exists(os.path.join(received, "file.txt")), \
+                    f"{timing} (mt={mt}) wiped a kept destination file"
+                assert os.path.exists(os.path.join(received, "extra.txt")), \
+                    f"{timing} (mt={mt}) deleted the extra although the scan could not read the root"
+            finally:
+                os.chmod(source, 0o755)
+
+    def test_delete_excluded_protection_is_sender_derived(self):
+        """Plain --delete protects destination mirrors of files the SOURCE scan
+        excluded, but a destination-only file that merely matches an exclude
+        rule is still an extra and is removed (protection never re-applies rules
+        to the destination)."""
+        source = os.path.join(TEST_DATA_DIR, "senderderived_src")
+        clean_dir(source)
+        self._write(os.path.join(source, "keep.txt"), b"kept\n")
+        self._write(os.path.join(source, "secret.log"), b"secret\n")
+        dest = os.path.join(TEST_DATA_DIR, "senderderived_dst")
+        clean_dir(dest)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, port=server.port)
+            assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+            received = get_dest_received_dir(dest, source)
+            # A destination-only file that happens to match the exclude rule.
+            self._write(os.path.join(received, "stray.log"), b"never on the source\n")
+            result, _ = run_client(source, dest, flags=["--exclude", "*.log", "--delete"],
+                                   port=server.port)
+            assert result.returncode == 0, \
+                f"delete sync failed: {(result.stderr or result.stdout)[:300]}"
+            assert os.path.exists(os.path.join(received, "secret.log")), \
+                "source-excluded mirror was deleted under plain --delete"
+            assert not os.path.exists(os.path.join(received, "stray.log")), \
+                "destination-only file matching the exclude rule was left (should be deleted)"
+
+
 def _pin_mtime(path, ts):
     os.utime(path, (ts, ts))
 

tests/test_client_cli.c

@@ -1648,6 +1648,73 @@ static void test_parse_args_files_from() {
   remove(list_path);
 }
 
+/* The deletion-policy family parses onto the config fields: --delete-excluded,
+ * --ignore-errors and --force are flags, --max-delete takes a non-negative
+ * number, and --prune-empty-dirs is the long-only spelling (FastSync's -m stays
+ * multithreading).  None of them implies --delete by itself. */
+static void test_parse_args_delete_policy_flags() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync",
+                  "--delete",
+                  "--delete-excluded",
+                  "--max-delete=5",
+                  "--ignore-errors",
+                  "--force",
+                  "--prune-empty-dirs",
+                  "/src",
+                  "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 9, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_delete);
+  EXPECT_TRUE(cfg->delete_excluded);
+  EXPECT_EQ_INT(cfg->max_delete, 5);
+  EXPECT_TRUE(cfg->ignore_errors);
+  EXPECT_TRUE(cfg->force_delete);
+  EXPECT_TRUE(cfg->prune_empty_dirs);
+  EXPECT_FALSE(cfg->delete_before);
+  config_delete(cfg);
+
+  /* --max-delete accepts the separated-argument and zero forms. */
+  cfg = config_create();
+  char* argv2[] = {"fastsync", "--max-delete", "0", "/src", "/dst"};
+  positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv2, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->max_delete, 0);
+  config_delete(cfg);
+}
+
+static void test_parse_args_delete_policy_invalid_values() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--max-delete=abc", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), -1);
+  config_delete(cfg);
+
+  cfg = config_create();
+  char* argv2[] = {"fastsync", "--max-delete=-3", "/src", "/dst"};
+  positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv2, positional_args, &positional_count), -1);
+  config_delete(cfg);
+}
+
+/* --max-delete without --delete is inert (it only bounds a --delete run); the
+ * config stays valid. */
+static void test_parse_args_max_delete_inert_without_delete() {
+  Config* cfg = config_create();
+  cfg->send_directory = str_dup("/src");
+  cfg->receive_root_directory = str_dup("/dst");
+  char* argv[] = {"fastsync", "--max-delete=5", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_FALSE(cfg->use_delete);
+  EXPECT_EQ_INT(cfg->max_delete, 5);
+  EXPECT_TRUE(validate_config(cfg));
+  config_delete(cfg);
+}
+
 void test_client_cli() {
   test_validate_config_required_paths();
   test_validate_config_incompatible_options();
@@ -1740,4 +1807,7 @@ void test_client_cli() {
   test_parse_args_basis_dirs();
   test_parse_args_basis_invalid_paths();
   test_validate_config_basis_rejects_chunk_serialization();
+  test_parse_args_delete_policy_flags();
+  test_parse_args_delete_policy_invalid_values();
+  test_parse_args_max_delete_inert_without_delete();
 }

tests/test_config.c

@@ -561,6 +561,61 @@ static void test_config_delete_timing_conflict_rejected() {
   config_delete(c);
 }
 
+/* The deletion-policy fields that cross the wire survive a config round trip:
+   --force (force_delete), --delete-excluded, --prune-empty-dirs and the
+   --max-delete number (default -1 == no client limit). */
+static void test_config_delete_policy_wire_roundtrip() {
+  if (is_running_under_valgrind())
+    return;
+
+  struct {
+    bool force_delete, delete_excluded, prune_empty_dirs;
+    int max_delete;
+  } cases[] = {
+      {false, false, false, -1},
+      {true, false, false, 0},
+      {false, true, true, 7},
+  };
+  for (size_t i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {
+    int p[2];
+    EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+    pid_t pid = fork();
+    if (pid == 0) {
+      close(p[1]);
+      io_set_fds(p[0], p[0]);
+      Config* recv = config_receive(p[0]);
+      bool ok = recv != NULL;
+      if (ok) {
+        ok = recv->force_delete == cases[i].force_delete &&
+             recv->delete_excluded == cases[i].delete_excluded &&
+             recv->prune_empty_dirs == cases[i].prune_empty_dirs &&
+             recv->max_delete == cases[i].max_delete;
+      }
+      config_delete(recv);
+      close(p[0]);
+      _exit(ok ? 0 : 1);
+    } else {
+      close(p[0]);
+      io_set_fds(p[1], p[1]);
+      Config* send_cfg = config_create();
+      EXPECT_NOT_NULL(send_cfg);
+      send_cfg->send_directory = str_dup("/src");
+      send_cfg->receive_root_directory = str_dup("/dst");
+      send_cfg->force_delete = cases[i].force_delete;
+      send_cfg->delete_excluded = cases[i].delete_excluded;
+      send_cfg->prune_empty_dirs = cases[i].prune_empty_dirs;
+      send_cfg->max_delete = cases[i].max_delete;
+      bool sent = config_send(p[1], send_cfg);
+      int status;
+      waitpid(pid, &status, 0);
+      close(p[1]);
+      config_delete(send_cfg);
+      EXPECT_TRUE(sent);
+      EXPECT_TRUE(WIFEXITED(status) && WEXITSTATUS(status) == 0);
+    }
+  }
+}
+
 /* Basis-dir lists survive the config wire: each entry's type and path must
    round-trip unchanged. */
 static void test_config_basis_roundtrip() {
@@ -712,6 +767,7 @@ void test_config() {
     test_config_delay_updates_reserved_backup_rejected();
     test_config_delete_timing_wire_roundtrip();
     test_config_delete_timing_conflict_rejected();
+    test_config_delete_policy_wire_roundtrip();
     test_config_basis_roundtrip();
     test_config_basis_wire_rejects_escaping();
     test_config_basis_normalization();

tests/test_scanner.c

@@ -397,9 +397,8 @@ static void test_parallel_scanner_root_chunks_without_workers() {
   create_test_file(file1, "a");
   create_test_file(file2, "b");
 
-  ScannerOptions options = {false, 1,     NULL, 0,     NULL,  0,     0,
+  ScannerOptions options = {0};
-                            0,     0,     0,    false, false, false, false,
+  options.chunk_size = 1;
-                            false, false, NULL, NULL,  false, false, false};
   ParallelScanner* scanner = parallel_scanner_create_with_options(dir, &options, NULL);
   EXPECT_NOT_NULL(scanner);
 

tests/test_server.c

@@ -476,7 +476,7 @@ static Config* make_late_delete_config(const char* root) {
   return cfg;
 }
 
-static int run_pending_receiver(Config* cfg, int fd, ArrayList** pending) {
+static int run_pending_receiver(Config* cfg, int fd, DeleteManifest** pending) {
   ReceiverSink sink = {0};
   return receiver_process_pending(cfg, fd, &sink, pending);
 }
@@ -492,9 +492,10 @@ static void test_late_manifest_abort_frees_keepset() {
   EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
   EXPECT_TRUE(send_int(p[1], 1));
   EXPECT_TRUE(send_str(p[1], "keep.txt"));
+  EXPECT_TRUE(send_int(p[1], 0)); /* protected-prefix section is empty */
   EXPECT_TRUE(send_status(p[1], STATUS_ABORT));
 
-  ArrayList* pending = NULL;
+  DeleteManifest* pending = NULL;
   EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
   EXPECT_NULL(pending);
 
@@ -514,9 +515,10 @@ static void test_late_manifest_eof_frees_keepset() {
   EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
   EXPECT_TRUE(send_int(p[1], 1));
   EXPECT_TRUE(send_str(p[1], "keep.txt"));
+  EXPECT_TRUE(send_int(p[1], 0)); /* protected-prefix section is empty */
   shutdown(p[1], SHUT_WR);
 
-  ArrayList* pending = NULL;
+  DeleteManifest* pending = NULL;
   EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
   EXPECT_NULL(pending);
 
@@ -536,11 +538,13 @@ static void test_late_second_manifest_frees_both() {
   EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
   EXPECT_TRUE(send_int(p[1], 1));
   EXPECT_TRUE(send_str(p[1], "first.txt"));
+  EXPECT_TRUE(send_int(p[1], 0)); /* protected-prefix section is empty */
   EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
   EXPECT_TRUE(send_int(p[1], 1));
   EXPECT_TRUE(send_str(p[1], "second.txt"));
+  EXPECT_TRUE(send_int(p[1], 0)); /* protected-prefix section is empty */
 
-  ArrayList* pending = NULL;
+  DeleteManifest* pending = NULL;
   EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
   EXPECT_NULL(pending);
 

tests/test_shared_utils.c

@@ -2,9 +2,255 @@
 #include "utils.h"
 #include "protocol.h"
 #include "test_utils.h"
+#include <dirent.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/stat.h>
 #include <threads.h>
+#include <unistd.h>
+
+/* ---- delete-walker tests ---- */
+
+static char* make_walk_root(const char* tag) {
+  char* path = malloc(256);
+  if (!path)
+    return NULL;
+  snprintf(path, 256, "/tmp/fastsync_walk_%s_%d", tag, (int)getpid());
+  rmdir(path);
+  if (mkdir(path, 0755) != 0) {
+    free(path);
+    return NULL;
+  }
+  return path;
+}
+
+static bool write_file_at(const char* dir, const char* name, const char* content) {
+  char* path = path_cat(dir, name);
+  if (!path)
+    return false;
+  int fd = open(path, O_WRONLY | O_CREAT | O_TRUNC, 0644);
+  bool ok = fd >= 0;
+  if (fd >= 0) {
+    if (content) {
+      const char* p = content;
+      size_t remaining = strlen(content);
+      while (remaining > 0) {
+        ssize_t n = write(fd, p, remaining);
+        if (n <= 0) {
+          ok = false;
+          break;
+        }
+        p += n;
+        remaining -= (size_t)n;
+      }
+    }
+    close(fd);
+  }
+  free(path);
+  return ok;
+}
+
+static bool file_exists(const char* dir, const char* name) {
+  char* path = path_cat(dir, name);
+  bool exists = path && access(path, F_OK) == 0;
+  free(path);
+  return exists;
+}
+
+static bool dir_exists(const char* dir, const char* name) {
+  char* path = path_cat(dir, name);
+  struct stat st;
+  bool exists = path && stat(path, &st) == 0 && S_ISDIR(st.st_mode);
+  free(path);
+  return exists;
+}
+
+static int make_subdir(const char* root, const char* name) {
+  char* path = path_cat(root, name);
+  int rc = -1;
+  if (path) {
+    rc = mkdir(path, 0755);
+    free(path);
+  }
+  return rc;
+}
+
+static void remove_walk_tree(const char* path) {
+  DIR* dir = opendir(path);
+  if (!dir) {
+    rmdir(path);
+    return;
+  }
+  const struct dirent* entry;
+  while ((entry = readdir(dir)) != NULL) {
+    if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
+      continue;
+    char* child = path_cat(path, entry->d_name);
+    if (child) {
+      struct stat st;
+      if (lstat(child, &st) == 0 && S_ISDIR(st.st_mode))
+        remove_walk_tree(child);
+      else
+        unlink(child);
+      free(child);
+    }
+  }
+  closedir(dir);
+  rmdir(path);
+}
+
+static ArrayList* make_manifest_strings(const char* const* entries, int count) {
+  ArrayList* manifest = array_list_create(free);
+  if (!manifest)
+    return NULL;
+  for (int i = 0; i < count; i++) {
+    char* dup = str_dup(entries[i]);
+    if (!dup || !array_list_add(manifest, dup)) {
+      free(dup);
+      array_list_delete(manifest);
+      return NULL;
+    }
+  }
+  return manifest;
+}
+
+static void test_walker_removes_extras_keeps_manifest_and_protected() {
+  char* root = make_walk_root("basic");
+  EXPECT_NOT_NULL(root);
+  EXPECT_TRUE(write_file_at(root, "a.txt", "extra"));
+  EXPECT_TRUE(write_file_at(root, "keep.txt", "kept"));
+  EXPECT_EQ_INT(make_subdir(root, "d"), 0);
+  EXPECT_TRUE(write_file_at(root, "d/e.txt", "extra"));
+  EXPECT_TRUE(write_file_at(root, "d/k.txt", "kept"));
+  EXPECT_EQ_INT(make_subdir(root, "prot"), 0);
+  EXPECT_TRUE(write_file_at(root, "prot/f.txt", "untouched"));
+
+  const char* keeps[] = {"keep.txt", "d/k.txt"};
+  ArrayList* manifest = make_manifest_strings(keeps, 2);
+  EXPECT_NOT_NULL(manifest);
+  DeleteSkipEntry skip = {"prot", false};
+  size_t deleted = 0;
+  DeleteWalkResult result = delete_extras_limited(root, manifest, 100000, &skip, 1, &deleted);
+  EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
+  EXPECT_FALSE(file_exists(root, "a.txt"));
+  EXPECT_TRUE(file_exists(root, "keep.txt"));
+  EXPECT_FALSE(file_exists(root, "d/e.txt"));
+  EXPECT_TRUE(file_exists(root, "d/k.txt"));
+  EXPECT_TRUE(dir_exists(root, "d"));
+  EXPECT_TRUE(file_exists(root, "prot/f.txt"));
+  EXPECT_TRUE(deleted >= 2);
+  array_list_delete(manifest);
+  remove_walk_tree(root);
+  free(root);
+}
+
+static void test_walker_max_delete_exceeded_deletes_nothing() {
+  char* root = make_walk_root("maxdel");
+  EXPECT_NOT_NULL(root);
+  EXPECT_TRUE(write_file_at(root, "a.txt", "extra"));
+  EXPECT_TRUE(write_file_at(root, "b.txt", "extra"));
+  EXPECT_TRUE(write_file_at(root, "c.txt", "extra"));
+  const char* keeps[1] = {NULL};
+  ArrayList* manifest = make_manifest_strings(keeps, 0);
+  EXPECT_NOT_NULL(manifest);
+  size_t deleted = 999;
+  DeleteWalkResult result = delete_extras_limited(root, manifest, 2, NULL, 0, &deleted);
+  EXPECT_EQ_INT((int)result, (int)DELETE_WALK_LIMIT_EXCEEDED);
+  EXPECT_EQ_INT((int)deleted, 0);
+  EXPECT_TRUE(file_exists(root, "a.txt"));
+  EXPECT_TRUE(file_exists(root, "b.txt"));
+  EXPECT_TRUE(file_exists(root, "c.txt"));
+  array_list_delete(manifest);
+  remove_walk_tree(root);
+  free(root);
+}
+
+static void test_walker_max_delete_exact_bound_deletes() {
+  char* root = make_walk_root("maxdel2");
+  EXPECT_NOT_NULL(root);
+  EXPECT_TRUE(write_file_at(root, "a.txt", "extra"));
+  EXPECT_TRUE(write_file_at(root, "b.txt", "extra"));
+  const char* keeps[1] = {NULL};
+  ArrayList* manifest = make_manifest_strings(keeps, 0);
+  EXPECT_NOT_NULL(manifest);
+  size_t deleted = 0;
+  DeleteWalkResult result = delete_extras_limited(root, manifest, 2, NULL, 0, &deleted);
+  EXPECT_EQ_INT((int)result, (int)DELETE_WALK_OK);
+  EXPECT_EQ_INT((int)deleted, 2);
+  EXPECT_FALSE(file_exists(root, "a.txt"));
+  EXPECT_FALSE(file_exists(root, "b.txt"));
+  array_list_delete(manifest);
+  remove_walk_tree(root);
+  free(root);
+}
+
+static void test_walker_unlimited_deletes_all() {
+  char* root = make_walk_root("unlim");
+  EXPECT_NOT_NULL(root);
+  EXPECT_TRUE(write_file_at(root, "a.txt", "extra"));
+  EXPECT_TRUE(write_file_at(root, "b.txt", "extra"));
+  EXPECT_EQ_INT(make_subdir(root, "emptydir"), 0);
+  const char* keeps[1] = {NULL};
+  ArrayList* manifest = make_manifest_strings(keeps, 0);
+  EXPECT_NOT_NULL(manifest);
+  EXPECT_TRUE(delete_extras(root, manifest));
+  EXPECT_FALSE(file_exists(root, "a.txt"));
+  EXPECT_FALSE(file_exists(root, "b.txt"));
+  EXPECT_FALSE(dir_exists(root, "emptydir"));
+  array_list_delete(manifest);
+  remove_walk_tree(root);
+  free(root);
+}
+
+/* The 100000-entry server hard bound (MAX_SERVER_DELETE_COUNT, which this test
+   exercises through a literal to avoid reaching into file_receive.c) is also
+   all-or-nothing: a destination holding more extras than the bound must be left
+   completely untouched.  Skipped under valgrind: 100k file creations would be
+   far too slow under instrumentation. */
+static void test_walker_hard_bound_all_or_nothing() {
+  if (is_running_under_valgrind())
+    return;
+  enum { HARD_BOUND = 100000 };
+  char* root = make_walk_root("hardbound");
+  EXPECT_NOT_NULL(root);
+  int rootfd = open(root, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
+  EXPECT_TRUE(rootfd >= 0);
+  bool created = true;
+  for (int i = 0; created && i < HARD_BOUND + 1; i++) {
+    char name[32];
+    snprintf(name, sizeof(name), "f%d", i);
+    int fd = openat(rootfd, name, O_WRONLY | O_CREAT | O_TRUNC, 0644);
+    if (fd < 0)
+      created = false;
+    else
+      close(fd);
+  }
+  EXPECT_TRUE(created);
+  const char* keeps[1] = {NULL};
+  ArrayList* manifest = make_manifest_strings(keeps, 0);
+  EXPECT_NOT_NULL(manifest);
+  size_t deleted = 999;
+  DeleteWalkResult result = delete_extras_limited(root, manifest, HARD_BOUND, NULL, 0, &deleted);
+  EXPECT_EQ_INT((int)result, (int)DELETE_WALK_LIMIT_EXCEEDED);
+  EXPECT_EQ_INT((int)deleted, 0);
+  EXPECT_TRUE(file_exists(root, "f0"));
+  EXPECT_TRUE(file_exists(root, "f100000"));
+  array_list_delete(manifest);
+  /* Fast cleanup: unlink every created name through the still-open root fd. */
+  if (rootfd >= 0) {
+    for (int i = 0; i < HARD_BOUND + 1; i++) {
+      char name[32];
+      snprintf(name, sizeof(name), "f%d", i);
+      (void)unlinkat(rootfd, name, 0);
+    }
+    close(rootfd);
+  }
+  rmdir(root);
+  free(root);
+}
 
 typedef struct {
   bool eight_bit_output;
@@ -24,6 +270,12 @@ static int escape_thread(void* arg) {
 }
 
 void test_shared_utils() {
+  test_walker_removes_extras_keeps_manifest_and_protected();
+  test_walker_max_delete_exceeded_deletes_nothing();
+  test_walker_max_delete_exact_bound_deletes();
+  test_walker_unlimited_deletes_all();
+  test_walker_hard_bound_all_or_nothing();
+
   char formatted[32];
   EXPECT_TRUE(format_human_bytes(0, formatted, sizeof(formatted)));
   EXPECT_EQ_STR(formatted, "0 B");