src/client/client_send.c

@@ -1189,7 +1189,8 @@ static int send_chunk_delete_plans(Client* client, DeletePlanSender* plans, cons
     File* f = chunk->items[i];
     if (!f)
       continue;
-    if (delete_plan_send_for_path(client->file_descriptor, plans, file_wire_path(f), f->is_dir) != 0)
+    if (delete_plan_send_for_path(client->file_descriptor, plans, file_wire_path(f), f->is_dir) !=
+        0)
       return -1;
   }
   return 0;

src/server/receiver.c

@@ -259,8 +259,7 @@ int receiver_process(Config* config, int file_descriptor, const ReceiverSink* si
    the whole transfer succeeded.  See receiver_process_pending() for how the -m
    receiver defers that commit until its disk writer has drained. */
 int receiver_process_pending(Config* config, int file_descriptor, const ReceiverSink* sink,
-                             DeleteManifest** pending_manifest,
+                             DeleteManifest** pending_manifest, DeletePlanSession** pending_plans) {
-                             DeletePlanSession** pending_plans) {
   Status status;
   if (!receive_status(file_descriptor, &status))
     return -1;

src/server/receiver_pipeline.c

@@ -167,8 +167,8 @@ int receive_thread(void* pipeline_context) {
 
   ReceiverSink sink = {
       receiver_enqueue_file, context, false, false, NULL, receiver_pipeline_note_delete_limit};
-  if (receiver_process_pending((Config*)config, file_descriptor, &sink,
+  if (receiver_process_pending((Config*)config, file_descriptor, &sink, &context->deferred_manifest,
-                               &context->deferred_manifest, &context->deferred_plans) != 0) {
+                               &context->deferred_plans) != 0) {
     receiver_thread_fail(context);
     protocol_session_unbind();
     return thrd_error;

src/server/server.c

@@ -966,8 +966,7 @@ void handler(int file_descriptor) {
          arrived; with the disk writer drained, commit the deferred removals.
          --delete-during already applied its plans on the receive thread. */
       if (context->deferred_plans) {
-        DeleteCommitResult deletion =
+        DeleteCommitResult deletion = delete_plan_session_commit(context->deferred_plans, config);
-            delete_plan_session_commit(context->deferred_plans, config);
         if (deletion == DELETE_COMMIT_ERROR) {
           transfer_ok = false;
         } else if (deletion == DELETE_COMMIT_LIMIT_REACHED) {

src/shared/delete_plan.c

@@ -484,7 +484,8 @@ typedef struct PlanSkips {
   int count;
 } PlanSkips;
 
-static bool build_plan_skips(const Config* config, const DeletePlanSession* session, PlanSkips* out) {
+static bool build_plan_skips(const Config* config, const DeletePlanSession* session,
+                             PlanSkips* out) {
   out->entries = NULL;
   out->count = 0;
   int count = (config->delay_updates ? 1 : 0) + config->basis_count +
@@ -569,8 +570,8 @@ static bool process_extra_dir(int dirfd, const char* name, const char* child_rel
     return false;
   }
   bool survives = false;
-  bool ok = process_children(childfd, child_rel, NULL, NULL, false, force_now, skips, session,
+  bool ok =
-                             &survives);
+      process_children(childfd, child_rel, NULL, NULL, false, force_now, skips, session, &survives);
   close(childfd);
   if (!ok)
     return false;
@@ -637,8 +638,8 @@ static bool process_children(int dirfd, const char* dir_rel, const ArrayList* ke
   while ((entry = readdir(dir)) != NULL) {
     if (strcmp(entry->d_name, ".") == 0 || strcmp(entry->d_name, "..") == 0)
       continue;
-    char* child_rel = (strcmp(dir_rel, ".") == 0) ? str_dup(entry->d_name)
+    char* child_rel =
-                                                  : path_cat(dir_rel, entry->d_name);
+        (strcmp(dir_rel, ".") == 0) ? str_dup(entry->d_name) : path_cat(dir_rel, entry->d_name);
     if (!child_rel) {
       operation_ok = false;
       continue;
@@ -704,8 +705,8 @@ static bool apply_plan_dir(DeletePlanSession* session, const Config* config, con
     return false;
   }
   bool survives = false;
-  bool ok = process_children(dirfd, dir, dirs, files, strcmp(dir, ".") == 0, false, &skips,
+  bool ok = process_children(dirfd, dir, dirs, files, strcmp(dir, ".") == 0, false, &skips, session,
-                             session, &survives);
+                             &survives);
   free(skips.entries);
   close(dirfd);
   if (!ok)
@@ -719,8 +720,8 @@ static bool apply_missing(DeletePlanSession* session, const Config* config) {
   session->missing_applied = true;
   if (session->missing->size == 0)
     return true;
-  DeleteManifest manifest = {.keeps = NULL, .protected = NULL, .missing = session->missing,
+  DeleteManifest manifest = {
-                             .dirs = NULL};
+      .keeps = NULL, .protected = NULL, .missing = session->missing, .dirs = NULL};
   size_t remaining = budget_available(session) ? session->max_delete - session->deleted : 0;
   size_t deleted = 0;
   size_t skipped = 0;

tests/integration/test_delete_timing_parity.py

@@ -0,0 +1,296 @@
+"""Differential + regression coverage for rsync's delete timing.
+
+``--delete-during``/``--delete-delay`` stream a per-directory delete plan instead
+of one whole-tree manifest, so the timing is observable:
+
+  * ``--delete-during`` removes a directory's extras as it processes that
+    directory (so an interrupted transfer has already removed the extras of the
+    directories it reached);
+  * ``--delete-delay`` snapshots those extras while scanning and commits the
+    removals only after a fully-successful transfer (so an extra created in the
+    destination after its directory's plan survives, and a failed transfer
+    removes nothing);
+  * ``--delete-after`` re-scans the destination at the end (so that same
+    late-created extra is removed).
+
+The final-state tests compare against real ``rsync 3.4.1`` where a deterministic
+comparison exists; the timing tests use a byte-slicing proxy to force a
+mid-transfer failure or to create a destination entry while the transfer is in
+flight.
+"""
+import os
+import select
+import shutil
+import socket
+import struct
+import subprocess
+import sys
+import threading
+import time
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(__file__))
+from common import (  # noqa: E402
+    BUILD_DIR,
+    TEST_DATA_DIR,
+    ServerManager,
+    clean_dir,
+    get_dest_received_dir,
+    run_client,
+)
+
+# Every test here is deterministic (the proxy throttles until the delete-plan
+# frames are processed), so the PR gate runs the whole module.
+pytestmark = pytest.mark.ci
+
+RSYNC = shutil.which("rsync")
+requires_rsync = pytest.mark.skipif(RSYNC is None, reason="rsync 3.4.1 not installed")
+
+BIG_BYTES = 8 * 1024 * 1024
+# Forward/cut this far into the stream: past the (small) delete-plan frames and
+# well into the big payload, so the receiver has already processed the plan.
+MID_TRANSFER_BYTES = 256 * 1024
+# Throttle the proxy so the receiver keeps up with the (fast) client and the
+# plan frames are provably processed before the hook/cut offset is reached.
+PROXY_THROTTLE = 0.001
+
+
+def _write(path, content):
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "wb") as fh:
+        fh.write(content)
+
+
+def _seed_pair(tag, big=False):
+    """Create a source tree and a destination mirror seeded with extras.
+
+    The tree is a single directory ``d`` containing the transferred files plus,
+    in the destination, an extra ``d/old_extra``.
+    """
+    source = os.path.join(TEST_DATA_DIR, f"dtp_{tag}_src")
+    dest = os.path.join(TEST_DATA_DIR, f"dtp_{tag}_dst")
+    clean_dir(source)
+    clean_dir(dest)
+    _write(os.path.join(source, "d", "keep.txt"), b"kept payload\n")
+    if big:
+        _write(os.path.join(source, "d", "big.bin"), b"B" * BIG_BYTES)
+    received = get_dest_received_dir(dest, source)
+    os.makedirs(os.path.join(received, "d"), exist_ok=True)
+    _write(os.path.join(received, "d", "old_extra"), b"stale extra\n")
+    return source, dest, received
+
+
+def _tree(root):
+    """Sorted relative paths of every entry below root (files and dirs)."""
+    out = []
+    for dirpath, dirs, files in os.walk(root):
+        for name in dirs:
+            out.append(os.path.relpath(os.path.join(dirpath, name), root))
+        for name in files:
+            out.append(os.path.relpath(os.path.join(dirpath, name), root))
+    return sorted(out)
+
+
+def _rsync(args):
+    env = dict(os.environ, LC_ALL="C")
+    return subprocess.run([RSYNC] + args, capture_output=True, text=True, env=env, timeout=120)
+
+
+class _SlicingProxy:
+    """Forward the client stream to a server, optionally cutting it or invoking a
+    hook after a byte threshold.  ``forward_limit`` mode resets both ends after
+    that many client bytes (a mid-transfer failure).  ``hook`` mode calls the
+    hook once and keeps forwarding to completion."""
+
+    def __init__(self, target_port, forward_limit=None, hook=None, hook_after=0,
+                 throttle=0.0):
+        self.target = ("127.0.0.1", target_port)
+        self.forward_limit = forward_limit
+        self.hook = hook
+        self.hook_after = hook_after
+        self.throttle = throttle
+        self.hook_called = threading.Event()
+        self.listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.listener.bind(("127.0.0.1", 0))
+        self.listener.listen(1)
+        self.listener.settimeout(20)
+        self.port = self.listener.getsockname()[1]
+        self._thread = threading.Thread(target=self._serve, daemon=True)
+        self._thread.start()
+
+    def _serve(self):
+        try:
+            client, _ = self.listener.accept()
+        except OSError:
+            return
+        try:
+            backend = socket.create_connection(self.target, timeout=10)
+        except OSError:
+            client.close()
+            return
+        client.settimeout(20)
+        backend.settimeout(20)
+        forwarded = 0
+        socks = [client, backend]
+        try:
+            while socks:
+                ready, _, _ = select.select(socks, [], [], 20)
+                if not ready:
+                    break
+                for sock in ready:
+                    data = sock.recv(65536)
+                    if not data:
+                        socks.remove(sock)
+                        peer = backend if sock is client else client
+                        try:
+                            peer.shutdown(socket.SHUT_WR)
+                        except OSError:
+                            pass
+                        continue
+                    if sock is client:
+                        if self.forward_limit is not None:
+                            room = self.forward_limit - forwarded
+                            if room <= 0:
+                                socks = []
+                                break
+                            data = data[:room]
+                        backend.sendall(data)
+                        forwarded += len(data)
+                        if (self.hook is not None and not self.hook_called.is_set()
+                                and forwarded >= self.hook_after):
+                            # Give the receiver time to process the (tiny) plan
+                            # frames that precede this offset before the hook
+                            # mutates the destination.
+                            if self.throttle > 0:
+                                time.sleep(0.2)
+                            self.hook()
+                            self.hook_called.set()
+                        if self.forward_limit is not None and forwarded >= self.forward_limit:
+                            socks = []
+                            break
+                        if self.throttle > 0:
+                            time.sleep(self.throttle)
+                    else:
+                        client.sendall(data)
+        except OSError:
+            pass
+        for sock in (client, backend):
+            try:
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+            except OSError:
+                pass
+            try:
+                sock.close()
+            except OSError:
+                pass
+        try:
+            self.listener.close()
+        except OSError:
+            pass
+
+    def finish(self):
+        self._thread.join(30)
+        try:
+            self.listener.close()
+        except OSError:
+            pass
+
+
+class TestDeleteTimingFinalStateParity:
+    """On a successful transfer the per-directory timings match rsync's result."""
+
+    def _run_fastsync(self, tag, timing):
+        source, dest, received = _seed_pair(tag)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, flags=[timing], port=server.port)
+        return result, received
+
+    @pytest.mark.parametrize("timing", ["--delete-during", "--delete-delay"])
+    @requires_rsync
+    def test_success_final_state_matches_rsync(self, timing):
+        # Build the rsync fixture from the same seed so both sides start equal.
+        source, dest, received = _seed_pair("parity_rsync")
+        source2 = source
+        rsync_dst = os.path.join(TEST_DATA_DIR, "dtp_parity_rsync_dst")
+        clean_dir(rsync_dst)
+        # rsync mirrors src/ into dst/; seed the same extra.
+        _write(os.path.join(rsync_dst, "d", "old_extra"), b"stale extra\n")
+
+        rsync_result = _rsync(["-a", timing, source2 + "/", rsync_dst + "/"])
+        assert rsync_result.returncode == 0, rsync_result.stderr
+        rsync_tree = _tree(rsync_dst)
+
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            result, _ = run_client(source, dest, flags=[timing], port=server.port)
+        assert result.returncode == 0, (result.stderr or result.stdout)[:300]
+        fastsync_tree = _tree(received)
+        assert fastsync_tree == rsync_tree, (
+            f"{timing}: fastsync tree {fastsync_tree} != rsync tree {rsync_tree}"
+        )
+
+
+class TestDeleteTimingFailure:
+    """A mid-transfer failure distinguishes during from delay."""
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_during_removes_delay_preserves_on_failure(self, mt):
+        source, dest, received = _seed_pair("failure", big=True)
+        extra = os.path.join(received, "d", "old_extra")
+        assert os.path.exists(extra)
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            for timing, expect_removed in (("--delete-during", True),
+                                           ("--delete-delay", False)):
+                # Re-seed the extra before each run.
+                _write(extra, b"stale extra\n")
+                proxy = _SlicingProxy(server.port, forward_limit=MID_TRANSFER_BYTES, throttle=PROXY_THROTTLE)
+                flags = [timing] + (["--threads"] if mt else [])
+                result, _ = run_client(source, dest, flags=flags, port=proxy.port)
+                proxy.finish()
+                assert result.returncode != 0, f"{timing}: truncated transfer succeeded"
+                present = os.path.exists(extra)
+                assert present != expect_removed, (
+                    f"{timing} (mt={mt}): extra present={present}, expected "
+                    f"removed={expect_removed}"
+                )
+
+
+class TestDeleteDelayVsAfterSnapshot:
+    """A destination entry created after its directory's scan survives under
+    --delete-delay but is removed by --delete-after's fresh end scan."""
+
+    @pytest.mark.parametrize("mt", [False, True])
+    def test_late_created_extra_survives_delay_not_after(self, mt):
+        source, dest, received = _seed_pair("latecreate", big=True)
+        old_extra = os.path.join(received, "d", "old_extra")
+        new_extra = os.path.join(received, "d", "new_extra")
+        with ServerManager() as server:
+            server.start(extra_args=["--allow-delete"])
+            for timing, new_survives in (("--delete-delay", True),
+                                         ("--delete-after", False)):
+                _write(old_extra, b"stale extra\n")
+                if os.path.exists(new_extra):
+                    os.unlink(new_extra)
+
+                def hook():
+                    # Runs on the proxy thread while the big file is in flight,
+                    # after the directory's plan (delay) has been processed.
+                    _write(new_extra, b"created mid-transfer\n")
+
+                proxy = _SlicingProxy(server.port, hook=hook, hook_after=MID_TRANSFER_BYTES, throttle=PROXY_THROTTLE)
+                flags = [timing] + (["--threads"] if mt else [])
+                result, _ = run_client(source, dest, flags=flags, port=proxy.port)
+                proxy.finish()
+                assert result.returncode == 0, (
+                    f"{timing}: {(result.stderr or result.stdout)[:300]}"
+                )
+                assert proxy.hook_called.is_set(), f"{timing}: hook never fired"
+                assert not os.path.exists(old_extra), f"{timing}: old extra survived"
+                assert os.path.exists(new_extra) == new_survives, (
+                    f"{timing} (mt={mt}): new_extra present="
+                    f"{os.path.exists(new_extra)}, expected survives={new_survives}"
+                )