feat: implement rsync delete timing (--delete-before/--delete-during/--delete-delay/--delete-after)

Deletion timing is now real and selected by the four rsync flags plus the
plain --delete default. Wire protocol bumps to 2.8.0: two new config
booleans (delete_during, delete_delay) are serialized and validated, joining
the existing delete_before/delete_after.

- Early modes (--delete-before, --delete-during/--del): the sender pre-scans
  the whole tree (paths only), transmits the keep-set manifest BEFORE any
  file data, and the receiver removes extras and acks STATUS_OK; the sender
  only streams data after the deletion committed. Deletion is thus performed
  even if a later transfer phase fails (rsync delete-before/during are
  destructive by definition). FastSync streams in a single scan so it cannot
  interleave per-directory like rsync delete-during; --delete-during selects
  the same engine mode as --delete-before (documented divergence).
- Late/commit modes (plain --delete, --delete-after, --delete-delay): the
  manifest closes the data stream and deletion is committed only after
  STATUS_FINISHED proves the whole transfer succeeded, preserving FastSync's
  commit-style safety. --delete-delay converges with --delete-after because
  FastSync never snapshots the destination during data flow (documented).
- The STATUS_MANIFEST frame is now self-delimiting and position-independent.
  Single-threaded receivers delete before the success frame; the -m receiver
  hands the keep-set to server.c, which commits the deletion only after the
  disk writer thread has drained (fixes a delete-vs-in-flight-temp race).
- Every timing flag implies --delete; at most one timing flag is allowed.
- Each timing flag implies --delete, matching rsync; conflicts are rejected.

src/shared/config.c

@@ -115,6 +115,8 @@ static void config_set_defaults(Config* config) {
   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 +163,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 +192,26 @@ 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_is_remote_dest(const char* s) {
   if (s == NULL)
     return false;
@@ -311,7 +335,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) {
@@ -418,7 +443,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) {

src/shared/config.h

@@ -147,6 +147,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,7 +187,7 @@ 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)
 
 Config* config_create(void);
@@ -184,4 +197,16 @@ 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);
+
 #endif

src/shared/file_receive.c

@@ -792,26 +792,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 +824,22 @@ 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;
-  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;
+  return delete_extras_limited(config->receive_root_directory, manifest, MAX_SERVER_DELETE_COUNT,
+                               skip_staging);
 }

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/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;

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,