docs: recount RSYNC_COMPAT summary after Phase-3 wave A

The seed run did not preserve timestamps, so the destination copy's mtime was
the write time; the incremental --remove-source-files rerun only skipped the
file when both writes happened to land in the same whole second, making the
test flaky (observed intermittently in local full-suite runs and on CI).  Seed
with -M so the destination stores the source's exact mtime.

README.md

@@ -102,7 +102,11 @@ partial, alternate, and planned behavior.
 | `-q, --quiet` | Suppress non-error output |
 | `--progress` | Show real-time transfer speed |
 | `-P` | Enables partial-transfer mode and progress output (partial retention is incomplete) |
-| `--delete` | Delete files on receiver not present in source |
+| `--delete` | Delete files on receiver not present in source (default timing: delete-after, i.e. only after the whole transfer succeeded) |
+| `--delete-before` | Delete extras before the transfer starts (implies `--delete`) |
+| `--delete-during`, `--del` | Delete extras once the keep-set is known, before data is applied (implies `--delete`) |
+| `--delete-delay` | Delete extras only after a successful transfer (implies `--delete`) |
+| `--delete-after` | Explicit delete-after timing (implies `--delete`) |
 | `--exclude <pattern>` | Exclude files matching glob pattern (repeatable) |
 | `--exclude-from <file>` | Read exclude patterns from a file (one per line) |
 | `--include <pattern>` | Only transfer files matching glob pattern (repeatable, whitelist) |
@@ -386,7 +390,11 @@ option.
 |---|---|
 | `-a`, `--archive` | Enable current archive preset. Full rsync archive semantics are planned. |
 | `-n`, `--dry-run` | Scan and report without writing files. |
-| `--delete` | Request removal of destination entries absent from the source. The server must allow deletion. |
+| `--delete` | Request removal of destination entries absent from the source. The server must allow deletion. Default timing is delete-after: extras are removed only after the whole transfer succeeded. |
+| `--delete-before` | Delete extras before the transfer starts (implies `--delete`). |
+| `--delete-during`, `--del` | Delete extras once the keep-set manifest is known, before data is applied (implies `--delete`; early mode, same engine behaviour as `--delete-before`). |
+| `--delete-delay` | Delete extras only after a successful transfer (implies `--delete`; commit mode, same behaviour as `--delete-after`). |
+| `--delete-after` | Explicit delete-after timing: delete only after the transfer succeeded (implies `--delete`). |
 | `--exclude <pattern>` | Exclude matching paths. Repeatable. |
 | `--include <pattern>` | Include matching paths. Repeatable. |
 | `--exclude-from <file>` | Read exclude patterns from a file. |

RSYNC_COMPAT.md

@@ -6,11 +6,11 @@ This document maps rsync's full feature set to FastSync's current implementation
 
 | Status | Count | Description |
 |--------|-------|-------------|
-| ✅ Implemented | 67 | Feature works end-to-end |
+| ✅ Implemented | 74 | Feature works end-to-end |
 | 🔀 Alt Arg | 3 | Functionality exists but under different flag/semantics |
 | ⚠️ Partial | 5 | Flag parsed/stored but behavior incomplete |
 | 🔄 Compatibility No-op | 1 | Flag is accepted for CLI compatibility but has no effect |
-| ❌ Not Implemented | 71 | Flag not recognized or no behavior |
+| ❌ Not Implemented | 64 | Flag not recognized or no behavior |
 | **Total** | **147** | |
 
 ---
@@ -103,17 +103,56 @@ 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 |
-| `--delete-before` | Delete before transfer | ❌ Not Implemented | Removed because it had no effect |
-| `--del`, `--delete-during` | Delete during transfer | ❌ Not Implemented | Both flags are recognized but rejected; delete timing is not implemented |
-| `--delete-delay` | Find deletions during, delete after | ❌ Not Implemented | |
-| `--delete-after` | Delete after transfer | ❌ Not Implemented | Removed because it had no effect |
+| `--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-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 |
+| `--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 |
 | `--max-delete=NUM` | Max files to delete | ❌ Not Implemented | Removed because it had no effect |
 | `--ignore-errors` | Delete even with I/O errors | ❌ Not Implemented | |
 | `--force` | Force deletion of non-empty dirs | ❌ Not Implemented | |
 | `--prune-empty-dirs` | Prune empty dir chains | ❌ Not Implemented | Removed because it had no effect |
 
+**Deletion-timing implementation notes (Phase 3):** the delete flags above are
+real. Two new config booleans (`delete_during`, `delete_delay`) join the already
+serialized `delete_before`/`delete_after`, so the on-the-wire config layout
+changed and `PROTOCOL_VERSION` was bumped **2.7.0 → 2.8.0** (peers must match).
+The `STATUS_MANIFEST` frame is count-delimited and position-independent: the
+receiver commits the deletion either when the manifest arrives (early modes:
+`--delete-before`/`--delete-during`, which additionally acknowledge with
+`STATUS_OK` before data flows) or after the terminal `STATUS_FINISHED` proves
+the whole transfer succeeded (commit modes: plain `--delete`/`--delete-after`/
+`--delete-delay`). Timing is chosen purely from the config, so server policy
+(`--allow-delete` off) still disables deletion without deadlocking the early
+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)
+is unbounded, but the receiver rejects any manifest beyond `MAX_MANIFEST_ENTRIES`
+(1 048 576 entries) / `MAX_MANIFEST_BYTES` (16 MB of paths) as a hard protocol
+error. In the commit modes this only means the deletion is refused after the
+data already arrived; in the NEW early modes (`--delete-before`/`--delete-during`)
+the manifest is the first frame, so an oversized keep-set now aborts the whole
+transfer BEFORE any data is sent (previously all data transferred and only the
+deletion step failed). 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`
+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
+gone still aborts the wait via connection close/error; the extended bound only
+protects against aborting after the deletion already committed on the receiver.
+
+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.
+
 ## 8. Metadata Preservation
 
 | Flag | Rsync Description | FastSync Status | Notes |
@@ -171,9 +210,9 @@ This document maps rsync's full feature set to FastSync's current implementation
 |------|-------------------|-----------------|-------|
 | `--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 | ❌ Not Implemented | Removed because it had no effect |
-| `--copy-dest=DIR` | Include copies of unchanged files | ❌ Not Implemented | Removed because it had no effect |
-| `--link-dest=DIR` | Hardlink to files when unchanged | ❌ Not Implemented | Removed because it had no effect |
+| `--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) |
+| `--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 |
+| `--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 |
 | `--fuzzy`, `--no-fuzzy` | Find similar file for basis | ❌ Not Implemented | |
 
 ## 12. Compression

src/client/client_cli.c

@@ -114,6 +114,27 @@ static int set_nonneg_int_option(int* dest, const char* value, const char* optio
   return 0;
 }
 
+/* Validate and append one --compare-dest/--copy-dest/--link-dest directory.
+ * The path is interpreted on the receiver relative to the destination root,
+ * so it must be a non-empty relative path with no "." / ".." components (an
+ * absolute or escaping path is rejected up front instead of failing on the
+ * server). Returns 0 on success, -1 on error. */
+static int set_basis_dest_option(Config* config, BasisDestType type, const char* value,
+                                 const char* option_name) {
+  if (!value || !value[0]) {
+    log_message(LOG_LEVEL_ERROR, "missing argument for %s", option_name);
+    return -1;
+  }
+  if (config_basis_append(config, type, value) != 0) {
+    log_message(LOG_LEVEL_ERROR,
+                "%s requires a non-empty relative directory name with no '.', '..', or absolute "
+                "path (resolved below the destination root)",
+                option_name);
+    return -1;
+  }
+  return 0;
+}
+
 static int set_stderr_mode(const char* value) {
   if (strcmp(value, "errors") == 0 || strcmp(value, "e") == 0)
     log_set_stderr_mode(LOG_STDERR_ERRORS);
@@ -370,7 +391,6 @@ typedef enum {
   OPT_POS_INT,
   OPT_NONNEG_INT,
   OPT_ULL,
-  OPT_UNSUPPORTED,
 } OptKind;
 
 typedef struct {
@@ -430,7 +450,10 @@ static const OptionEntry OPTION_TABLE[] = {
     {"--old-d", NULL, OPT_FLAG, offsetof(Config, dirs)},
     {"--relative", "-R", OPT_FLAG, offsetof(Config, relative)},
     {"--mkpath", NULL, OPT_FLAG, offsetof(Config, mkpath)},
-    {"--delete-during", "--del", OPT_UNSUPPORTED, 0},
+    {"--delete-before", NULL, OPT_FLAG, offsetof(Config, delete_before)},
+    {"--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)},
 
     {"--source-dir", NULL, OPT_STRING, offsetof(Config, send_directory)},
     {"--dest-dir", NULL, OPT_STRING, offsetof(Config, receive_root_directory)},
@@ -547,8 +570,7 @@ static int apply_negation(Config* config, const char* arg) {
   return 0;
 }
 
-static int apply_table_option(Config* config, const OptionEntry* entry, const char* option_name,
-                              const char* value) {
+static int apply_table_option(Config* config, const OptionEntry* entry, const char* value) {
   if (entry->kind == OPT_NOOP)
     return 0;
   void* field = (char*)config + entry->offset;
@@ -578,11 +600,6 @@ static int apply_table_option(Config* config, const OptionEntry* entry, const ch
     *(unsigned long long*)field = v;
     return 0;
   }
-  case OPT_UNSUPPORTED: {
-    const char* reason = "delete-during is not implemented";
-    log_message(LOG_LEVEL_ERROR, "%s: %s; refusing to ignore option", option_name, reason);
-    return -1;
-  }
   }
   return -1;
 }
@@ -665,10 +682,8 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args,
     if (!entry)
       entry = find_table_option_with_equals(argv[i], &inline_value);
     if (entry) {
-      const char* option_name = argv[i];
       const char* value = NULL;
       if (entry->kind != OPT_FLAG) {
-        if (entry->kind != OPT_UNSUPPORTED) {
         value = inline_value;
         if (!value && i + 1 < argc)
           value = argv[++i];
@@ -676,12 +691,11 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args,
           log_message(LOG_LEVEL_ERROR, "missing argument for %s", entry->name);
           return -1;
         }
-        }
         if (strcmp(entry->name, "--compress-choice") == 0) {
           if (set_compression_choice(config, value) != 0)
             return -1;
         } else {
-          if (apply_table_option(config, entry, option_name, value) != 0)
+          if (apply_table_option(config, entry, value) != 0)
             return -1;
           if (strcmp(entry->name, "--compress-level") == 0 &&
               (config->compression_level < 1 || config->compression_level > 22)) {
@@ -697,11 +711,18 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args,
             config->use_metadata = true;
           }
         }
-      } else if (apply_table_option(config, entry, option_name, NULL) != 0) {
+      } else if (apply_table_option(config, entry, NULL) != 0) {
         return -1;
       }
       if (entry->offset == offsetof(Config, eight_bit_output))
         protocol_set_8_bit_output(true);
+      /* A delete-timing flag selects when --delete removes extras, so it
+         implies --delete exactly like the rsync options do. */
+      if (entry->offset == offsetof(Config, delete_before) ||
+          entry->offset == offsetof(Config, delete_during) ||
+          entry->offset == offsetof(Config, delete_delay) ||
+          entry->offset == offsetof(Config, delete_after))
+        config->use_delete = true;
       continue;
     }
 
@@ -974,6 +995,36 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args,
       }
       log_message(LOG_LEVEL_ERROR, "%s is not supported yet (xxHash64 is used)", argv[i]);
       return -1;
+    } else if (strncmp(argv[i], "--compare-dest=", 15) == 0) {
+      if (set_basis_dest_option(config, BASIS_DEST_COMPARE, argv[i] + 15, "--compare-dest") != 0)
+        return -1;
+    } else if (opt_is(argv[i], "--compare-dest", NULL)) {
+      if (i + 1 >= argc) {
+        log_message(LOG_LEVEL_ERROR, "missing argument for %s", argv[i]);
+        return -1;
+      }
+      if (set_basis_dest_option(config, BASIS_DEST_COMPARE, argv[++i], "--compare-dest") != 0)
+        return -1;
+    } else if (strncmp(argv[i], "--copy-dest=", 12) == 0) {
+      if (set_basis_dest_option(config, BASIS_DEST_COPY, argv[i] + 12, "--copy-dest") != 0)
+        return -1;
+    } else if (opt_is(argv[i], "--copy-dest", NULL)) {
+      if (i + 1 >= argc) {
+        log_message(LOG_LEVEL_ERROR, "missing argument for %s", argv[i]);
+        return -1;
+      }
+      if (set_basis_dest_option(config, BASIS_DEST_COPY, argv[++i], "--copy-dest") != 0)
+        return -1;
+    } else if (strncmp(argv[i], "--link-dest=", 12) == 0) {
+      if (set_basis_dest_option(config, BASIS_DEST_LINK, argv[i] + 12, "--link-dest") != 0)
+        return -1;
+    } else if (opt_is(argv[i], "--link-dest", NULL)) {
+      if (i + 1 >= argc) {
+        log_message(LOG_LEVEL_ERROR, "missing argument for %s", argv[i]);
+        return -1;
+      }
+      if (set_basis_dest_option(config, BASIS_DEST_LINK, argv[++i], "--link-dest") != 0)
+        return -1;
     } else if (argv[i][0] == '-') {
       char* escaped = output_escape(argv[i], false);
       fprintf(stderr, "Unknown option: %s\n", escaped ? escaped : "<allocation failed>");
@@ -1010,6 +1061,13 @@ int parse_args(Config* config, int argc, char* argv[], int* positional_args,
     config->files_from_set = set;
   }
 
+  /* The "unchanged" decision for --compare-dest/--copy-dest/--link-dest must
+   * be made on the receiver against the basis directories, which requires the
+   * per-file STATUS_CHECK handshake: basis-dir options therefore imply
+   * --incremental (and, via the block below, metadata) on the sender. */
+  if (config_has_basis(config))
+    config->use_incremental = true;
+
   /* Incremental and delta transfers need metadata unless the user disabled it. */
   if ((config->use_incremental || config->use_delta) && !config->use_metadata &&
       !config->metadata_explicitly_disabled) {

src/client/client_send.c

@@ -218,6 +218,49 @@ static bool files_from_list_valid(const Config* config) {
   return no_implied_dirs_files_from_valid(config);
 }
 
+/* Basis directories are honored by the receiver's per-file incremental check,
+   which (like every whole-file payload path in FastSync) is bounded by
+   MAX_RECEIVE_WHOLE_FILE_SIZE.  rsync would apply basis dirs to files of any
+   size; FastSync cannot, so when basis dirs are requested this preflight scan
+   refuses the run up front with a clear diagnostic instead of letting the
+   receiver abort the whole transfer mid-stream with no client explanation.
+   Returns true when the tree can be transferred. */
+static bool basis_oversize_preflight(const Config* config) {
+  PreparedScanner prepared;
+  if (!prepare_scanner(config, 0, &prepared))
+    return false;
+  DirectoryScanner* scanner =
+      directory_scanner_create_with_options(config->send_directory, &prepared.options);
+  prepared_scanner_destroy(&prepared);
+  if (!scanner)
+    return false;
+  bool ok = true;
+  Chunk* chunk;
+  while ((chunk = directory_scanner_next(scanner)) != NULL) {
+    for (int i = 0; i < chunk->element_count; i++) {
+      File* f = chunk->items[i];
+      if (f == NULL || f->is_dir || f->data == NULL || f->data->size <= MAX_RECEIVE_WHOLE_FILE_SIZE)
+        continue;
+      char* escaped = output_escape(file_wire_path(f), config->eight_bit_output);
+      log_message(LOG_LEVEL_ERROR,
+                  "%s is %llu bytes, larger than the %llu-byte whole-file transfer limit; "
+                  "--compare-dest/--copy-dest/--link-dest cannot sync files above this limit",
+                  escaped ? escaped : "<allocation failed>", (unsigned long long)f->data->size,
+                  (unsigned long long)MAX_RECEIVE_WHOLE_FILE_SIZE);
+      free(escaped);
+      ok = false;
+      break;
+    }
+    chunk_destroy(chunk);
+    if (!ok)
+      break;
+  }
+  if (directory_scanner_failed(scanner))
+    ok = false;
+  directory_scanner_destroy(scanner);
+  return ok;
+}
+
 /* Select the configured transport for both transfer execution paths. */
 static Client* connect_transfer_client(const Config* config) {
   if (config->transport == TRANSPORT_SSH) {
@@ -254,10 +297,6 @@ static void disconnect_transfer_client(Client* client) {
   client_delete(client);
 }
 
-static ArrayList* create_transfer_manifest(const Config* config) {
-  return config->use_delete ? array_list_create(free) : NULL;
-}
-
 static bool add_chunk_to_manifest(ArrayList* manifest, const Chunk* chunk) {
   if (!manifest)
     return true;
@@ -597,6 +636,59 @@ static int send_delete_manifest(int fd, ArrayList* manifest) {
   return 0;
 }
 
+/* Transmit the keep-set manifest and wait for the receiver's verdict.  Used by
+   --delete-before/--delete-during, where the extras are removed on the receiver
+   BEFORE the first byte of file data is sent: the receiver acknowledges with
+   STATUS_OK once the bounded delete committed, or STATUS_ERROR if it could not
+   (in which case the sender aborts without streaming any data).  The ACK may
+   take much longer than an ordinary per-message round trip because the receiver
+   performs the whole bounded deletion walk (up to MAX_SERVER_DELETE_COUNT
+   unlinks) before replying, so the wait uses a generous explicit deadline
+   instead of the default 60 s receive window. */
+#define DELETE_ACK_TIMEOUT_SEC 3600
+
+static bool send_delete_manifest_early(Client* client, ArrayList* manifest) {
+  if (!client || !manifest)
+    return false;
+  if (send_delete_manifest(client->file_descriptor, manifest) != 0)
+    return false;
+  Status ack;
+  if (!receive_status_timed(client->file_descriptor, &ack, DELETE_ACK_TIMEOUT_SEC))
+    return false;
+  if (ack != STATUS_OK) {
+    log_message(LOG_LEVEL_ERROR, "Server failed to delete files before the transfer");
+    return false;
+  }
+  return true;
+}
+
+/* Walk the whole source tree once collecting only destination-relative wire
+   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. */
+static bool scan_paths_only(const Config* config, const ScannerOptions* options,
+                            ArrayList* manifest) {
+  DirectoryScanner* scanner =
+      directory_scanner_create_with_options(config->send_directory, options);
+  if (!scanner)
+    return false;
+  bool ok = true;
+  Chunk* chunk;
+  while ((chunk = directory_scanner_next(scanner)) != NULL) {
+    if (!add_chunk_to_manifest(manifest, chunk)) {
+      ok = false;
+      chunk_destroy(chunk);
+      break;
+    }
+    chunk_destroy(chunk);
+  }
+  if (ok && directory_scanner_failed(scanner))
+    ok = false;
+  directory_scanner_destroy(scanner);
+  return ok;
+}
+
 static int incremental_check(Client* client, File* file, const Config* config,
                              DeltaSignature** out_sig) {
   *out_sig = NULL;
@@ -613,7 +705,10 @@ static int incremental_check(Client* client, File* file, const Config* config,
     return -1;
   if (!send_n_data(client->file_descriptor, &mtime_nsec, sizeof(mtime_nsec)))
     return -1;
-  if (config->checksum) {
+  /* With alternate basis directories the receiver must be able to verify the
+   * content of every candidate basis file, so the sender supplies its xxHash64
+   * for every file even when --checksum was not requested. */
+  if (config->checksum || config_has_basis(config)) {
     uint64_t checksum;
     if (!file_checksum(file, &checksum) ||
         !send_n_data(client->file_descriptor, &checksum, sizeof(checksum)))
@@ -906,6 +1001,17 @@ static int send_chunks_multithreaded(void* pipeline_context) {
     protocol_session_unbind();
     return thrd_error;
   }
+  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)) {
+      pipeline_cancel(context);
+      disconnect_transfer_client(client);
+      mark_sender_done(context);
+      protocol_session_unbind();
+      return thrd_error;
+    }
+  }
 
   while (true) {
     Chunk* current_chunk = queue_dequeue_multithreaded(
@@ -919,7 +1025,7 @@ static int send_chunks_multithreaded(void* pipeline_context) {
         protocol_session_unbind();
         return thrd_error;
       }
-      if (context->config->use_delete) {
+      if (context->config->use_delete && !context->early_delete) {
         if (send_delete_manifest(client->file_descriptor, context->manifest) != 0)
           goto send_fail;
       }
@@ -1013,7 +1119,7 @@ static int scan_directory_multithreaded(void* pipeline_context) {
       failed = dirs_mode ? directory_scanner_failed(dscanner) : parallel_scanner_failed(scanner);
       break;
     }
-    if (context->config->use_delete) {
+    if (context->config->use_delete && !context->early_delete) {
       mtx_lock(&context->mutex_scanner);
       bool manifest_ok = add_chunk_to_manifest(context->manifest, current_chunk);
       mtx_unlock(&context->mutex_scanner);
@@ -1156,6 +1262,8 @@ int send_files(Config* config) {
     return send_dry_run_manifest(config);
   if (!files_from_list_valid(config))
     return 1;
+  if (config_has_basis(config) && !basis_oversize_preflight(config))
+    return 1;
 
   Client* client = connect_transfer_client(config);
   if (!client) {
@@ -1172,19 +1280,46 @@ int send_files(Config* config) {
   DirectoryScanner* scanner = NULL;
   ArrayList* manifest = NULL;
   ArrayList* remove_sources = NULL;
+  bool delete_early = config->use_delete && config_delete_timing_early(config);
+  bool send_failed = false;
   PreparedScanner prepared;
   memset(&prepared, 0, sizeof(prepared));
   if (!config_send(client->file_descriptor, config))
     goto send_fail;
   if (!prepare_scanner(config, 0, &prepared))
     goto send_fail;
-  scanner = directory_scanner_create_with_options(config->send_directory, &prepared.options);
-  manifest = create_transfer_manifest(config);
   if (config->remove_source_files)
     remove_sources = array_list_create(source_file_destroy);
-  if (!scanner || (config->use_delete && !manifest) ||
-      (config->remove_source_files && !remove_sources))
+  if (config->remove_source_files && !remove_sources)
     goto send_fail;
+  /* 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
+     dedicated path-only pre-scan, so no manifest is kept during the data pass. */
+  if (delete_early) {
+    /* Pass 1: collect the complete keep-set (paths only, no data loaded) and
+       transmit it now, before any file data.  The receiver removes extras and
+       acks; the transfer aborts here if the deletion could not commit. */
+    ArrayList* early_manifest = array_list_create(free);
+    if (!early_manifest)
+      goto send_fail;
+    if (!scan_paths_only(config, &prepared.options, early_manifest)) {
+      array_list_delete(early_manifest);
+      goto send_fail;
+    }
+    bool early_ok = send_delete_manifest_early(client, early_manifest);
+    array_list_delete(early_manifest);
+    if (!early_ok)
+      goto send_fail;
+  } else if (config->use_delete) {
+    manifest = array_list_create(free);
+    if (!manifest)
+      goto send_fail;
+  }
+  scanner = directory_scanner_create_with_options(config->send_directory, &prepared.options);
+  if (!scanner)
+    goto send_fail;
+
   Chunk* current_chunk;
   unsigned long long total_bytes = 0;
   int total_files = 0;
@@ -1196,7 +1331,7 @@ int send_files(Config* config) {
       chunk_bytes += current_chunk->items[i]->data->size;
       total_files++;
     }
-    if (!add_chunk_to_manifest(manifest, current_chunk)) {
+    if (manifest && !add_chunk_to_manifest(manifest, current_chunk)) {
       chunk_destroy(current_chunk);
       goto send_fail;
     }
@@ -1220,9 +1355,7 @@ int send_files(Config* config) {
     if (send_chunk_with_removal(client, current_chunk, config, remove_sources) != 0) {
       log_message(LOG_LEVEL_ERROR, "Failed to send chunk");
       chunk_destroy(current_chunk);
-      if (manifest)
-        array_list_delete(manifest);
-      manifest = NULL;
+      send_failed = true;
       break;
     }
     total_bytes += chunk_bytes;
@@ -1235,9 +1368,18 @@ int send_files(Config* config) {
     }
     chunk_destroy(current_chunk);
   }
-  if (directory_scanner_failed(scanner) || (config->use_delete && manifest == NULL))
+  if (send_failed) {
+    if (manifest) {
+      array_list_delete(manifest);
+      manifest = NULL;
+    }
     goto send_fail;
-  if (config->use_delete) {
+  }
+  if (directory_scanner_failed(scanner))
+    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) {
       array_list_delete(manifest);
       manifest = NULL;
@@ -1295,6 +1437,8 @@ int send_files_multithreaded(Config** config_ptr) {
     return send_dry_run_manifest(config);
   if (!files_from_list_valid(config))
     return 1;
+  if (config_has_basis(config) && !basis_oversize_preflight(config))
+    return 1;
 
   long pages = sysconf(_SC_AVPHYS_PAGES);
   long page_size = sysconf(_SC_PAGE_SIZE);
@@ -1324,8 +1468,28 @@ int send_files_multithreaded(Config** config_ptr) {
     return 1;
   }
   *config_ptr = NULL; /* context now owns config through all remaining paths */
-  if (config->use_delete)
+  if (config->use_delete) {
     context->manifest = array_list_create(free);
+    if (!context->manifest) {
+      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. */
+      PreparedScanner prepared;
+      memset(&prepared, 0, sizeof(prepared));
+      bool prebuilt = prepare_scanner(config, 4, &prepared) &&
+                      scan_paths_only(config, &prepared.options, context->manifest);
+      prepared_scanner_destroy(&prepared);
+      if (!prebuilt) {
+        pipeline_context_sender_destroy(context);
+        return 1;
+      }
+      context->early_delete = true;
+    }
+  }
   if (config->remove_source_files)
     context->remove_source_files = array_list_create(source_file_destroy);
   if ((config->use_delete && !context->manifest) ||

src/client/client_validation.c

@@ -11,6 +11,12 @@ bool validate_config(const Config* config) {
     print_usage();
     return false;
   }
+  if (config_has_basis(config) && config->use_chunk_serialization) {
+    log_message(LOG_LEVEL_ERROR,
+                "--compare-dest/--copy-dest/--link-dest require per-file incremental checks and "
+                "cannot be combined with -s (chunk serialization)");
+    return false;
+  }
   if (config->use_sendfile && (config->use_chunk_serialization || config->use_compression)) {
     log_message(LOG_LEVEL_ERROR, "-f/--sendfile cannot be combined with -c (compression) or -s "
                                  "(chunk serialization)");
@@ -71,5 +77,11 @@ bool validate_config(const Config* config) {
                 "staging directory)");
     return false;
   }
+  if (!config_has_valid_delete_timing(config)) {
+    log_message(LOG_LEVEL_ERROR,
+                "--delete-before/--delete-during/--delete-delay/--delete-after select the delete "
+                "timing; at most one may be given and each implies --delete");
+    return false;
+  }
   return true;
 }

src/client/usage.c

@@ -24,6 +24,19 @@ void print_usage(void) {
   printf("  -P                  Partial mode with progress (retention incomplete)\n");
   printf("  -8, --8-bit-output  Leave high-bit characters unescaped in output\n");
   printf("  --delete            Delete files on receiver not in source\n");
+  printf("                      (default timing: delete only after the whole\n");
+  printf("                      transfer has succeeded)\n");
+  printf("  --delete-before     Delete extras before the transfer starts\n");
+  printf("                      (implies --delete)\n");
+  printf("  --delete-during     Delete extras once the keep-set manifest is known,\n");
+  printf("                      before the data is applied (implies --delete)\n");
+  printf("  --del               Alias for --delete-during\n");
+  printf("  --delete-delay      Delete extras only after a successful transfer\n");
+  printf("                      (implies --delete)\n");
+  printf("  --delete-after      Delete only after the whole transfer succeeded\n");
+  printf("                      (the default --delete timing; implies --delete)\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");
   printf("  --delay-updates    Put updated files into place only at the end of transfer\n");
   printf("  --dirs, -d, --old-dirs, --old-d  Transfer the named directory entries without\n");
@@ -37,7 +50,6 @@ void print_usage(void) {
   printf("                    parent directory is not itself listed\n");
   printf("  --mkpath          Create the destination root directory on the server when it\n");
   printf("                    does not exist yet\n");
-  printf("  --del               Alias for --delete-during (not implemented)\n");
   printf("  --exclude <pattern> Exclude files matching pattern\n");
   printf("  --include <pattern> Only include files matching pattern\n");
   printf("  --exclude-from <file> Read exclude patterns from file\n");
@@ -58,6 +70,13 @@ void print_usage(void) {
   printf("  -@, --modify-window <sec>  Modification time tolerance\n");
   printf("  -u, --update        Skip files newer than the source on receiver\n");
   printf("  --existing          Skip files not already present at destination\n");
+  printf("  --compare-dest <dir>  Treat DIR (relative to destination root) as an extra\n");
+  printf("                    comparison basis: unchanged files are not transferred\n");
+  printf("                    (requires --incremental, which is implied)\n");
+  printf("  --copy-dest <dir>   Like --compare-dest, but copies the unchanged file from DIR\n");
+  printf("                    into the destination instead of transferring its data\n");
+  printf("  --link-dest <dir>   Like --copy-dest, but hard-links the unchanged file from DIR\n");
+  printf("                    into the destination (repeatable; earlier DIRs win)\n");
   printf("  --checksum-choice, --cc <alg>  Checksum algorithm (not supported yet; xxHash64 is "
          "used)\n");
   printf("  --delta             Delta transfer for changed files (requires --incremental)\n");

src/server/receiver.c

@@ -129,20 +129,40 @@ static bool receiver_process_batch(Config* config, int file_descriptor) {
 }
 
 int receiver_process(Config* config, int file_descriptor, const ReceiverSink* sink) {
+  return receiver_process_pending(config, file_descriptor, sink, NULL);
+}
+
+/* Runs the whole receive loop.  The delete manifest may legitimately arrive
+   either FIRST (--delete-before / --delete-during: the sender transmits the
+   validated keep-set before any file data) or LAST (plain --delete /
+   --delete-after / --delete-delay: the manifest closes the data stream).  In
+   the early modes the receiver deletes as soon as the manifest has been read
+   and acknowledges with STATUS_OK so the sender only starts streaming once the
+   deletion has committed (or failed); in the late modes the manifest is held
+   and the deletion is committed only after the terminal STATUS_FINISHED proves
+   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) {
   Status status;
   if (!receive_status(file_descriptor, &status))
     return -1;
+  bool early_delete = config_delete_timing_early(config);
+  /* 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;
   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_MKDIR || status == STATUS_MANIFEST) {
     if (status == STATUS_KEEPALIVE) {
       if (!send_status(file_descriptor, STATUS_KEEPALIVE))
-        return -1;
-      goto next;
+        goto fail;
+      goto next_status;
     }
     if (status == STATUS_ABORT) {
       log_message(LOG_LEVEL_INFO, "Received abort from client, cleaning up");
-      return -1;
+      goto fail;
     }
     if (status == STATUS_CHECK) {
       bool skipped;
@@ -155,12 +175,46 @@ int receiver_process(Config* config, int file_descriptor, const ReceiverSink* si
         goto receive_error;
     } else if (status == STATUS_CHECK_BATCH) {
       if (!receiver_process_batch(config, file_descriptor))
-        return -1;
-      goto next;
+        goto fail;
+      goto next_status;
     } else if (status == STATUS_MKDIR) {
       File* dir = file_receive_directory(file_descriptor);
       if (!dir || !sink->store_file(dir, sink->context))
         goto receive_error;
+    } else if (status == STATUS_MANIFEST) {
+      ArrayList* manifest = receive_manifest_entries(file_descriptor);
+      if (!manifest)
+        goto fail; /* receive_manifest_entries already sent STATUS_ERROR */
+      if (early_delete) {
+        /* --delete-before / --delete-during: the manifest is authoritative the
+           moment it arrives, before any file data.  Delete now and acknowledge
+           so the sender only starts streaming once the deletion committed (or
+           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);
+        if (!deletion_ok) {
+          send_status(file_descriptor, STATUS_ERROR);
+          goto fail;
+        }
+        if (!send_status(file_descriptor, STATUS_OK))
+          goto fail;
+      } else if (config->use_delete) {
+        /* Plain --delete / --delete-after / --delete-delay: hold the keep-set
+           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);
+          deferred_manifest = NULL;
+          array_list_delete(manifest);
+          send_status(file_descriptor, STATUS_ERROR);
+          goto fail;
+        }
+        deferred_manifest = manifest;
+      } else {
+        array_list_delete(manifest);
+      }
+      goto next_status;
     } else {
       File* file = file_receive(config, file_descriptor);
       if (!file) {
@@ -170,27 +224,61 @@ int receiver_process(Config* config, int file_descriptor, const ReceiverSink* si
       if (!sink->store_file(file, sink->context))
         goto receive_error;
     }
-  next:
+  next_status:
     if (!receive_status(file_descriptor, &status))
       goto receive_error;
   }
-  if (status == STATUS_MANIFEST && receive_manifest(file_descriptor, config, &status) != 0)
-    return -1;
   if (status != STATUS_FINISHED) {
     log_message(LOG_LEVEL_ERROR, "Did not receive FINISHED Status");
     goto receive_error;
   }
+  /* Commit-style (late) deletion: every data frame has been received and the
+     sender proved the whole tree with STATUS_FINISHED.  The single-threaded
+     receiver stores files synchronously, so everything is on disk here and the
+     deletion can be committed before the --delay-updates publication in
+     send_success (the walker skips the staging dir, so staged files are never
+     treated as extras).  The -m receiver passes `pending_manifest` because its
+     disk writer may still be draining; the caller commits after the writer has
+     joined so no extra file is removed unless the transfer is known to have
+     succeeded. */
+  if (deferred_manifest) {
+    if (pending_manifest) {
+      *pending_manifest = deferred_manifest;
+      deferred_manifest = NULL;
+    } else {
+      bool deletion_ok = manifest_delete_extras(config, deferred_manifest);
+      array_list_delete(deferred_manifest);
+      deferred_manifest = NULL;
+      if (!deletion_ok) {
+        send_status(file_descriptor, STATUS_ERROR);
+        goto fail;
+      }
+    }
+  }
   if (sink->send_success) {
     if (sink->send_success_frame) {
       if (!sink->send_success_frame(file_descriptor, sink->context))
-        return -1;
+        goto fail;
     } else if (!send_status(file_descriptor, STATUS_OK)) {
-      return -1;
+      goto fail;
     }
   }
   return 0;
 
+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);
+    deferred_manifest = NULL;
+  }
+  return -1;
+
 receive_error:
+  if (deferred_manifest) {
+    array_list_delete(deferred_manifest);
+    deferred_manifest = NULL;
+  }
   if (sink->send_error)
     send_status(file_descriptor, STATUS_ERROR);
   return -1;
@@ -214,7 +302,7 @@ static bool receiver_save_file(File* file, void* context_pointer) {
     result = file_save_to_disk_full(context->config->receive_root_directory, file, context->config);
   }
   if (result != FILE_SAVE_ERROR && context->config->remove_source_files && !file->is_dir &&
-      !receiver_outcomes_append(&context->outcomes, (unsigned char)result)) {
+      !file->skip && !receiver_outcomes_append(&context->outcomes, (unsigned char)result)) {
     file_destroy(file);
     return false;
   }

src/server/receiver.h

@@ -35,6 +35,14 @@ void receiver_outcomes_destroy(ReceiverOutcomes* outcomes);
 bool receiver_send_final_success(int fd, const Config* config, const ReceiverOutcomes* outcomes);
 
 int receiver_process(Config* config, int file_descriptor, const ReceiverSink* sink);
+/* receiver_process with an escape hatch for the commit-style (late) deletion:
+   when `pending_manifest` is non-NULL the receiver does NOT delete at
+   STATUS_FINISHED itself; instead it stores the owned keep-set manifest there
+   (leaving *pending_manifest untouched on early modes/errors) so the caller can
+   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);
 int receiver_receive_files(Config* config, int file_descriptor);
 
 #endif

src/server/server.c

@@ -255,6 +255,20 @@ void handler(int file_descriptor) {
     thrd_join(receiver, &receiver_result);
     thrd_join(writer, &writer_result);
     bool transfer_ok = receiver_result == thrd_success && writer_result == thrd_success;
+    if (transfer_ok) {
+      /* Commit-style (late) deletion: receive_thread handed the keep-set
+         manifest here instead of deleting while write_thread might still be
+         draining, so by now every file is on disk and the whole transfer is
+         known to have succeeded.  Remove the extras before publishing a
+         --delay-updates run; the walker skips the staging directory. */
+      if (context->deferred_manifest) {
+        if (!manifest_delete_extras(config, context->deferred_manifest)) {
+          transfer_ok = false;
+        }
+        array_list_delete(context->deferred_manifest);
+        context->deferred_manifest = NULL;
+      }
+    }
     if (transfer_ok) {
       /* --delay-updates: receive_thread has finished the whole protocol stream
          (including manifest/delete handling) and write_thread has drained its

src/shared/config.c

@@ -109,12 +109,13 @@ static void config_set_defaults(Config* config) {
   config->rsync_path = NULL;
   config->old_args = false;
   config->temp_dir = NULL;
-  config->compare_dest = NULL;
-  config->copy_dest = NULL;
-  config->link_dest = NULL;
+  config->basis_dirs = NULL;
+  config->basis_count = 0;
   config->partial_dir = NULL;
   config->suffix = NULL;
   config->delete_before = false;
+  config->delete_during = false;
+  config->delete_delay = false;
   config->address = NULL;
   config->bind_address = NULL;
   config->ipv6 = false;
@@ -161,12 +162,14 @@ static bool validate_received_config(const Config* config) {
          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_delay) && valid_wire_bool(config->delete_during) &&
          valid_wire_bool(config->relative) && valid_wire_bool(config->prune_empty_dirs) &&
          valid_wire_bool(config->delay_updates) && 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) &&
+         config_has_valid_delete_timing(config) &&
          !(config->skip_compress_set && config->use_chunk_serialization) &&
          (!config->use_compression ||
           (config->compression_level >= 1 && config->compression_level <= 22)) &&
@@ -188,6 +191,107 @@ Config* config_create(void) {
   return config;
 }
 
+bool config_delete_timing_early(const Config* config) {
+  if (!config)
+    return false;
+  return config->delete_before || config->delete_during;
+}
+
+/* A delete-timing flag is only meaningful together with --delete.  At most one
+   of the four flags may be set; several simultaneous timings are a client bug
+   and are rejected on both ends. */
+bool config_has_valid_delete_timing(const Config* config) {
+  if (!config)
+    return false;
+  if (!config->use_delete)
+    return !config->delete_before && !config->delete_during && !config->delete_delay &&
+           !config->delete_after;
+  int timing_count = (config->delete_before ? 1 : 0) + (config->delete_during ? 1 : 0) +
+                     (config->delete_delay ? 1 : 0) + (config->delete_after ? 1 : 0);
+  return timing_count <= 1;
+}
+
+bool config_has_basis(const Config* config) {
+  return config && config->basis_count > 0;
+}
+
+/* A basis-dir path travels from the client to the receiver and is resolved
+ * below the destination root, so it must be a non-empty relative path with no
+ * "." or ".." component and no traversal: an absolute or escaping path would
+ * make the receiver read or link files outside its authorized root.
+ *
+ * Returns a malloc'd CANONICAL copy of an accepted path, or NULL when the path
+ * is rejected.  Canonicalization collapses interior empty components ("a//b" ->
+ * "a/b"), drops "." components and trailing "/"s, so validation, the delete
+ * walker prefix match and the receiver's basis lookup all agree on one form.
+ * The normalizer is the single source of truth for both config_basis_path_valid
+ * and config_basis_append. */
+static char* basis_path_normalize(const char* path) {
+  if (!path || path[0] == '\0' || path[0] == '/' || has_path_traversal(path))
+    return NULL;
+  if (strcmp(path, ".") == 0)
+    return NULL;
+  char* dup = str_dup(path);
+  if (!dup)
+    return NULL;
+  size_t out_len = 0;
+  char* out = malloc(strlen(path) + 1);
+  if (!out) {
+    free(dup);
+    return NULL;
+  }
+  char* saveptr = NULL;
+  bool ok = true;
+  for (char* part = strtok_r(dup, "/", &saveptr); part; part = strtok_r(NULL, "/", &saveptr)) {
+    if (strcmp(part, "..") == 0) {
+      ok = false;
+      break;
+    }
+    if (strcmp(part, ".") == 0)
+      continue;
+    if (out_len > 0)
+      out[out_len++] = '/';
+    size_t len = strlen(part);
+    memcpy(out + out_len, part, len);
+    out_len += len;
+  }
+  free(dup);
+  if (!ok || out_len == 0) {
+    free(out);
+    return NULL;
+  }
+  out[out_len] = '\0';
+  return out;
+}
+
+bool config_basis_path_valid(const char* path) {
+  char* normalized = basis_path_normalize(path);
+  if (!normalized)
+    return false;
+  free(normalized);
+  return true;
+}
+
+int config_basis_append(Config* config, BasisDestType type, const char* path) {
+  if (!config ||
+      (type != BASIS_DEST_COMPARE && type != BASIS_DEST_COPY && type != BASIS_DEST_LINK) ||
+      config->basis_count >= MAX_BASIS_DIRS)
+    return -1;
+  char* normalized = basis_path_normalize(path);
+  if (!normalized)
+    return -1;
+  BasisDest* grown = realloc(config->basis_dirs, (config->basis_count + 1) * sizeof(BasisDest));
+  if (!grown) {
+    free(normalized);
+    return -1;
+  }
+  config->basis_dirs = grown;
+  config->basis_dirs[config->basis_count].type = type;
+  config->basis_dirs[config->basis_count].path = normalized;
+  config->basis_count++;
+  return 0;
+}
+
 bool config_is_remote_dest(const char* s) {
   if (s == NULL)
     return false;
@@ -244,9 +348,13 @@ void config_delete(Config* config) {
   free(config->rsh_command);
   free(config->rsync_path);
   free(config->temp_dir);
-  free(config->compare_dest);
-  free(config->copy_dest);
-  free(config->link_dest);
+  for (int i = 0; i < config->basis_count; i++) {
+    free(config->basis_dirs[i].path);
+    config->basis_dirs[i].path = NULL;
+  }
+  free(config->basis_dirs);
+  config->basis_dirs = NULL;
+  config->basis_count = 0;
   free(config->partial_dir);
   free(config->suffix);
   free(config->address);
@@ -311,7 +419,8 @@ static bool send_selection_options(int fd, const Config* c) {
          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_n_data(fd, &c->max_delete, sizeof(c->max_delete)) && send_int(fd, c->relative) &&
-         send_int(fd, c->prune_empty_dirs) && send_int(fd, c->mkpath);
+         send_int(fd, c->prune_empty_dirs) && 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) {
@@ -333,6 +442,17 @@ static bool send_resume_options(int fd, const Config* c) {
          send_str(fd, c->chmod_spec ? c->chmod_spec : "") && send_skip_compress_options(fd, c);
 }
 
+static bool send_basis_options(int fd, const Config* c) {
+  if (!send_int(fd, c->basis_count))
+    return false;
+  for (int i = 0; i < c->basis_count; i++) {
+    if (!send_int(fd, (int)c->basis_dirs[i].type) ||
+        !send_str(fd, c->basis_dirs[i].path ? c->basis_dirs[i].path : ""))
+      return false;
+  }
+  return true;
+}
+
 static bool receive_core_fields(int fd, Config* c) {
   int value;
   if (!receive_wire_bool(fd, &c->eight_bit_output))
@@ -418,7 +538,9 @@ static bool receive_selection_options(int fd, Config* c) {
     return false;
   if (!receive_wire_bool(fd, &c->mkpath))
     return false;
-  return true;
+  if (!receive_wire_bool(fd, &c->delete_during))
+    return false;
+  return receive_wire_bool(fd, &c->delete_delay);
 }
 
 static bool receive_resume_options(int fd, Config* c) {
@@ -479,12 +601,35 @@ static bool receive_resume_options(int fd, Config* c) {
   return true;
 }
 
+static bool receive_basis_options(int fd, Config* c) {
+  int count;
+  if (!receive_int(fd, &count))
+    return false;
+  if (count < 0 || count > MAX_BASIS_DIRS)
+    return false;
+  for (int i = 0; i < count; i++) {
+    int type;
+    if (!receive_int(fd, &type) || type <= BASIS_DEST_NONE || type > BASIS_DEST_LINK)
+      return false;
+    char* path = receive_str(fd);
+    if (!path)
+      return false;
+    /* config_basis_append validates and canonicalizes the path; a rejected
+       path (absolute / traversal / empty) drops the whole connection. */
+    bool ok = config_basis_append(c, (BasisDestType)type, path) == 0;
+    free(path);
+    if (!ok)
+      return false;
+  }
+  return true;
+}
+
 bool config_send(int file_descriptor, const Config* config) {
   protocol_session_set_max_alloc(NULL, config->max_alloc);
   if (!send_core_fields(file_descriptor, config) || !send_delta_fields(file_descriptor, config) ||
       !send_file_options(file_descriptor, config) ||
       !send_selection_options(file_descriptor, config) ||
-      !send_resume_options(file_descriptor, config))
+      !send_resume_options(file_descriptor, config) || !send_basis_options(file_descriptor, config))
     return false;
   Status status;
   if (!receive_status(file_descriptor, &status))
@@ -516,7 +661,8 @@ Config* config_receive(int file_descriptor) {
       !receive_delta_fields(file_descriptor, config) ||
       !receive_file_options(file_descriptor, config) ||
       !receive_selection_options(file_descriptor, config) ||
-      !receive_resume_options(file_descriptor, config))
+      !receive_resume_options(file_descriptor, config) ||
+      !receive_basis_options(file_descriptor, config))
     goto error;
   if (config->compress_choice[0] != '\0' && strcmp(config->compress_choice, "zstd") != 0 &&
       strcmp(config->compress_choice, "none") != 0) {

src/shared/config.h

@@ -12,6 +12,22 @@ typedef enum { TRANSPORT_TCP, TRANSPORT_SSH } TransportType;
    Config can carry it; the concrete type lives in delay_updates.h. */
 typedef struct DelayUpdatesContext DelayUpdatesContext;
 
+/* Alternate basis-directory modes (--compare-dest / --copy-dest /
+ * --link-dest).  Each flag adds one entry to the ordered Config->basis_dirs
+ * list; the receiver consults entries in command-line order and stops at the
+ * first exact match, mirroring rsync's basis-dir priority rules. */
+typedef enum {
+  BASIS_DEST_NONE = 0,
+  BASIS_DEST_COMPARE, /* compare only: never copies, never materializes */
+  BASIS_DEST_COPY,    /* local copy of the matched basis file */
+  BASIS_DEST_LINK     /* hard link to the matched basis file */
+} BasisDestType;
+
+typedef struct BasisDest {
+  BasisDestType type;
+  char* path; /* relative to the destination root (receiver-confined) */
+} BasisDest;
+
 typedef struct Config {
   char* version;
   char* send_directory;
@@ -134,9 +150,12 @@ typedef struct Config {
   char* rsync_path;
   bool old_args;
   char* temp_dir;
-  char* compare_dest;
-  char* copy_dest;
-  char* link_dest;
+  /* Alternate basis directories, ordered by command-line appearance.  Each
+   * entry's type selects compare/copy/link behavior on an exact match.  These
+   * cross the wire so the receiver can consult them; they are interpreted
+   * relative to the destination root and confined there. */
+  BasisDest* basis_dirs;
+  int basis_count;
 
   // PR #174: Partial transfer resumption
   char* partial_dir;
@@ -147,6 +166,19 @@ typedef struct Config {
   // PR #179: Delete policies
   bool delete_before;
 
+  /* rsync deletion-timing family (real from Phase 3).  At most one of
+     delete_before / delete_during / delete_delay / delete_after may be set, and
+     only together with use_delete (the CLI implies --delete for each of them).
+     delete_before and delete_during select the EARLY engine mode: the keep-set
+     manifest is transmitted before any file data and extras are removed then,
+     acknowledged, before the first data byte.  delete_delay and delete_after
+     select the LATE commit mode: extras are removed only after the whole
+     transfer has succeeded (plain --delete keeps this mode).  The exact
+     semantics and the divergences from rsync are documented in RSYNC_COMPAT.md
+     and in config_delete_timing_early() below. */
+  bool delete_during;
+  bool delete_delay;
+
   // PR #181: IPv6 and bind address
   char* address;
   char* bind_address;
@@ -174,8 +206,10 @@ typedef struct Config {
   DelayUpdatesContext* delay_context;
 } Config;
 
-#define PROTOCOL_VERSION "2.7.0"
+#define PROTOCOL_VERSION "2.8.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
 
 Config* config_create(void);
 void config_delete(Config* config);
@@ -184,4 +218,22 @@ Config* config_receive(int file_descriptor);
 bool config_is_remote_dest(const char* s);
 void config_parse_ssh_dest(Config* config);
 
+/* True when the negotiated delete timing performs the extra-file deletion
+ * BEFORE the transfer data (--delete-before / --delete-during).  The flag is
+ * a pure function of the config and is used identically on the sender (to pick
+ * the manifest-first frame order) and the receiver (to delete when the early
+ * manifest arrives).  When false the deletion is committed only after the whole
+ * transfer succeeded (--delete / --delete-after / --delete-delay). */
+bool config_delete_timing_early(const Config* config);
+/* Delete-timing sanity: with deletion enabled at most one timing flag may be
+ * set (none = the default delete-after commit timing); without deletion no
+ * timing flag may be set (each timing flag implies --delete). */
+bool config_has_valid_delete_timing(const Config* config);
+/* True when at least one --compare-dest/--copy-dest/--link-dest was set. */
+bool config_has_basis(const Config* config);
+/* Append one basis-dir entry. Returns 0 on success, -1 on allocation failure. */
+int config_basis_append(Config* config, BasisDestType type, const char* path);
+/* Validate a client-provided basis-dir path (relative, confined, non-empty). */
+bool config_basis_path_valid(const char* path);
+
 #endif

src/shared/file.c

@@ -83,6 +83,7 @@ File* file_create(const char* path) {
   file->metadata = NULL;
   file->skip = false;
   file->is_dir = false;
+  file->basis_link = NULL;
   return file;
 }
 
@@ -98,6 +99,8 @@ void file_destroy(void* item) {
   file->path = NULL;
   free(file->send_path);
   file->send_path = NULL;
+  free(file->basis_link);
+  file->basis_link = NULL;
   free(file);
 }
 
@@ -630,6 +633,106 @@ bool file_to_disk_secure_no_replace(const char* path, const void* data,
                                   preserve_executability, false, true, false, temp_dir);
 }
 
+/* Atomic --link-dest install.  The destination is replaced (via a temporary
+ * name and a final rename) with a hard link to `basis_path`.  When a hard
+ * link cannot be created (the basis lives on a different filesystem, the
+ * filesystem refuses hard links, ...) the install falls back to writing a
+ * local copy from `data`/`data_size`, which the caller has already verified is
+ * byte-identical to the basis file.  `metadata` is only applied on that copy
+ * fallback; a successful hard link keeps the basis inode's own attributes
+ * (applying metadata through the shared inode would mutate the basis file).
+ * Returns false only when both the link and the copy fallback fail. */
+bool file_to_disk_secure_link(const char* path, const char* basis_path, const void* data,
+                              unsigned long long data_size, const FileMetadata* metadata,
+                              bool preserve_executability, bool use_fsync, const char* temp_dir) {
+  if (!path || !basis_path)
+    return false;
+  char* leaf = NULL;
+  int dirfd = file_open_secure_parent(path, &leaf, true);
+  if (dirfd < 0)
+    return false;
+
+  int scratch_dirfd = -1;
+  if (temp_dir) {
+    scratch_dirfd = file_open_private_dir(temp_dir);
+    if (scratch_dirfd < 0) {
+      int saved_errno = errno;
+      log_message(LOG_LEVEL_ERROR, "could not open --temp-dir scratch directory '%s': %s", temp_dir,
+                  strerror(saved_errno));
+      close(dirfd);
+      free(leaf);
+      return false;
+    }
+  }
+
+  char* basis_leaf = NULL;
+  int basis_dirfd = file_open_secure_parent(basis_path, &basis_leaf, false);
+  bool linked = false;
+  if (basis_dirfd >= 0 && basis_leaf != NULL) {
+    int tmp_size = snprintf(NULL, 0, ".%s.tmp.%ld.%llu", leaf, (long)getpid(), ~0ULL);
+    char* tmp = NULL;
+    if (tmp_size >= 0)
+      tmp = malloc((size_t)tmp_size + 1);
+    if (!tmp) {
+      log_message(LOG_LEVEL_ERROR, "memory allocation failed while hard-linking basis file");
+    } else {
+      for (unsigned int i = 0; i < 100 && !linked; ++i) {
+        if (scratch_dirfd >= 0)
+          snprintf(tmp, (size_t)tmp_size + 1, ".%s.tmp.%ld.%llu", leaf, (long)getpid(),
+                   next_temp_sequence());
+        else
+          snprintf(tmp, (size_t)tmp_size + 1, ".%s.tmp.%ld.%u", leaf, (long)getpid(), i);
+        if (linkat(basis_dirfd, basis_leaf, scratch_dirfd >= 0 ? scratch_dirfd : dirfd, tmp, 0) ==
+            0) {
+          linked = true;
+          break;
+        }
+        if (errno != EEXIST)
+          break; /* EXDEV / EPERM / ...: give up and fall back to a copy */
+      }
+      if (linked) {
+        int target_dirfd = scratch_dirfd >= 0 ? scratch_dirfd : dirfd;
+        if (use_fsync) {
+          int tfd = openat(target_dirfd, tmp, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+          if (tfd < 0 || fsync(tfd) != 0) {
+            linked = false;
+            if (tfd >= 0)
+              close(tfd);
+          } else {
+            close(tfd);
+          }
+        }
+        if (linked && renameat(target_dirfd, tmp, dirfd, leaf) != 0)
+          linked = false;
+        if (!linked)
+          unlinkat(target_dirfd, tmp, 0);
+      }
+      free(tmp);
+    }
+  }
+  if (basis_dirfd >= 0)
+    close(basis_dirfd);
+  free(basis_leaf);
+  basis_leaf = NULL;
+
+  if (!linked) {
+    if (scratch_dirfd >= 0)
+      close(scratch_dirfd);
+    close(dirfd);
+    free(leaf);
+    /* The basis file could not be linked in (missing, cross-device, refused
+       by the filesystem).  Write a byte-identical local copy instead. */
+    return file_to_disk_secure_with_fsync(path, data, data_size, false, false, metadata,
+                                          preserve_executability, use_fsync, temp_dir);
+  }
+
+  if (scratch_dirfd >= 0)
+    close(scratch_dirfd);
+  close(dirfd);
+  free(leaf);
+  return true;
+}
+
 bool file_write_to_disk(const char* path, const void* data, unsigned long long data_size,
                         bool inplace, bool sparse) {
   if (!path || (!data && data_size != 0) || has_path_traversal(path))

src/shared/file.h

@@ -63,5 +63,12 @@ bool file_to_disk_secure_no_replace(const char* path, const void* data,
                                     unsigned long long data_size, bool sparse,
                                     const FileMetadata* metadata, bool preserve_executability,
                                     const char* temp_dir);
+/* Atomic --link-dest install: replace `path` with a hard link to `basis_path`
+   (via a temp name + rename); fall back to a byte-identical local copy from
+   `data` when the link is impossible (EXDEV/EPERM/unsupported filesystem).
+   `metadata` is applied only on the copy fallback. */
+bool file_to_disk_secure_link(const char* path, const char* basis_path, const void* data,
+                              unsigned long long data_size, const FileMetadata* metadata,
+                              bool preserve_executability, bool use_fsync, const char* temp_dir);
 
 #endif

src/shared/file_receive.c

@@ -75,10 +75,17 @@ static FileSaveResult file_stage_delayed_update(const char* root_directory,
   /* The staged location is brand new (stale leftovers from a prior crash were
      wiped by prepare), so the plain atomic temp+rename engine installs the
      complete file there.  --temp-dir scratch is deliberately not layered on
-     top of the delay-updates staging tree. */
-  bool ok =
-      file_to_disk_secure_with_fsync(staged_path, file->data->data, file->data->size, false, sparse,
+     top of the delay-updates staging tree.  A --link-dest basis file is hard
+     linked into the staging tree (so publication's rename keeps the link). */
+  bool ok;
+  if (file->basis_link) {
+    ok = file_to_disk_secure_link(staged_path, file->basis_link, file->data->data, file->data->size,
                                   metadata, preserve_executability, config->use_fsync, NULL);
+  } else {
+    ok = file_to_disk_secure_with_fsync(staged_path, file->data->data, file->data->size, false,
+                                        sparse, metadata, preserve_executability, config->use_fsync,
+                                        NULL);
+  }
   if (!ok) {
     free(staged_path);
     return FILE_SAVE_ERROR;
@@ -272,16 +279,27 @@ FileSaveResult file_save_to_disk_full(const char* root_directory, const File* fi
     while (temp_len > 1 && confined_temp[temp_len - 1] == '/')
       confined_temp[--temp_len] = '\0';
   }
-  bool ok =
-      config && config->ignore_existing
+  /* A --link-dest basis hit installs an atomic hard link (with a byte-copy
+     fallback); --inplace and the update/no-replace write variants do not
+     apply to a fresh hard link, whose inode attributes already match.  The
+     existing/ignore-existing/update/backup preamble above has already made the
+     policy decision. */
+  bool ok;
+  if (config && file->basis_link) {
+    ok = file_to_disk_secure_link(disk_path, file->basis_link, file->data->data, file->data->size,
+                                  metadata, preserve_executability, config->use_fsync,
+                                  confined_temp);
+  } else {
+    ok = config && config->ignore_existing
              ? file_to_disk_secure_no_replace(disk_path, file->data->data, file->data->size, sparse,
                                               metadata, preserve_executability, confined_temp)
          : config && config->update
              ? file_to_disk_secure_update(disk_path, file->data->data, file->data->size, inplace,
                                           sparse, metadata, preserve_executability, confined_temp)
-          : file_to_disk_secure_with_fsync(disk_path, file->data->data, file->data->size, inplace,
-                                           sparse, metadata, preserve_executability,
+             : file_to_disk_secure_with_fsync(disk_path, file->data->data, file->data->size,
+                                              inplace, sparse, metadata, preserve_executability,
                                               config && config->use_fsync, confined_temp);
+  }
   free(confined_temp);
   confined_temp = NULL;
   if (!ok)
@@ -505,6 +523,143 @@ static File* receive_delta_file(int fd, const Config* config, const char* check_
   return NULL;
 }
 
+/* ---- Alternate basis directories (--compare-dest / --copy-dest / --link-dest) ----
+ * The receiver consults the ordered basis-dir list only when the destination
+ * entry is NOT already up to date.  An "exact match" requires an equal size,
+ * an equal mtime (unless --size-only), and an equal content xxHash64, so a
+ * hard link / local copy is only ever made from byte-identical content. */
+
+typedef struct BasisMatch {
+  bool hit;
+  BasisDestType type;
+  char* basis_path; /* owned absolute path of the matched basis file */
+  struct stat st;   /* fstat() of the matched basis file */
+  Data* content;    /* owned basis bytes (or empty Data), NULL when not loaded */
+} BasisMatch;
+
+static void basis_match_free(BasisMatch* match) {
+  if (!match)
+    return;
+  free(match->basis_path);
+  match->basis_path = NULL;
+  data_destroy(match->content);
+  match->content = NULL;
+  match->hit = false;
+  match->type = BASIS_DEST_NONE;
+}
+
+/* Open `path` (via the secure, root-confined primitives) and require it to be
+   a regular file of exactly `expected_size` bytes.  Returns an open read-only
+   descriptor and its fstat on success. */
+static bool basis_open_regular(const char* path, unsigned long long expected_size, int* out_fd,
+                               struct stat* out_st) {
+  char* leaf = NULL;
+  int parent_fd = file_open_secure_parent(path, &leaf, false);
+  if (parent_fd < 0)
+    return false;
+  int fd = openat(parent_fd, leaf, O_RDONLY | O_CLOEXEC | O_NOFOLLOW);
+  free(leaf);
+  close(parent_fd);
+  if (fd < 0)
+    return false;
+  struct stat st;
+  if (fstat(fd, &st) != 0 || !S_ISREG(st.st_mode) ||
+      (unsigned long long)st.st_size != expected_size) {
+    close(fd);
+    return false;
+  }
+  *out_fd = fd;
+  *out_st = st;
+  return true;
+}
+
+/* Read the whole remaining content of an open descriptor.  A zero-length file
+   yields an empty Data (data pointer NULL). */
+static Data* basis_read_content(int fd, unsigned long long size) {
+  if (size == 0)
+    return data_create_reserve(0);
+  if (size > MAX_RECEIVE_WHOLE_FILE_SIZE || size > SIZE_MAX)
+    return NULL;
+  void* buf = protocol_alloc((size_t)size);
+  if (!buf)
+    return NULL;
+  size_t got = 0;
+  while (got < (size_t)size) {
+    ssize_t n = read(fd, (char*)buf + got, (size_t)size - got);
+    if (n <= 0) {
+      free(buf);
+      return NULL;
+    }
+    got += (size_t)n;
+  }
+  return data_create(buf, (size_t)size);
+}
+
+/* --ignore-times forces every file to be updated, so no basis hit is ever
+   declared (matching rsync, where -I prevents link-dest from linking). */
+static bool basis_quick_matches(const Config* config, const struct stat* st, time_t check_mtime,
+                                long check_mtime_nsec) {
+  if (config->size_only)
+    return true;
+  long mtime_nsec = 0;
+#ifdef __linux__
+  mtime_nsec = st->st_mtim.tv_nsec;
+#endif
+  return metadata_mtime_matches(st->st_mtime, mtime_nsec, check_mtime, check_mtime_nsec,
+                                config->modify_window);
+}
+
+/* Search the basis-dir list in command-line order and return the first exact
+   match.  When load_content is true the matched bytes are kept in out->content
+   so the caller can materialize the file without re-reading it. */
+static bool basis_match_find(const Config* config, const char* check_path,
+                             unsigned long long check_size, time_t check_mtime,
+                             long check_mtime_nsec, uint64_t check_checksum, bool load_content,
+                             BasisMatch* out) {
+  memset(out, 0, sizeof(*out));
+  if (!config || !config_has_basis(config) || config->ignore_times)
+    return false;
+  for (int i = 0; i < config->basis_count; i++) {
+    const BasisDest* entry = &config->basis_dirs[i];
+    char* basis_dir = path_cat(config->receive_root_directory, entry->path);
+    if (!basis_dir)
+      continue;
+    char* candidate = path_cat(basis_dir, check_path);
+    free(basis_dir);
+    if (!candidate)
+      continue;
+
+    int fd;
+    struct stat st;
+    if (basis_open_regular(candidate, check_size, &fd, &st)) {
+      if (basis_quick_matches(config, &st, check_mtime, check_mtime_nsec)) {
+        Data* content = basis_read_content(fd, check_size);
+        if (content) {
+          uint64_t basis_hash = check_size == 0 ? delta_xxhash64("", 0)
+                                : content->data ? delta_xxhash64(content->data, content->size)
+                                                : 0;
+          if (basis_hash == check_checksum) {
+            out->hit = true;
+            out->type = entry->type;
+            out->basis_path = candidate;
+            candidate = NULL; /* ownership transferred to out */
+            out->st = st;
+            out->content = load_content ? content : NULL;
+            if (!load_content)
+              data_destroy(content);
+            close(fd);
+            return true;
+          }
+        }
+        data_destroy(content);
+      }
+      close(fd);
+    }
+    free(candidate);
+  }
+  return false;
+}
+
 File* receive_incremental_check(int fd, const Config* config, bool* skipped) {
   if (!config || !skipped) {
     send_status(fd, STATUS_ERROR);
@@ -531,7 +686,8 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) {
     send_status(fd, STATUS_ERROR);
     return NULL;
   }
-  if (config->checksum && !receive_n_data(fd, &check_checksum, sizeof(check_checksum))) {
+  if ((config->checksum || config_has_basis(config)) &&
+      !receive_n_data(fd, &check_checksum, sizeof(check_checksum))) {
     free(check_path);
     return NULL;
   }
@@ -642,6 +798,78 @@ File* receive_incremental_check(int fd, const Config* config, bool* skipped) {
     return NULL;
   }
 
+  /* ---- Alternate basis directories ---- */
+  if (config_has_basis(config)) {
+    BasisMatch basis;
+    basis_match_find(config, check_path, check_size, (time_t)check_mtime, (long)check_mtime_nsec,
+                     check_checksum, true, &basis);
+    if (basis.hit) {
+      if (basis.type == BASIS_DEST_COMPARE) {
+        /* compare-dest never copies: an exact match only suppresses the data
+           for a file the destination does not already hold (sparse backup).
+           When the destination holds a DIFFERENT version FastSync falls back to
+           a normal transfer rather than deleting the stale entry the way rsync
+           does (see RSYNC_COMPAT.md). */
+        basis_match_free(&basis);
+        if (!has_old_file) {
+          if (!send_status(fd, STATUS_OK)) {
+            close(old_fd);
+            free(full_path);
+            free(check_path);
+            return NULL;
+          }
+          free(old_data);
+          close(old_fd);
+          free(full_path);
+          free(check_path);
+          *skipped = true;
+          return NULL;
+        }
+      } else {
+        /* copy-dest / link-dest: materialize the unchanged file locally so the
+           sender can skip the data.  The store engine re-applies the normal
+           existing/ignore-existing/update/backup/delay-updates policy. */
+        File* materialized = file_create(check_path);
+        if (materialized && basis.content) {
+          materialized->data = basis.content;
+          basis.content = NULL;
+          materialized->metadata = file_metadata_create(&basis.st);
+          materialized->skip = true; /* receiver must not ack this as a data file */
+          if (basis.type == BASIS_DEST_LINK) {
+            materialized->basis_link = basis.basis_path;
+            basis.basis_path = NULL;
+          }
+          if (!materialized->metadata) {
+            file_destroy(materialized);
+            materialized = NULL;
+          }
+        } else {
+          file_destroy(materialized);
+          materialized = NULL;
+        }
+        if (materialized) {
+          if (!send_status(fd, STATUS_OK)) {
+            basis_match_free(&basis);
+            file_destroy(materialized);
+            close(old_fd);
+            free(full_path);
+            free(check_path);
+            return NULL;
+          }
+          basis_match_free(&basis);
+          free(old_data);
+          close(old_fd);
+          free(full_path);
+          free(check_path);
+          *skipped = false;
+          return materialized;
+        }
+        /* Materialization setup failed: fall through to the normal transfer. */
+      }
+    }
+    basis_match_free(&basis);
+  }
+
   if (try_delta && old_data != NULL) {
     bool delta_failed = false;
     File* delta_file =
@@ -792,26 +1020,27 @@ File* file_receive_directory(int file_descriptor) {
   return file;
 }
 
-int receive_manifest(int fd, const Config* config, int* next_status) {
-  if (!config) {
-    send_status(fd, STATUS_ERROR);
-    return -1;
-  }
-  int received_status = STATUS_ERROR;
-  int* status_out = next_status ? next_status : &received_status;
+/* Read a delete-manifest frame (the STATUS_MANIFEST leading code has already
+   been consumed): an entry count followed by that many destination-relative
+   paths.  The frame is self-delimiting (the count is authoritative), so the
+   caller decides what to do next and continues reading the following STATUS_*
+   frame.  Returns an owned ArrayList of validated path strings, or NULL after
+   sending STATUS_ERROR when the frame is malformed (bad count, empty/absolute
+   path, path traversal, or an aggregate size beyond MAX_MANIFEST_BYTES). */
+ArrayList* receive_manifest_entries(int fd) {
   int count;
   if (!receive_int(fd, &count)) {
     send_status(fd, STATUS_ERROR);
-    return -1;
+    return NULL;
   }
   if (count < 0 || count > MAX_MANIFEST_ENTRIES) {
     send_status(fd, STATUS_ERROR);
-    return -1;
+    return NULL;
   }
   ArrayList* manifest = array_list_create(free);
   if (!manifest) {
     send_status(fd, STATUS_ERROR);
-    return -1;
+    return NULL;
   }
   size_t manifest_bytes = 0;
   for (int i = 0; i < count; i++) {
@@ -823,32 +1052,51 @@ int receive_manifest(int fd, const Config* config, int* next_status) {
       free(s);
       array_list_delete(manifest);
       send_status(fd, STATUS_ERROR);
-      return -1;
+      return NULL;
     }
   }
-  if (!receive_status(fd, status_out)) {
-    array_list_delete(manifest);
-    send_status(fd, STATUS_ERROR);
-    return -1;
-  }
-  /* Deletion is a commit operation: never perform it until the sender has
-     completed the manifest frame successfully. */
-  if (*status_out != STATUS_FINISHED || !config->use_delete) {
-    array_list_delete(manifest);
-    if (*status_out != STATUS_FINISHED)
-      send_status(fd, STATUS_ERROR);
-    return *status_out == STATUS_FINISHED ? 0 : -1;
-  }
+  return manifest;
+}
+
+/* Remove every destination entry under the receive root that is not listed in
+   `manifest`, bounded by MAX_SERVER_DELETE_COUNT, 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.  Prints a notice and returns true on success. */
+bool manifest_delete_extras(const Config* config, ArrayList* manifest) {
+  if (!config || !manifest)
+    return false;
   fprintf(stderr, "Deleting files not in manifest...\n");
   /* With --delay-updates the staged (not yet published) files live directly
      under the receive root in the staging directory; the delete walker must
      not treat them as extras or it would remove every staged file before it
-     can be published. */
-  const char* skip_staging = config->delay_updates ? DELAY_UPDATES_STAGING_DIR : NULL;
+     can be published.  That staging name is protected only as a DIRECT child
+     of the receive root so a nested destination directory that happens to be
+     named .fastsync-stage is still ordinary content.  Alternate basis
+     directories (--compare-dest / --copy-dest / --link-dest) are excluded at
+     any depth: they are extra comparison snapshots the user pointed at, not
+     destination content, and deleting them would destroy the very files a
+     --link-dest run just linked into place. */
+  int skip_count = (config->delay_updates ? 1 : 0) + config->basis_count;
+  DeleteSkipEntry* skips = NULL;
+  if (skip_count > 0) {
+    skips = calloc((size_t)skip_count, sizeof(DeleteSkipEntry));
+    if (!skips)
+      return false;
+    int idx = 0;
+    if (config->delay_updates) {
+      skips[idx].prefix = DELAY_UPDATES_STAGING_DIR;
+      skips[idx].top_level_only = true;
+      idx++;
+    }
+    for (int i = 0; i < config->basis_count; i++) {
+      skips[idx].prefix = config->basis_dirs[i].path;
+      skips[idx].top_level_only = false;
+      idx++;
+    }
+  }
   bool deletion_ok = delete_extras_limited(config->receive_root_directory, manifest,
-                                           MAX_SERVER_DELETE_COUNT, skip_staging);
-  array_list_delete(manifest);
-  if (!deletion_ok)
-    send_status(fd, STATUS_ERROR);
-  return deletion_ok ? 0 : -1;
+                                           MAX_SERVER_DELETE_COUNT, skips, skip_count);
+  free(skips);
+  return deletion_ok;
 }

src/shared/file_receive.h

@@ -10,7 +10,14 @@
 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);
-int receive_manifest(int fd, const Config* config, int* next_status);
+/* Read a delete-manifest frame: entry count then paths (self-delimiting; the
+   leading STATUS_MANIFEST code has been consumed).  Returns an owned path
+   ArrayList, or NULL after signalling STATUS_ERROR on a malformed frame. */
+ArrayList* 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
+   run it based on the negotiated delete timing. */
+bool manifest_delete_extras(const Config* config, ArrayList* 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/file_types.h

@@ -29,6 +29,12 @@ typedef struct {
   /* True when this entry is an explicit directory entry (--dirs mode): the
    * receiver creates the directory instead of writing a regular file. */
   bool is_dir;
+  /* Receiver-only, --link-dest: when set, install the destination entry as a
+   * hard link to this absolute (root-confined) path instead of writing
+   * `data`.  The matching code has already verified the link target's content
+   * equals the incoming file, and `data` is kept as the cross-filesystem
+   * fallback (a local copy) if the hard link cannot be created. */
+  char* basis_link;
 } File;
 
 /* The path that should be sent on the wire and used for the receiver-side

src/shared/multiprocessing.c

@@ -27,6 +27,7 @@ PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* que
   context->loader_done = false;
   context->manifest = NULL;
   context->remove_source_files = NULL;
+  context->early_delete = false;
   context->total_files = 0;
   context->progress_bytes = 0;
   context->total_bytes = 0;
@@ -113,6 +114,7 @@ PipelineContextReceiver* pipeline_context_receiver_create(Config* config, Queue*
   context->receiver_done = false;
   context->queued_bytes = 0;
   context->max_queue_bytes = 0;
+  context->deferred_manifest = NULL;
   atomic_init(&context->cancelled, false);
   int init = 0;
   if (mtx_init(&context->mutex, mtx_plain) != thrd_success)
@@ -141,6 +143,8 @@ fail:
 
 void pipeline_context_receiver_destroy(PipelineContextReceiver* context) {
   config_delete(context->config);
+  if (context->deferred_manifest)
+    array_list_delete(context->deferred_manifest);
   queue_destroy(context->queue);
   receiver_outcomes_destroy(&context->outcomes);
   mtx_destroy(&context->mutex);
@@ -236,7 +240,8 @@ int receive_thread(void* pipeline_context) {
   mtx_unlock(&context->mutex);
 
   ReceiverSink sink = {receiver_enqueue_file, context, false, false, NULL};
-  if (receiver_process((Config*)config, file_descriptor, &sink) != 0) {
+  if (receiver_process_pending((Config*)config, file_descriptor, &sink,
+                               &context->deferred_manifest) != 0) {
     receiver_thread_fail(context);
     protocol_session_unbind();
     return thrd_error;
@@ -297,7 +302,7 @@ int write_thread(void* pipeline_context) {
     /* Record the per-file outcome so a --remove-source-files sender learns
        which sources were actually written versus skipped on the receiver.
        Explicit directory entries have no source and are never acknowledged. */
-    if (context->config->remove_source_files && !file->is_dir &&
+    if (context->config->remove_source_files && !file->is_dir && !file->skip &&
         !receiver_outcomes_append(&context->outcomes, (unsigned char)result)) {
       file_destroy(file);
       pipeline_context_receiver_note_bytes_released(context, file_bytes);

src/shared/multiprocessing.h

@@ -26,6 +26,11 @@ typedef struct {
   bool loader_done;
   ArrayList* manifest;
   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
+     a path-only pre-scan on the calling thread and the pipeline scanner must
+     not append to it.  Set once before the worker threads start. */
+  bool early_delete;
   mtx_t mutex_progress;
   int total_files;
   unsigned long long progress_bytes;
@@ -55,6 +60,14 @@ typedef struct PipelineContextReceiver {
      budget instead of growing without bound. */
   size_t queued_bytes;
   size_t max_queue_bytes;
+  /* Keep-set manifest for the commit-style (late) deletion
+     (--delete/--delete-after/--delete-delay).  receive_thread parses the whole
+     protocol stream but hands the manifest here instead of deleting while the
+     disk writer may still be draining; the caller (server.c) commits the
+     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;
 } PipelineContextReceiver;
 
 PipelineContextSender* pipeline_context_sender_create(Config* config, Queue* queue_scanner,

src/shared/protocol.c

@@ -302,15 +302,25 @@ bool protocol_send_n_data(ProtocolSession* session, const void* data, size_t dat
   return true;
 }
 
+bool protocol_receive_n_data_timed(ProtocolSession* session, void* data, size_t data_size,
+                                   int timeout_sec);
+
 bool protocol_receive_n_data(ProtocolSession* session, void* data, size_t data_size) {
+  return protocol_receive_n_data_timed(session, data, data_size, RECEIVE_TIMEOUT_SEC);
+}
+
+bool protocol_receive_n_data_timed(ProtocolSession* session, void* data, size_t data_size,
+                                   int timeout_sec) {
   log_debug_message(LOG_DEBUG_IO, "    Receiving n Data: %zu", data_size);
   if (!session)
     return false;
   int fd = session->read_fd;
+  if (timeout_sec <= 0)
+    timeout_sec = RECEIVE_TIMEOUT_SEC;
 
   struct timespec deadline;
   clock_gettime(CLOCK_MONOTONIC, &deadline);
-  deadline.tv_sec += RECEIVE_TIMEOUT_SEC;
+  deadline.tv_sec += timeout_sec;
 
   size_t total_bytes_received = 0;
   short wait_events = POLLIN;
@@ -319,7 +329,7 @@ bool protocol_receive_n_data(ProtocolSession* session, void* data, size_t data_s
       struct pollfd pfd = {.fd = fd, .events = wait_events};
       int poll_result = poll(&pfd, 1, deadline_remaining_ms(&deadline));
       if (poll_result == 0) {
-        log_message(LOG_LEVEL_ERROR, "Receive timeout after %ds", RECEIVE_TIMEOUT_SEC);
+        log_message(LOG_LEVEL_ERROR, "Receive timeout after %ds", timeout_sec);
         return false;
       }
       if (poll_result < 0) {
@@ -516,6 +526,17 @@ bool protocol_receive_status(ProtocolSession* session, Status* status) {
   return true;
 }
 
+/* protocol_receive_status with an explicit per-message deadline (seconds).
+   Used where a single reply may legitimately take far longer than the default
+   60 s receive window - e.g. the sender waiting for the early-delete ACK after
+   the receiver committed a large (up to MAX_SERVER_DELETE_COUNT) deletion. */
+bool protocol_receive_status_timed(ProtocolSession* session, Status* status, int timeout_sec) {
+  if (!protocol_receive_n_data_timed(session, status, sizeof(Status), timeout_sec))
+    return false;
+  log_debug_message(LOG_DEBUG_PROTO, "Received Status: %s", status_to_string(*status));
+  return true;
+}
+
 bool send_str(int fd, const char* data) {
   return protocol_send_str(legacy_session(-1, fd), data);
 }
@@ -543,3 +564,6 @@ bool send_status(int fd, Status status) {
 bool receive_status(int fd, Status* status) {
   return protocol_receive_status(legacy_session(fd, -1), status);
 }
+bool receive_status_timed(int fd, Status* status, int timeout_sec) {
+  return protocol_receive_status_timed(legacy_session(fd, -1), status, timeout_sec);
+}

src/shared/protocol.h

@@ -112,5 +112,10 @@ bool send_int(int file_descriptor, int data);
 bool receive_int(int file_descriptor, int* data);
 bool send_status(int file_descriptor, Status status);
 bool receive_status(int file_descriptor, Status* status);
+/* receive_status with an explicit per-message deadline in seconds, instead of
+   the default RECEIVE_TIMEOUT_SEC.  A reply that may legitimately take longer
+   (e.g. the early-delete ACK after a large receiver-side deletion) must use
+   this so the sender does not abort after the deletion already committed. */
+bool receive_status_timed(int file_descriptor, Status* status, int timeout_sec);
 
 #endif

src/shared/utils.c

@@ -204,9 +204,26 @@ static bool is_dir_in_manifest(const char* rel_path, ArrayList* manifest) {
   return false;
 }
 
+/* True when child_rel is, or lies below, a protected entry.  A prefix "a"
+   therefore protects "a" and "a/b/c" but not "ab".  Entries with top_level_only
+   set only protect DIRECT children of the receive root (at_root); nested
+   directories that share such a name stay ordinary destination content. */
+static bool path_under_skip_prefix(const char* child_rel, bool at_root,
+                                   const DeleteSkipEntry* skips, int skip_count) {
+  for (int i = 0; i < skip_count; i++) {
+    if (skips[i].top_level_only && !at_root)
+      continue;
+    size_t prefix_len = strlen(skips[i].prefix);
+    if (strncmp(child_rel, skips[i].prefix, prefix_len) == 0 &&
+        (child_rel[prefix_len] == '\0' || child_rel[prefix_len] == '/'))
+      return true;
+  }
+  return false;
+}
+
 static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifest,
-                             size_t max_delete, size_t* deleted_count,
-                             const char* skip_root_child) {
+                             size_t max_delete, size_t* deleted_count, const DeleteSkipEntry* skips,
+                             int skip_count) {
   int scanfd = dup(dirfd);
   if (scanfd < 0)
     return false;
@@ -220,18 +237,22 @@ static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes
   while ((entry = readdir(dir)) != NULL) {
     if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
       continue;
-    /* A --delay-updates run keeps its staging directory as a direct child of
-       the receive root.  Its contents are not manifest entries yet (they are
-       published after deletion), so descending into it would delete every
-       staged file as an "extra".  Skip only the top-level staging name; nested
-       directories with the same name are ordinary destination content. */
-    if (rel_path[0] == '\0' && skip_root_child && strcmp(entry->d_name, skip_root_child) == 0)
-      continue;
     char* child_rel = path_cat((char*)rel_path, entry->d_name);
     if (!child_rel) {
       operation_ok = false;
       continue;
     }
+    /* A --delay-updates run keeps its staging directory as a direct child of
+       the receive root, and basis-dir snapshots live below it too.  Their
+       contents are not manifest entries, so descending into them would delete
+       every staged / basis file as an "extra".  Only the staging name (a
+       top-level-only prefix) and the basis prefixes are protected: a nested
+       destination directory that happens to be called .fastsync-stage is
+       ordinary content. */
+    if (path_under_skip_prefix(child_rel, rel_path[0] == '\0', skips, skip_count)) {
+      free(child_rel);
+      continue;
+    }
     struct stat st;
     if (fstatat(dirfd, entry->d_name, &st, AT_SYMLINK_NOFOLLOW) != 0) {
       if (errno != ENOENT)
@@ -249,7 +270,7 @@ static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes
       bool child_removed = false;
       if (childfd >= 0) {
         child_removed = delete_extras_fd(childfd, child_rel, manifest, max_delete, deleted_count,
-                                         skip_root_child);
+                                         skips, skip_count);
         if (!child_removed)
           operation_ok = false;
         close(childfd);
@@ -301,7 +322,7 @@ static bool delete_extras_fd(int dirfd, const char* rel_path, ArrayList* manifes
 }
 
 bool delete_extras_limited(const char* dest_root, ArrayList* manifest, size_t max_delete,
-                           const char* skip_root_child) {
+                           const DeleteSkipEntry* skips, int skip_count) {
   if (!manifest)
     return false;
   int rootfd;
@@ -318,14 +339,14 @@ bool delete_extras_limited(const char* dest_root, ArrayList* manifest, size_t ma
   if (rootfd < 0)
     return false;
   size_t deleted_count = 0;
-  bool ok = delete_extras_fd(rootfd, "", manifest, max_delete, &deleted_count, skip_root_child);
+  bool ok = delete_extras_fd(rootfd, "", manifest, max_delete, &deleted_count, skips, skip_count);
   if (close(rootfd) != 0)
     ok = false;
   return ok;
 }
 
 bool delete_extras(const char* dest_root, ArrayList* manifest) {
-  return delete_extras_limited(dest_root, manifest, SIZE_MAX, NULL);
+  return delete_extras_limited(dest_root, manifest, SIZE_MAX, NULL, 0);
 }
 
 bool has_path_traversal(const char* path) {

src/shared/utils.h

@@ -10,12 +10,21 @@ 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);
-/* Remove files/dirs under dest_root that are not listed in manifest.  When
-   skip_root_child is non-NULL, a direct child of dest_root with that exact
-   name is left untouched (used to protect the --delay-updates staging
-   directory, which holds files that are still to be published). */
+/* 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
+   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). */
 bool delete_extras_limited(const char* dest_root, ArrayList* manifest, size_t max_delete,
-                           const char* skip_root_child);
+                           const DeleteSkipEntry* skips, int skip_count);
 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

@@ -147,7 +147,10 @@ class TestRemoveSourceFiles:
         with open(source_file, "wb") as f:
             f.write(b"keep after skip")
 
-        result, _ = run_client(source, dest, port=shared_server.port)
+        # The seed run preserves timestamps (-M) so the destination copy has the
+        # source's exact mtime; otherwise the incremental skip would depend on
+        # both writes landing in the same whole second (a race).
+        result, _ = run_client(source, dest, flags=["-M"], port=shared_server.port)
         assert result.returncode == 0
         result, _ = run_client(source, dest,
                                flags=["--remove-source-files", "--incremental"],
@@ -1971,3 +1974,508 @@ class TestFilters:
     """--filter/-C/-F rule layer: excludes prune, ordering is first-match-wins,
     the default with no matching rule is include, and legacy --exclude remains
     an independent layer."""
+
+
+class TestDeleteTiming:
+    """rsync deletion-timing family.  --delete-before/--delete-during transmit
+    the keep-set manifest BEFORE any file data (the receiver deletes extras and
+    acks first); --delete/--delete-after/--delete-delay commit deletions only
+    after the whole transfer succeeded.  Every timing flag implies --delete."""
+
+    def _seed(self, tag):
+        source = os.path.join(TEST_DATA_DIR, f"deltiming_{tag}_src")
+        clean_dir(source)
+        entries = {
+            "top.txt": b"top level\n",
+            "sub/deep.txt": b"deeply nested file\n",
+        }
+        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)
+        return source
+
+    @pytest.mark.parametrize("flag", ["--delete-before", "--delete-during", "--del",
+                                      "--delete-after", "--delete-delay"])
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_flag_removes_extras_on_success(self, flag, mt):
+        """Every timing flag is accepted, implies --delete, and on a successful
+        transfer removes the destination extras exactly like plain --delete."""
+        source = self._seed("ok")
+        dest = os.path.join(TEST_DATA_DIR, "deltiming_ok_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)
+            extra = os.path.join(received, "extra.txt")
+            with open(extra, "wb") as fh:
+                fh.write(b"should be deleted")
+
+            flags = [flag] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"{flag} sync failed: {(result.stderr or result.stdout)[:300]}"
+            assert not os.path.exists(extra), f"{flag} did not remove the extra file"
+            mismatches, missing = verify_transfer(source, received)
+            assert not missing, f"{flag} missing files: {missing}"
+            assert not mismatches, f"{flag} mismatched files: {mismatches}"
+
+    @pytest.mark.parametrize("flag", ["--delete-before", "--delete-during", "--del"])
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_early_flags_delete_before_data(self, flag, mt):
+        """--delete-before/--delete-during remove extras (and a file blocking a
+        destination directory) BEFORE data is applied, so a nested write that
+        would fail while the blocker still exists succeeds."""
+        source = self._seed("early")
+        dest = os.path.join(TEST_DATA_DIR, "deltiming_early_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)
+            extra = os.path.join(received, "extra.txt")
+            with open(extra, "wb") as fh:
+                fh.write(b"extra file")
+            blocker = os.path.join(received, "sub")
+            shutil.rmtree(blocker)
+            with open(blocker, "wb") as fh:
+                fh.write(b"blocks the nested destination directory")
+
+            flags = [flag] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode == 0, \
+                f"{flag} (early delete) did not remove the blocker in time: " \
+                f"{(result.stderr or result.stdout)[:300]}"
+            assert not os.path.exists(extra), f"{flag} did not delete the extra before data"
+            assert _read_file(os.path.join(received, "sub", "deep.txt")) == b"deeply nested file\n", \
+                f"{flag}: nested file was not written after the early deletion"
+
+    @pytest.mark.parametrize("flag", ["--delete", "--delete-after", "--delete-delay"])
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_late_flags_commit_only_after_success(self, flag, mt):
+        """Plain --delete/--delete-after/--delete-delay defer deletion until the
+        whole transfer succeeds: a mid-transfer write failure must leave every
+        extra in place (commit-style safety).  The -m receiver must also keep
+        the extras: the deferred keep-set is committed by the server only after
+        the disk-writer thread has finished, and a failing writer means the
+        manifest is freed, never applied."""
+        source = self._seed("late")
+        dest = os.path.join(TEST_DATA_DIR, "deltiming_late_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)
+            extra = os.path.join(received, "extra.txt")
+            with open(extra, "wb") as fh:
+                fh.write(b"extra file")
+            blocker = os.path.join(received, "sub")
+            shutil.rmtree(blocker)
+            with open(blocker, "wb") as fh:
+                fh.write(b"blocks the nested destination directory")
+
+            flags = [flag] + (["-m"] if mt else [])
+            result, _ = run_client(source, dest, flags=flags, port=server.port)
+            assert result.returncode != 0, \
+                f"{flag} (mt={mt}) unexpectedly succeeded (deletion must be deferred)"
+            assert os.path.exists(extra), \
+                f"{flag} (mt={mt}) removed an extra although the transfer failed"
+            assert os.path.isfile(blocker), \
+                f"{flag} (mt={mt}) deleted the blocker although the transfer failed"
+
+    def test_early_flag_respected_when_server_refuses_delete(self, shared_server):
+        """With an --allow-delete-less server the client's early timing still
+        completes (no deadlock on the pre-delete ack) and simply never deletes,
+        exactly like the plain server policy."""
+        source = self._seed("refused")
+        dest = os.path.join(TEST_DATA_DIR, "deltiming_refused_dst")
+        clean_dir(dest)
+        result, _ = run_client(source, dest, port=shared_server.port)
+        assert result.returncode == 0, f"seed sync failed: {result.stderr[:200]}"
+        received = get_dest_received_dir(dest, source)
+        extra = os.path.join(received, "extra.txt")
+        with open(extra, "wb") as fh:
+            fh.write(b"extra file")
+        result, _ = run_client(source, dest, flags=["--delete-before"], port=shared_server.port)
+        assert result.returncode == 0, \
+            f"--delete-before against a refuse-delete server failed: {result.stderr[:300]}"
+        assert os.path.exists(extra), "unauthorized delete removed an extra file"
+
+
+def _pin_mtime(path, ts):
+    os.utime(path, (ts, ts))
+
+
+class TestBasisDestDirs:
+    """--compare-dest / --copy-dest / --link-dest alternate basis directories.
+
+    FastSync's basis directories are relative to the destination root and are
+    confined below it.  The "unchanged" decision is receiver-side and requires
+    the per-file --incremental handshake (implied by these flags), so the basis
+    snapshot must reproduce the exact destination-relative mirror path of the
+    incoming files.
+    """
+
+    STAGING = ".fastsync-stage"
+    TS = 1577836800  # 2020-01-01 00:00:00 UTC, used to pin matching mtimes
+
+    # fixture files: source and basis share the mtime pin, so a basis "match"
+    # is decided purely by content (xxHash).  unchanged.txt is byte-identical;
+    # changed.txt is byte-DIFFERENT but has the SAME SIZE as the source (and
+    # the same pinned mtime), which is what forces the content-hash gate;
+    # added.txt does not exist in the basis at all.
+    UNCHANGED = "unchanged.txt"
+    CHANGED = "changed.txt"
+    ADDED = "added.txt"
+
+    def _make_source(self, name, source_files):
+        src = os.path.join(TEST_DATA_DIR, name)
+        clean_dir(src)
+        for rel, content in source_files.items():
+            full = os.path.join(src, rel)
+            os.makedirs(os.path.dirname(full), exist_ok=True)
+            with open(full, "wb") as fh:
+                fh.write(content)
+            _pin_mtime(full, self.TS)
+        return src
+
+    def _seed_basis_file(self, dest, source, basis_dir, rel, content, ts=None):
+        base = os.path.join(dest, basis_dir, os.path.relpath(
+            get_dest_received_dir(dest, source), dest))
+        full = os.path.join(base, rel)
+        os.makedirs(os.path.dirname(full), exist_ok=True)
+        with open(full, "wb") as fh:
+            fh.write(content)
+        _pin_mtime(full, self.TS if ts is None else ts)
+        return full
+
+    def _seed_basis(self, dest, source, basis_dir, basis_files):
+        for rel, content in basis_files.items():
+            self._seed_basis_file(dest, source, basis_dir, rel, content)
+        return os.path.join(dest, basis_dir, os.path.relpath(
+            get_dest_received_dir(dest, source), dest))
+
+    def _source_tree(self, prefix):
+        return {
+            self.UNCHANGED: b"stable content v1\n",
+            self.CHANGED: b"changed content now\n",
+            self.ADDED: b"brand new content\n",
+        }
+
+    def _basis_tree(self, prefix):
+        # unchanged.txt is identical to the source; changed.txt has the SAME
+        # byte size and pinned mtime but a different body (equal size forces
+        # the xxHash gate); added.txt is missing from the basis.
+        return {
+            self.UNCHANGED: b"stable content v1\n",
+            self.CHANGED: b"CHANGED CONTENT NOW\n",
+        }
+
+    def test_same_size_different_content_is_not_a_basis_match(self, shared_server):
+        # Core safety property: equal size + pinned mtime but different content
+        # must NEVER be hard-linked or copied from the basis -- the xxHash gate
+        # rejects it and the sender's data is transferred instead.
+        for flag, basis_dir in (("--link-dest", "szlb"), ("--copy-dest", "szcp"),
+                                ("--compare-dest", "szcmp")):
+            source = self._make_source("basis_same_size_src",
+                                       {self.UNCHANGED: b"same length body\n"})
+            dest = os.path.join(TEST_DATA_DIR, f"basis_same_size_dst_{basis_dir}")
+            clean_dir(dest)
+            basis_file = self._seed_basis_file(dest, source, basis_dir, self.UNCHANGED,
+                                               b"SAME LENGTH BODY!")
+            result, _ = run_client(source, dest, flags=[f"{flag}={basis_dir}"],
+                                   port=shared_server.port)
+            assert result.returncode == 0, \
+                f"{flag} same-size mismatch failed: {result.stderr[:300]}"
+            received = get_dest_received_dir(dest, source)
+            dest_file = os.path.join(received, self.UNCHANGED)
+            assert _read_file(dest_file) == b"same length body\n", \
+                f"{flag}: basis content leaked into the destination on a hash mismatch"
+            if flag != "--compare-dest":
+                assert os.stat(dest_file).st_ino != os.stat(basis_file).st_ino, \
+                    f"{flag}: linked/copied from a content-mismatched basis file"
+
+    def test_compare_dest_skips_matching_and_transfers_missing(self, shared_server):
+        source = self._make_source("basis_compare_src", self._source_tree("c"))
+        dest = os.path.join(TEST_DATA_DIR, "basis_compare_dst")
+        clean_dir(dest)
+        self._seed_basis(dest, source, "cbasis", self._basis_tree("c"))
+        result, _ = run_client(source, dest,
+                               flags=["--compare-dest=cbasis"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"compare-dest failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        # compare-dest never copies: an exact basis match is skipped, leaving a
+        # sparse destination (rsync parity).
+        assert not os.path.exists(os.path.join(received, self.UNCHANGED)), \
+            "compare-dest materialized the unchanged file"
+        # Files the destination lacks AND the basis cannot satisfy are still
+        # transferred normally.
+        assert _read_file(os.path.join(received, self.CHANGED)) == \
+            self._source_tree("c")[self.CHANGED], "changed file not transferred"
+        assert _read_file(os.path.join(received, self.ADDED)) == \
+            self._source_tree("c")[self.ADDED], "added file not transferred"
+
+    def test_compare_dest_content_mismatch_forces_transfer(self, shared_server):
+        # The basis holds a file with a DIFFERENT body: even though it shares
+        # the mtime pin, the xxHash check fails and the data must be sent.
+        source = self._make_source("basis_compare_mismatch_src", {self.UNCHANGED: b"real data\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_compare_mismatch_dst")
+        clean_dir(dest)
+        basis = self._seed_basis(dest, source, "cbasis", {self.UNCHANGED: b"stale data!!\n"})
+        result, _ = run_client(source, dest, flags=["--compare-dest=cbasis"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"compare-dest mismatch failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert _read_file(os.path.join(received, self.UNCHANGED)) == b"real data\n", \
+            "content mismatch did not fall back to a normal transfer"
+        assert os.stat(os.path.join(received, self.UNCHANGED)).st_ino != \
+            os.stat(os.path.join(basis, self.UNCHANGED)).st_ino
+
+    def test_copy_dest_copies_unchanged_and_transfers_changed(self, shared_server):
+        source = self._make_source("basis_copy_src", self._source_tree("cp"))
+        dest = os.path.join(TEST_DATA_DIR, "basis_copy_dst")
+        clean_dir(dest)
+        basis = self._seed_basis(dest, source, "cpbasis", self._basis_tree("cp"))
+        result, _ = run_client(source, dest, flags=["--copy-dest=cpbasis"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"copy-dest failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        unchanged = os.path.join(received, self.UNCHANGED)
+        assert _read_file(unchanged) == b"stable content v1\n", "unchanged file not materialized"
+        # A real local copy, NOT a hard link to the basis file.
+        assert os.stat(unchanged).st_ino != os.stat(os.path.join(basis, self.UNCHANGED)).st_ino
+        # Equal-size/different-content basis file falls back to the sender data.
+        assert _read_file(os.path.join(received, self.CHANGED)) == \
+            self._source_tree("cp")[self.CHANGED]
+        assert _read_file(os.path.join(received, self.ADDED)) == \
+            self._source_tree("cp")[self.ADDED]
+
+    def test_link_dest_hardlinks_and_falls_back(self, shared_server):
+        source = self._make_source("basis_link_src", self._source_tree("ln"))
+        dest = os.path.join(TEST_DATA_DIR, "basis_link_dst")
+        clean_dir(dest)
+        basis = self._seed_basis(dest, source, "lnbasis", self._basis_tree("ln"))
+        result, _ = run_client(source, dest, flags=["--link-dest=lnbasis"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"link-dest failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        unchanged = os.path.join(received, self.UNCHANGED)
+        basis_file = os.path.join(basis, self.UNCHANGED)
+        # Real hard link: same inode as the DIR file, nlink >= 2, no data copy.
+        assert os.path.exists(unchanged)
+        assert os.stat(unchanged).st_ino == os.stat(basis_file).st_ino, \
+            "link-dest did not produce a hard link"
+        assert os.stat(unchanged).st_nlink >= 2
+        # Equal-size/different-content basis file must fall back to a plain
+        # transfer (not a link).
+        changed = os.path.join(received, self.CHANGED)
+        assert _read_file(changed) == self._source_tree("ln")[self.CHANGED]
+        assert os.stat(changed).st_ino != os.stat(os.path.join(basis, self.CHANGED)).st_ino
+
+    @pytest.mark.parametrize("flag", ["--compare-dest", "--copy-dest", "--link-dest"])
+    def test_basis_dir_missing_is_a_clean_noop(self, shared_server, flag):
+        # A basis directory that does not exist must simply transfer everything.
+        source = self._make_source("basis_missing_src", {self.UNCHANGED: b"content\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_missing_dst")
+        clean_dir(dest)
+        result, _ = run_client(source, dest, flags=[f"{flag}=nope"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"{flag} with missing dir failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert _read_file(os.path.join(received, self.UNCHANGED)) == b"content\n"
+
+    def test_link_dest_multithreaded(self, shared_server):
+        source = self._make_source("basis_link_mt_src", self._source_tree("mt"))
+        dest = os.path.join(TEST_DATA_DIR, "basis_link_mt_dst")
+        clean_dir(dest)
+        basis = self._seed_basis(dest, source, "mtbasis", self._basis_tree("mt"))
+        result, _ = run_client(source, dest, flags=["--link-dest=mtbasis", "-m"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"-m link-dest failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert os.stat(os.path.join(received, self.UNCHANGED)).st_ino == \
+            os.stat(os.path.join(basis, self.UNCHANGED)).st_ino
+        assert _read_file(os.path.join(received, self.ADDED)) == \
+            self._source_tree("mt")[self.ADDED]
+
+    def test_link_dest_with_delay_updates_stages_and_publishes_link(self, shared_server):
+        source = self._make_source("basis_link_delay_src", {self.UNCHANGED: b"v1\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_link_delay_dst")
+        clean_dir(dest)
+        basis = self._seed_basis(dest, source, "delaybasis", {self.UNCHANGED: b"v1\n"})
+        result, _ = run_client(source, dest,
+                               flags=["--link-dest=delaybasis", "--delay-updates"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"delay-updates link-dest failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        unchanged = os.path.join(received, self.UNCHANGED)
+        assert os.stat(unchanged).st_ino == \
+            os.stat(os.path.join(basis, self.UNCHANGED)).st_ino
+        assert not os.path.isdir(os.path.join(dest, self.STAGING)), \
+            "delay-updates staging tree was not cleaned up"
+
+    def test_delete_does_not_touch_basis_dir(self):
+        """--delete removes genuine extras but must never treat a basis-dir
+        snapshot (which a --link-dest run just linked from) as destination
+        content."""
+        source = self._make_source("basis_delete_src", {self.UNCHANGED: b"v1\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_delete_dst")
+        clean_dir(dest)
+        basis = self._seed_basis(dest, source, "delbasis", {self.UNCHANGED: b"v1\n"})
+        received = get_dest_received_dir(dest, source)
+        os.makedirs(received, exist_ok=True)
+        extra = os.path.join(received, "extra.txt")
+        with open(extra, "wb") as fh:
+            fh.write(b"extra")
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, flags=["--link-dest=delbasis", "--delete"],
+                                   port=server.port)
+            assert result.returncode == 0, \
+                f"delete+link-dest failed: {result.stderr[:300]}"
+            assert not os.path.exists(extra), "genuine extra file was not deleted"
+            assert _read_file(os.path.join(received, self.UNCHANGED)) == b"v1\n"
+            assert os.path.exists(os.path.join(basis, self.UNCHANGED)), \
+                "basis directory was deleted by --delete"
+            assert os.stat(os.path.join(received, self.UNCHANGED)).st_ino == \
+                os.stat(os.path.join(basis, self.UNCHANGED)).st_ino
+
+    def test_delay_delete_keeps_nested_staging_named_dir_as_content(self):
+        # The real --delay-updates staging directory is protected from --delete
+        # only as a DIRECT child of the receive root.  A nested destination
+        # directory that merely shares the staging name is ordinary content, so
+        # its extras must still be deleted (regression guard for the walker).
+        source = self._make_source("basis_nested_stage_src",
+                                   {"top.txt": b"top\n", "sub/real.txt": b"real\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_nested_stage_dst")
+        clean_dir(dest)
+        self._seed_basis(dest, source, "nstbasis",
+                         {"top.txt": b"top\n", "sub/real.txt": b"real\n"})
+        received = get_dest_received_dir(dest, source)
+        nested = os.path.join(received, "sub", self.STAGING)
+        os.makedirs(nested, exist_ok=True)
+        extra = os.path.join(nested, "extra.txt")
+        with open(extra, "wb") as fh:
+            fh.write(b"nested extra")
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest,
+                                   flags=["--link-dest=nstbasis", "--delete",
+                                          "--delay-updates"],
+                                   port=server.port)
+            assert result.returncode == 0, \
+                f"delay-delete nested staging failed: {result.stderr[:300]}"
+            assert not os.path.exists(extra), \
+                "extra inside a nested .fastsync-stage dir was not deleted"
+            assert not os.path.isdir(nested), \
+                "nested .fastsync-stage dir should have been removed after its extra"
+            assert _read_file(os.path.join(received, "sub", "real.txt")) == b"real\n"
+            assert not os.path.isdir(os.path.join(dest, self.STAGING)), \
+                "real delay-updates staging tree was not cleaned up"
+
+    def test_basis_priority_first_match_wins(self, shared_server):
+        # Two link-dest dirs both hold the exact file: the FIRST (command-line
+        # order) basis directory must win and supply the hard link.
+        source = self._make_source("basis_prio_src", {"f.txt": b"content\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_prio_dst")
+        clean_dir(dest)
+        first = self._seed_basis_file(dest, source, "b1", "f.txt", b"content\n")
+        self._seed_basis_file(dest, source, "b2", "f.txt", b"content\n")
+        result, _ = run_client(source, dest, flags=["--link-dest=b1", "--link-dest=b2"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"link-dest priority failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert os.stat(os.path.join(received, "f.txt")).st_ino == os.stat(first).st_ino, \
+            "first basis dir did not win over the second"
+
+    def test_basis_priority_across_compare_and_link(self, shared_server):
+        # A compare-dest entry listed BEFORE a link-dest entry shadows it (the
+        # exact match is found first and nothing is materialized); reversing the
+        # order lets the link-dest entry win and materialize a hard link.
+        source = self._make_source("basis_prio_mixed_src", {"f.txt": b"content\n"})
+
+        dest = os.path.join(TEST_DATA_DIR, "basis_prio_mixed_dst")
+        clean_dir(dest)
+        self._seed_basis_file(dest, source, "cmpb", "f.txt", b"content\n")
+        self._seed_basis_file(dest, source, "lnb", "f.txt", b"content\n")
+        result, _ = run_client(source, dest,
+                               flags=["--compare-dest=cmpb", "--link-dest=lnb"],
+                               port=shared_server.port)
+        assert result.returncode == 0, \
+            f"mixed priority (compare first) failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert not os.path.exists(os.path.join(received, "f.txt")), \
+            "compare-dest matched first, so the file must stay sparse (no link-dest materialize)"
+
+        dest = os.path.join(TEST_DATA_DIR, "basis_prio_mixed_dst2")
+        clean_dir(dest)
+        self._seed_basis_file(dest, source, "cmpb", "f.txt", b"content\n")
+        linkb2 = self._seed_basis_file(dest, source, "lnb", "f.txt", b"content\n")
+        result, _ = run_client(source, dest,
+                               flags=["--link-dest=lnb", "--compare-dest=cmpb"],
+                               port=shared_server.port)
+        assert result.returncode == 0, \
+            f"mixed priority (link first) failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert os.stat(os.path.join(received, "f.txt")).st_ino == os.stat(linkb2).st_ino, \
+            "link-dest did not materialize when listed before compare-dest"
+
+    def test_link_dest_size_only_ignores_mtime(self, shared_server):
+        # --size-only drops the mtime leg of the quick check: a basis file with
+        # the SAME content but a DIFFERENT mtime is still an exact match.
+        source = self._make_source("basis_sizeonly_src", {"f.txt": b"content\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_sizeonly_dst")
+        clean_dir(dest)
+        basis_file = self._seed_basis_file(dest, source, "sob", "f.txt", b"content\n",
+                                           ts=self.TS + 500)
+        result, _ = run_client(source, dest, flags=["--link-dest=sob", "--size-only"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"size-only link-dest failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert os.stat(os.path.join(received, "f.txt")).st_ino == os.stat(basis_file).st_ino, \
+            "--size-only should link a basis file whose mtime differs"
+
+    def test_link_dest_ignore_times_never_links(self, shared_server):
+        # -I/--ignore-times forces every file to be updated, so a basis dir is
+        # never used to hard-link (rsync parity).  The file is transferred and
+        # stored as a fresh inode even though it matches the basis exactly.
+        source = self._make_source("basis_igntimes_src", {"f.txt": b"content\n"})
+        dest = os.path.join(TEST_DATA_DIR, "basis_igntimes_dst")
+        clean_dir(dest)
+        basis_file = self._seed_basis_file(dest, source, "itb", "f.txt", b"content\n")
+        result, _ = run_client(source, dest, flags=["--link-dest=itb", "--ignore-times"],
+                               port=shared_server.port)
+        assert result.returncode == 0, f"ignore-times link-dest failed: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        dest_file = os.path.join(received, "f.txt")
+        assert _read_file(dest_file) == b"content\n"
+        assert os.stat(dest_file).st_ino != os.stat(basis_file).st_ino, \
+            "--ignore-times must not hard-link to a basis file"
+
+    def test_basis_refuses_file_above_whole_file_limit(self, shared_server):
+        # Every whole-file payload path in FastSync (basis dirs included) is
+        # bounded by MAX_RECEIVE_WHOLE_FILE_SIZE.  rsync supports basis dirs for
+        # arbitrary sizes; FastSync refuses such a run up front with a clear
+        # diagnostic instead of letting the receiver abort the whole transfer
+        # mid-stream with no client-side explanation.
+        source = self._make_source("basis_oversize_src", {"small.txt": b"ok\n"})
+        big = os.path.join(source, "huge.bin")
+        with open(big, "wb") as fh:
+            os.ftruncate(fh.fileno(), 256 * 1024 * 1024 + 4096)
+        dest = os.path.join(TEST_DATA_DIR, "basis_oversize_dst")
+        clean_dir(dest)
+        result, _ = run_client(source, dest, flags=["--link-dest=nope"],
+                               port=shared_server.port)
+        assert result.returncode != 0, \
+            "basis run with an over-limit file unexpectedly succeeded"
+        assert "larger than" in result.stderr, \
+            f"no clear over-limit diagnostic: {result.stderr[:300]}"
+        received = get_dest_received_dir(dest, source)
+        assert not os.path.exists(received), \
+            "over-limit basis run transferred files before failing"

tests/test_client_cli.c

@@ -595,8 +595,89 @@ static void test_parse_args_relative_no_implied_mkpath() {
   config_delete(cfg);
 }
 
-/* --del is recognized as the rsync alias, but its timing mode is not implemented. */
-static void test_parse_args_delete_during_alias_unimplemented() {
+/* Parse --compare-dest/--copy-dest/--link-dest, including the =value and
+   separate-argument forms, and verify the ordered (repeatable) basis list. */
+static void test_parse_args_basis_dirs() {
+  Config* cfg = config_create();
+  int positional_args[2];
+  int positional_count = 0;
+  char* argv[] = {"fastsync", "--link-dest=prior", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(config_has_basis(cfg));
+  EXPECT_EQ_INT(cfg->basis_count, 1);
+  EXPECT_EQ_INT(cfg->basis_dirs[0].type, BASIS_DEST_LINK);
+  EXPECT_EQ_STR(cfg->basis_dirs[0].path, "prior");
+  /* Basis dirs are honored by the receiver-side per-file check, so they imply
+     --incremental (and, unless disabled, metadata) on the sender. */
+  EXPECT_TRUE(cfg->use_incremental);
+  EXPECT_TRUE(cfg->use_metadata);
+  config_delete(cfg);
+
+  cfg = config_create();
+  positional_count = 0;
+  char* argv2[] = {"fastsync", "--compare-dest", "cmp", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv2, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->basis_count, 1);
+  EXPECT_EQ_INT(cfg->basis_dirs[0].type, BASIS_DEST_COMPARE);
+  EXPECT_EQ_STR(cfg->basis_dirs[0].path, "cmp");
+  config_delete(cfg);
+
+  /* Repetition is supported: entries keep command-line order and type. */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv3[] = {"fastsync",      "--link-dest=a", "--compare-dest=b",
+                   "--link-dest=c", "--copy-dest=d", "/src",
+                   "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 7, argv3, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->basis_count, 4);
+  EXPECT_EQ_INT(cfg->basis_dirs[0].type, BASIS_DEST_LINK);
+  EXPECT_EQ_STR(cfg->basis_dirs[0].path, "a");
+  EXPECT_EQ_INT(cfg->basis_dirs[1].type, BASIS_DEST_COMPARE);
+  EXPECT_EQ_STR(cfg->basis_dirs[1].path, "b");
+  EXPECT_EQ_INT(cfg->basis_dirs[2].type, BASIS_DEST_LINK);
+  EXPECT_EQ_STR(cfg->basis_dirs[2].path, "c");
+  EXPECT_EQ_INT(cfg->basis_dirs[3].type, BASIS_DEST_COPY);
+  EXPECT_EQ_STR(cfg->basis_dirs[3].path, "d");
+  config_delete(cfg);
+
+  /* Nested relative basis dirs are allowed (they resolve below the root). */
+  cfg = config_create();
+  positional_count = 0;
+  char* argv4[] = {"fastsync", "--copy-dest=snap/2026-01", "/src", "/dst"};
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv4, positional_args, &positional_count), 0);
+  EXPECT_EQ_INT(cfg->basis_count, 1);
+  EXPECT_EQ_STR(cfg->basis_dirs[0].path, "snap/2026-01");
+  config_delete(cfg);
+}
+
+/* Absolute, escaping, or degenerate basis-dir values must be rejected up
+   front: they would resolve outside the destination root on the receiver. */
+static void test_parse_args_basis_invalid_paths() {
+  static const char* const invalid[] = {"/abs", "..", "a/../b", "."};
+  for (size_t i = 0; i < sizeof(invalid) / sizeof(invalid[0]); i++) {
+    Config* cfg = config_create();
+    char* argv[] = {"fastsync", "--link-dest", (char*)invalid[i], "/src", "/dst"};
+    int positional_args[2];
+    int positional_count = 0;
+    EXPECT_EQ_INT(parse_args(cfg, 5, argv, positional_args, &positional_count), -1);
+    config_delete(cfg);
+  }
+}
+
+/* Basis dirs require the per-file incremental handshake, which -s disables. */
+static void test_validate_config_basis_rejects_chunk_serialization() {
+  Config* cfg = valid_client_config();
+  EXPECT_EQ_INT(config_basis_append(cfg, BASIS_DEST_LINK, "prior"), 0);
+  cfg->use_chunk_serialization = true;
+  EXPECT_FALSE(validate_config(cfg));
+  cfg->use_chunk_serialization = false;
+  EXPECT_TRUE(validate_config(cfg));
+  config_delete(cfg);
+}
+
+/* --del is accepted as the rsync alias for --delete-during: it enables
+ * deletion with the during (early) timing. */
+static void test_parse_args_delete_during_alias() {
   static const char* const options[] = {"--del", "--delete-during"};
 
   for (size_t i = 0; i < sizeof(options) / sizeof(options[0]); i++) {
@@ -605,12 +686,81 @@ static void test_parse_args_delete_during_alias_unimplemented() {
     int positional_args[2];
     int positional_count = 0;
 
-    EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), -1);
-    EXPECT_FALSE(cfg->use_delete);
+    EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+    EXPECT_TRUE(cfg->use_delete);
+    EXPECT_TRUE(cfg->delete_during);
+    EXPECT_FALSE(cfg->delete_before);
+    EXPECT_FALSE(cfg->delete_delay);
+    EXPECT_FALSE(cfg->delete_after);
     config_delete(cfg);
   }
 }
 
+/* Each rsync deletion-timing flag is accepted and implies --delete. */
+static void test_parse_args_delete_timing_flags() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--delete-before", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_delete);
+  EXPECT_TRUE(cfg->delete_before);
+  config_delete(cfg);
+
+  cfg = config_create();
+  char* argv_after[] = {"fastsync", "--delete-after", "/src", "/dst"};
+  positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv_after, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_delete);
+  EXPECT_TRUE(cfg->delete_after);
+  EXPECT_FALSE(cfg->delete_before);
+  config_delete(cfg);
+
+  cfg = config_create();
+  char* argv_delay[] = {"fastsync", "--delete-delay", "/src", "/dst"};
+  positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 4, argv_delay, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_delete);
+  EXPECT_TRUE(cfg->delete_delay);
+  EXPECT_FALSE(cfg->delete_before);
+  EXPECT_FALSE(cfg->delete_after);
+  config_delete(cfg);
+}
+
+/* Two different delete-timing flags on one command line are a conflict, not a
+ * silent last-one-wins choice. */
+static void test_parse_args_delete_timing_conflict_rejected() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--delete-before", "--delete-after", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv, positional_args, &positional_count), 0);
+  EXPECT_TRUE(cfg->use_delete);
+  EXPECT_FALSE(validate_config(cfg));
+  config_delete(cfg);
+
+  cfg = config_create();
+  char* argv2[] = {"fastsync", "--delete-during", "--delete-delay", "/src", "/dst"};
+  positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv2, positional_args, &positional_count), 0);
+  EXPECT_FALSE(validate_config(cfg));
+  config_delete(cfg);
+}
+
+/* A timing flag whose --delete was then negated away must be rejected: timing
+ * without deletion is meaningless. */
+static void test_parse_args_delete_timing_without_delete_rejected() {
+  Config* cfg = config_create();
+  char* argv[] = {"fastsync", "--delete-before", "--no-delete", "/src", "/dst"};
+  int positional_args[2];
+  int positional_count = 0;
+  EXPECT_EQ_INT(parse_args(cfg, 5, argv, positional_args, &positional_count), 0);
+  EXPECT_FALSE(cfg->use_delete);
+  EXPECT_TRUE(cfg->delete_before);
+  EXPECT_FALSE(validate_config(cfg));
+  config_delete(cfg);
+}
+
 /* Parsed-but-unimplemented options must fail instead of being silently accepted. */
 static void test_parse_args_rejects_unimplemented_options() {
   static const char* const options[] = {"--silent",
@@ -626,16 +776,11 @@ static void test_parse_args_rejects_unimplemented_options() {
                                         "--append",
                                         "--append-verify",
                                         "--delete-excluded",
-                                        "--delete-after",
                                         "--max-delete",
                                         "--prune-empty-dirs",
                                         "-e",
                                         "--rsh",
                                         "--rsync-path",
-                                        "--compare-dest",
-                                        "--copy-dest",
-                                        "--link-dest",
-                                        "--delete-before",
                                         "--address",
                                         "--bind-address",
                                         "--ipv6",
@@ -1542,7 +1687,10 @@ void test_client_cli() {
   test_parse_args_unknown_option();
   test_parse_args_dirs_aliases();
   test_parse_args_relative_no_implied_mkpath();
-  test_parse_args_delete_during_alias_unimplemented();
+  test_parse_args_delete_during_alias();
+  test_parse_args_delete_timing_flags();
+  test_parse_args_delete_timing_conflict_rejected();
+  test_parse_args_delete_timing_without_delete_rejected();
   test_parse_args_rejects_unimplemented_options();
   test_parse_args_quiet();
   test_parse_args_human_readable();
@@ -1589,4 +1737,7 @@ void test_client_cli() {
   test_parse_args_files_from();
   test_parse_args_filter_rules();
   test_parse_args_from0_cvs_filter_file_flags();
+  test_parse_args_basis_dirs();
+  test_parse_args_basis_invalid_paths();
+  test_validate_config_basis_rejects_chunk_serialization();
 }

tests/test_config.c

@@ -438,6 +438,242 @@ static void test_config_delay_updates_reserved_backup_rejected() {
   config_delete(c);
 }
 
+static void test_config_delete_timing_early_helper() {
+  Config* cfg = config_create();
+  EXPECT_NOT_NULL(cfg);
+  EXPECT_FALSE(config_delete_timing_early(cfg));
+  EXPECT_TRUE(config_has_valid_delete_timing(cfg));
+  cfg->use_delete = true;
+  EXPECT_TRUE(config_has_valid_delete_timing(cfg));
+  EXPECT_FALSE(config_delete_timing_early(cfg));
+  config_delete(cfg);
+
+  cfg = config_create();
+  cfg->use_delete = true;
+  cfg->delete_before = true;
+  EXPECT_TRUE(config_delete_timing_early(cfg));
+  EXPECT_TRUE(config_has_valid_delete_timing(cfg));
+  config_delete(cfg);
+
+  cfg = config_create();
+  cfg->use_delete = true;
+  cfg->delete_during = true;
+  EXPECT_TRUE(config_delete_timing_early(cfg));
+  EXPECT_TRUE(config_has_valid_delete_timing(cfg));
+  config_delete(cfg);
+
+  cfg = config_create();
+  cfg->use_delete = true;
+  cfg->delete_delay = true;
+  EXPECT_FALSE(config_delete_timing_early(cfg));
+  EXPECT_TRUE(config_has_valid_delete_timing(cfg));
+  config_delete(cfg);
+
+  cfg = config_create();
+  cfg->use_delete = true;
+  cfg->delete_after = true;
+  EXPECT_FALSE(config_delete_timing_early(cfg));
+  EXPECT_TRUE(config_has_valid_delete_timing(cfg));
+  config_delete(cfg);
+
+  /* Two simultaneous timings are invalid. */
+  cfg = config_create();
+  cfg->use_delete = true;
+  cfg->delete_before = true;
+  cfg->delete_after = true;
+  EXPECT_TRUE(config_delete_timing_early(cfg));
+  EXPECT_FALSE(config_has_valid_delete_timing(cfg));
+  config_delete(cfg);
+
+  /* A timing flag without deletion is invalid. */
+  cfg = config_create();
+  cfg->delete_delay = true;
+  EXPECT_FALSE(config_has_valid_delete_timing(cfg));
+  EXPECT_FALSE(config_delete_timing_early(cfg));
+  config_delete(cfg);
+}
+
+/* New delete-timing fields must survive config_send/config_receive unchanged,
+   and a config carrying two conflicting timings must be rejected. */
+static void test_config_delete_timing_wire_roundtrip() {
+  if (is_running_under_valgrind())
+    return;
+
+  struct {
+    bool before, during, delay, after;
+  } cases[] = {
+      {false, false, false, false}, {true, false, false, false}, {false, true, false, false},
+      {false, false, true, false},  {false, false, false, true},
+  };
+  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->use_delete && recv->delete_before == cases[i].before &&
+             recv->delete_during == cases[i].during && recv->delete_delay == cases[i].delay &&
+             recv->delete_after == cases[i].after;
+      }
+      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->use_delete = true;
+      send_cfg->delete_before = cases[i].before;
+      send_cfg->delete_during = cases[i].during;
+      send_cfg->delete_delay = cases[i].delay;
+      send_cfg->delete_after = cases[i].after;
+      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);
+    }
+  }
+}
+
+/* The receiver-side wire validation rejects a keep-set config with two
+   conflicting delete-timing flags. */
+static void test_config_delete_timing_conflict_rejected() {
+  if (is_running_under_valgrind())
+    return;
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->use_delete = true;
+  c->delete_before = true;
+  c->delete_delay = true;
+  EXPECT_FALSE(roundtrip_config_ok(c));
+  config_delete(c);
+}
+
+/* Basis-dir lists survive the config wire: each entry's type and path must
+   round-trip unchanged. */
+static void test_config_basis_roundtrip() {
+  if (is_running_under_valgrind())
+    return;
+  Config* send_cfg = config_create();
+  EXPECT_NOT_NULL(send_cfg);
+  send_cfg->send_directory = str_dup("/send/src");
+  send_cfg->receive_root_directory = str_dup("/send/dst");
+  EXPECT_EQ_INT(config_basis_append(send_cfg, BASIS_DEST_LINK, "prior"), 0);
+  EXPECT_EQ_INT(config_basis_append(send_cfg, BASIS_DEST_COMPARE, "snap/2026-01"), 0);
+  EXPECT_EQ_INT(config_basis_append(send_cfg, BASIS_DEST_COPY, "copy"), 0);
+
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(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 && recv->basis_count == 3 && recv->basis_dirs != NULL;
+    if (ok) {
+      ok = recv->basis_dirs[0].type == BASIS_DEST_LINK &&
+           strcmp(recv->basis_dirs[0].path, "prior") == 0;
+      ok = ok && recv->basis_dirs[1].type == BASIS_DEST_COMPARE &&
+           strcmp(recv->basis_dirs[1].path, "snap/2026-01") == 0;
+      ok = ok && recv->basis_dirs[2].type == BASIS_DEST_COPY &&
+           strcmp(recv->basis_dirs[2].path, "copy") == 0;
+    }
+    config_delete(recv);
+    close(p[0]);
+    close(p[1]);
+    _exit(ok ? 0 : 1);
+  } else {
+    close(p[0]);
+    io_set_fds(p[1], p[1]);
+    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);
+  }
+}
+
+/* The receiver must reject a basis-dir path that would escape the destination
+   root.  The values are injected directly (bypassing the client-side append
+   validator) so the receiver-side wire validation is what is exercised. */
+static void test_config_basis_wire_rejects_escaping() {
+  if (is_running_under_valgrind())
+    return;
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->basis_count = 1;
+  c->basis_dirs = calloc(1, sizeof(BasisDest));
+  c->basis_dirs[0].type = BASIS_DEST_LINK;
+  c->basis_dirs[0].path = str_dup("../../etc");
+  EXPECT_FALSE(roundtrip_config_ok(c));
+  config_delete(c);
+
+  c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->basis_count = 1;
+  c->basis_dirs = calloc(1, sizeof(BasisDest));
+  c->basis_dirs[0].type = BASIS_DEST_LINK;
+  c->basis_dirs[0].path = str_dup("/abs");
+  EXPECT_FALSE(roundtrip_config_ok(c));
+  config_delete(c);
+
+  /* A well-formed list still round-trips even with a manually built struct. */
+  c = config_create();
+  EXPECT_NOT_NULL(c);
+  c->send_directory = str_dup("/src");
+  c->receive_root_directory = str_dup("/dst");
+  c->basis_count = 1;
+  c->basis_dirs = calloc(1, sizeof(BasisDest));
+  c->basis_dirs[0].type = BASIS_DEST_COPY;
+  c->basis_dirs[0].path = str_dup("safe");
+  EXPECT_TRUE(roundtrip_config_ok(c));
+  config_delete(c);
+}
+
+/* Basis-dir paths are canonicalized on the way in: trailing slashes and
+   interior empty / "." components are dropped so validation, the delete-walker
+   prefix and the receiver lookup all agree on one stored form. */
+static void test_config_basis_normalization() {
+  Config* c = config_create();
+  EXPECT_NOT_NULL(c);
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, "prior/"), 0);
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, "a//b"), 0);
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, "./x/./y/"), 0);
+  EXPECT_EQ_INT(c->basis_count, 3);
+  EXPECT_EQ_STR(c->basis_dirs[0].path, "prior");
+  EXPECT_EQ_STR(c->basis_dirs[1].path, "a/b");
+  EXPECT_EQ_STR(c->basis_dirs[2].path, "x/y");
+
+  /* Degenerate values that normalize away to nothing stay rejected. */
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, "."), -1);
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, ".."), -1);
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, "/abs"), -1);
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, "a/../b"), -1);
+  EXPECT_EQ_INT(config_basis_append(c, BASIS_DEST_LINK, ""), -1);
+  config_delete(c);
+}
+
 static void test_config_is_remote_dest() {
   /* Valid SSH-style destinations */
   EXPECT_TRUE(config_is_remote_dest("user@host:/path"));
@@ -474,6 +710,12 @@ void test_config() {
     test_config_string_null_vs_empty_roundtrip();
     test_config_temp_dir_roundtrip();
     test_config_delay_updates_reserved_backup_rejected();
+    test_config_delete_timing_wire_roundtrip();
+    test_config_delete_timing_conflict_rejected();
+    test_config_basis_roundtrip();
+    test_config_basis_wire_rejects_escaping();
+    test_config_basis_normalization();
   }
+  test_config_delete_timing_early_helper();
   test_config_is_remote_dest();
 }

tests/test_protocol.c

@@ -412,6 +412,25 @@ static void test_protocol_accounting_release_does_not_underflow() {
   protocol_session_unbind();
 }
 
+static void test_send_receive_status_timed() {
+  int p[2];
+  EXPECT_EQ_INT(pipe(p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  /* The extended-deadline variant must read an ordinary status just like the
+     default window, and must fail cleanly on EOF rather than block. */
+  EXPECT_TRUE(send_status(0, STATUS_OK));
+  Status received = -1;
+  EXPECT_TRUE(receive_status_timed(0, &received, 5));
+  EXPECT_EQ_INT((int)received, (int)STATUS_OK);
+
+  close(p[1]);
+  EXPECT_FALSE(receive_status_timed(0, &received, 5));
+
+  close(p[0]);
+}
+
 void test_protocol() {
   test_send_receive_n_data();
   test_send_receive_n_data_zero();
@@ -421,6 +440,7 @@ void test_protocol() {
   test_send_receive_data();
   test_send_receive_int();
   test_send_receive_status();
+  test_send_receive_status_timed();
   test_receive_n_data_truncated();
   test_receive_str_truncated();
   test_max_alloc_rejects_single_buffer();

tests/test_server.c

@@ -180,7 +180,10 @@ static void test_receive_manifest_rejects_traversal() {
   io_set_fds(p[0], p[1]);
   EXPECT_TRUE(send_int(p[1], 1));
   EXPECT_TRUE(send_str(p[1], "../outside"));
-  EXPECT_EQ_INT(receive_manifest(p[0], cfg, NULL), -1);
+  EXPECT_NULL(receive_manifest_entries(p[0]));
+  Status status;
+  EXPECT_TRUE(receive_status(p[1], &status));
+  EXPECT_EQ_INT(status, STATUS_ERROR);
   close(p[0]);
   close(p[1]);
   config_delete(cfg);
@@ -457,6 +460,95 @@ static void test_incremental_check_delta_oversize_reports_failure() {
   }
 }
 
+/* Late-timing keep-set leak guard: a manifest parked by the commit path must
+   be freed on every error exit, never leaked.  These tests drive
+   receiver_process_pending() through an error AFTER the manifest was parked and
+   are exercised under ASan/valgrind to prove the list is released. */
+
+static Config* make_late_delete_config(const char* root) {
+  Config* cfg = config_create();
+  if (!cfg)
+    return NULL;
+  cfg->send_directory = str_dup("/src");
+  cfg->receive_root_directory = str_dup(root);
+  cfg->use_delete = true;
+  cfg->delete_after = true;
+  return cfg;
+}
+
+static int run_pending_receiver(Config* cfg, int fd, ArrayList** pending) {
+  ReceiverSink sink = {0};
+  return receiver_process_pending(cfg, fd, &sink, pending);
+}
+
+static void test_late_manifest_abort_frees_keepset() {
+  Config* cfg = make_late_delete_config("/tmp/fastsync_late_abort");
+  EXPECT_NOT_NULL(cfg);
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  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_status(p[1], STATUS_ABORT));
+
+  ArrayList* pending = NULL;
+  EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
+  EXPECT_NULL(pending);
+
+  close(p[0]);
+  close(p[1]);
+  config_delete(cfg);
+}
+
+static void test_late_manifest_eof_frees_keepset() {
+  Config* cfg = make_late_delete_config("/tmp/fastsync_late_eof");
+  EXPECT_NOT_NULL(cfg);
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  EXPECT_TRUE(send_status(p[1], STATUS_MANIFEST));
+  EXPECT_TRUE(send_int(p[1], 1));
+  EXPECT_TRUE(send_str(p[1], "keep.txt"));
+  shutdown(p[1], SHUT_WR);
+
+  ArrayList* pending = NULL;
+  EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
+  EXPECT_NULL(pending);
+
+  close(p[0]);
+  close(p[1]);
+  config_delete(cfg);
+}
+
+static void test_late_second_manifest_frees_both() {
+  Config* cfg = make_late_delete_config("/tmp/fastsync_late_second");
+  EXPECT_NOT_NULL(cfg);
+  int p[2];
+  EXPECT_EQ_INT(socketpair(AF_UNIX, SOCK_STREAM, 0, p), 0);
+  io_set_fds(p[0], p[1]);
+  io_set_bwlimit(0);
+
+  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_status(p[1], STATUS_MANIFEST));
+  EXPECT_TRUE(send_int(p[1], 1));
+  EXPECT_TRUE(send_str(p[1], "second.txt"));
+
+  ArrayList* pending = NULL;
+  EXPECT_EQ_INT(run_pending_receiver(cfg, p[0], &pending), -1);
+  EXPECT_NULL(pending);
+
+  close(p[0]);
+  close(p[1]);
+  config_delete(cfg);
+}
+
 void test_server() {
   if (!is_running_under_valgrind()) {
     test_receive_files_finished();
@@ -467,5 +559,8 @@ void test_server() {
     test_incremental_check_quick_skip_by_mtime();
     test_incremental_check_size_mismatch_full_transfer();
     test_incremental_check_delta_oversize_reports_failure();
+    test_late_manifest_abort_frees_keepset();
+    test_late_manifest_eof_frees_keepset();
+    test_late_second_manifest_frees_both();
   }
 }