fix(engine): give local --append-verify the phase-2 redo model

crates/engine/src/local_copy/context.rs

@@ -401,14 +401,27 @@ impl<'a> FinalizeMetadataParams<'a> {
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 pub(crate) struct FileCopyOutcome {
     literal_bytes: u64,
+    matched_bytes: u64,
     compressed_bytes: Option<u64>,
 }
 
+/// Matched-byte count for copy paths that never match a basis block.
+///
+/// A whole-file copy, a sparse whole-file copy, and an append all reach the
+/// destination without consulting a signature, so nothing is "matched" even
+/// though most of the resulting file may not be literal either. upstream:
+/// match_sums() only reaches `matched()` (match.c:121) via `hash_search()`,
+/// which is skipped when `s->count == 0` - the whole-file case - and append
+/// mode zeroes `s->count` outright (match.c:389-390).
+pub(crate) const MATCHED_NONE: u64 = 0;
+
 impl FileCopyOutcome {
-    /// Creates a new outcome with the given literal and optional compressed byte counts.
-    const fn new(literal_bytes: u64, compressed_bytes: Option<u64>) -> Self {
+    /// Creates a new outcome with the given literal, matched, and optional
+    /// compressed byte counts.
+    const fn new(literal_bytes: u64, matched_bytes: u64, compressed_bytes: Option<u64>) -> Self {
         Self {
             literal_bytes,
+            matched_bytes,
             compressed_bytes,
         }
     }
@@ -418,6 +431,16 @@ impl FileCopyOutcome {
         self.literal_bytes
     }
 
+    /// Returns the number of bytes supplied by block matches against the basis.
+    ///
+    /// Reported by the copy paths that actually emit matches, never inferred
+    /// from the file size. upstream: match.c:121 - `stats.matched_data` grows
+    /// only inside `matched()`, so a whole-file copy and an append both leave
+    /// it at zero even though most of their bytes were not literal.
+    pub(crate) const fn matched_bytes(self) -> u64 {
+        self.matched_bytes
+    }
+
     /// Returns the compressed byte count, if compression was used.
     pub(crate) const fn compressed_bytes(self) -> Option<u64> {
         self.compressed_bytes
@@ -997,21 +1020,21 @@ mod tests {
 
     #[test]
     fn file_copy_outcome_new_stores_values() {
-        let outcome = FileCopyOutcome::new(1000, Some(500));
+        let outcome = FileCopyOutcome::new(1000, 0, Some(500));
         assert_eq!(outcome.literal_bytes(), 1000);
         assert_eq!(outcome.compressed_bytes(), Some(500));
     }
 
     #[test]
     fn file_copy_outcome_new_without_compression() {
-        let outcome = FileCopyOutcome::new(2000, None);
+        let outcome = FileCopyOutcome::new(2000, 0, None);
         assert_eq!(outcome.literal_bytes(), 2000);
         assert!(outcome.compressed_bytes().is_none());
     }
 
     #[test]
     fn file_copy_outcome_zero_bytes() {
-        let outcome = FileCopyOutcome::new(0, Some(0));
+        let outcome = FileCopyOutcome::new(0, 0, Some(0));
         assert_eq!(outcome.literal_bytes(), 0);
         assert_eq!(outcome.compressed_bytes(), Some(0));
     }
@@ -1025,25 +1048,25 @@ mod tests {
 
     #[test]
     fn file_copy_outcome_clone() {
-        let outcome = FileCopyOutcome::new(100, Some(50));
+        let outcome = FileCopyOutcome::new(100, 0, Some(50));
         let cloned = outcome;
         assert_eq!(cloned.literal_bytes(), 100);
         assert_eq!(cloned.compressed_bytes(), Some(50));
     }
 
     #[test]
     fn file_copy_outcome_debug_format() {
-        let outcome = FileCopyOutcome::new(100, None);
+        let outcome = FileCopyOutcome::new(100, 0, None);
         let debug = format!("{outcome:?}");
         assert!(debug.contains("FileCopyOutcome"));
         assert!(debug.contains("100"));
     }
 
     #[test]
     fn file_copy_outcome_eq() {
-        let a = FileCopyOutcome::new(100, Some(50));
-        let b = FileCopyOutcome::new(100, Some(50));
-        let c = FileCopyOutcome::new(100, Some(60));
+        let a = FileCopyOutcome::new(100, 0, Some(50));
+        let b = FileCopyOutcome::new(100, 0, Some(50));
+        let c = FileCopyOutcome::new(100, 0, Some(60));
         assert_eq!(a, b);
         assert_ne!(a, c);
     }

crates/engine/src/local_copy/context_impl/delta_transfer.rs

@@ -119,6 +119,14 @@ impl<'a> CopyContext<'a> {
         // match_report()).
         let mut probe = ProbeCounters::default();
         let mut file_matches = 0u64;
+        // Accumulated where matches are produced, never derived from
+        // `total_size - literal_bytes`. upstream: match.c:121 `matched()` is the
+        // only place `stats.matched_data` grows, so a byte counts as matched
+        // exactly when a block match emitted it - bytes that are neither
+        // literal nor matched (the prefix an append skips over) count as
+        // neither. upstream: match.c:389-390 sets `last_match = s->flength;
+        // s->count = 0;` so append mode never reaches `matched()` at all.
+        let mut matched_bytes = 0u64;
         let mut sparse_state = SparseWriteState::default();
         sparse_state.set_preallocated_len(preallocated_len);
         let mut window: VecDeque<u8> = VecDeque::with_capacity(index.block_length());
@@ -281,6 +289,7 @@ impl<'a> CopyContext<'a> {
                 }
 
                 total_bytes = total_bytes.saturating_add(block_len as u64);
+                matched_bytes = matched_bytes.saturating_add(block_len as u64);
                 let progressed = initial_bytes.saturating_add(total_bytes);
                 self.notify_progress(relative, Some(total_size), progressed, start.elapsed());
                 window.clear();
@@ -395,6 +404,7 @@ impl<'a> CopyContext<'a> {
             }
 
             total_bytes = total_bytes.saturating_add(block_len as u64);
+            matched_bytes = matched_bytes.saturating_add(block_len as u64);
             let progressed = initial_bytes.saturating_add(total_bytes);
             self.notify_progress(relative, Some(total_size), progressed, start.elapsed());
             window.clear();
@@ -477,9 +487,9 @@ impl<'a> CopyContext<'a> {
             let delta = compressed_total.saturating_sub(compressed_progress);
             self.register_limiter_bytes(delta);
             self.record_adaptive_compression(literal_bytes, compressed_total);
-            FileCopyOutcome::new(literal_bytes, Some(compressed_total))
+            FileCopyOutcome::new(literal_bytes, matched_bytes, Some(compressed_total))
         } else {
-            FileCopyOutcome::new(literal_bytes, None)
+            FileCopyOutcome::new(literal_bytes, matched_bytes, None)
         };
 
         Ok(outcome)

crates/engine/src/local_copy/context_impl/transfer.rs

@@ -634,7 +634,7 @@ impl<'a> CopyContext<'a> {
                 let progressed = initial_bytes.saturating_add(copied);
                 self.notify_progress(relative, Some(total_size), progressed, start.elapsed());
             }
-            return Ok(FileCopyOutcome::new(copied, None));
+            return Ok(FileCopyOutcome::new(copied, MATCHED_NONE, None));
         }
 
         if sparse {
@@ -715,9 +715,9 @@ impl<'a> CopyContext<'a> {
             let delta = compressed_total.saturating_sub(compressed_progress);
             self.register_limiter_bytes(delta);
             self.record_adaptive_compression(literal_bytes, compressed_total);
-            FileCopyOutcome::new(literal_bytes, Some(compressed_total))
+            FileCopyOutcome::new(literal_bytes, MATCHED_NONE, Some(compressed_total))
         } else {
-            FileCopyOutcome::new(literal_bytes, None)
+            FileCopyOutcome::new(literal_bytes, MATCHED_NONE, None)
         };
 
         Ok(outcome)
@@ -830,9 +830,9 @@ impl<'a> CopyContext<'a> {
             let delta = compressed_total.saturating_sub(compressed_progress);
             self.register_limiter_bytes(delta);
             self.record_adaptive_compression(literal_bytes, compressed_total);
-            FileCopyOutcome::new(literal_bytes, Some(compressed_total))
+            FileCopyOutcome::new(literal_bytes, MATCHED_NONE, Some(compressed_total))
         } else {
-            FileCopyOutcome::new(literal_bytes, None)
+            FileCopyOutcome::new(literal_bytes, MATCHED_NONE, None)
         };
 
         Ok(outcome)

crates/engine/src/local_copy/executor/file/append.rs

@@ -14,15 +14,32 @@ pub(crate) enum AppendMode {
     Disabled,
     /// Destination is already at least as large as the source; skip the file.
     Skip,
-    /// Destination is shorter; append starting from the given offset.
-    Append(u64),
+    /// Destination is shorter; append starting from `offset`.
+    Append {
+        /// Byte offset the appended tail starts at - the destination's length.
+        offset: u64,
+        /// Whether `--append-verify`'s whole-file re-checksum will fail, so the
+        /// caller must retain the appended result and redo the file in phase 2.
+        verify_failed: bool,
+    },
 }
 
 /// Decides the append strategy for a file based on existing destination size.
 ///
 /// Returns `Disabled` when append is off, `Skip` when the destination is
-/// already at least as large, or `Append(offset)` when the destination is
-/// shorter and the transfer should resume from that offset.
+/// already at least as large, or `Append` when the destination is shorter and
+/// the transfer should resume from that offset.
+///
+/// Under `--append-verify` a mismatching prefix does **not** cancel the append.
+/// Upstream appends first and only then compares whole-file checksums: the
+/// sender sums the source's first `flength` bytes and the receiver sums the
+/// destination's, both followed by the identical appended tail
+/// (match.c:372-391, receiver.c:352-379), so the comparison reduces exactly to
+/// "do the two prefixes agree". A disagreement is reported through
+/// `verify_failed` rather than acted on here, because upstream keeps the
+/// appended bytes on disk (receiver.c:1029, reached because `--append` implies
+/// `--inplace` - options.c:2400-2411) and redoes the file in phase 2 against
+/// that retained partial.
 // upstream: receiver.c:recv_files() - append mode size comparison
 pub(crate) fn determine_append_mode(
     append_allowed: bool,
@@ -60,21 +77,23 @@ pub(crate) fn determine_append_mode(
         return Ok(AppendMode::Skip);
     }
 
-    if append_verify {
-        let matches = verify_append_prefix(reader, source, destination, existing_len)?;
-        reader
-            .seek(SeekFrom::Start(0))
-            .map_err(|error| LocalCopyError::io("copy file", source, error))?;
-        if !matches {
-            return Ok(AppendMode::Disabled);
-        }
+    // Plain `--append` (append_mode == 1) never re-checksums: upstream skips the
+    // prefix in `sum_update` on both sides (match.c:373-391 runs the CHUNK_SIZE
+    // loop only when `append_mode == 2`), so the two whole-file sums cover just
+    // the appended tail and always agree.
+    let verify_failed = if append_verify {
+        !verify_append_prefix(reader, source, destination, existing_len)?
     } else {
-        reader
-            .seek(SeekFrom::Start(0))
-            .map_err(|error| LocalCopyError::io("copy file", source, error))?;
-    }
+        false
+    };
+    reader
+        .seek(SeekFrom::Start(0))
+        .map_err(|error| LocalCopyError::io("copy file", source, error))?;
 
-    Ok(AppendMode::Append(existing_len))
+    Ok(AppendMode::Append {
+        offset: existing_len,
+        verify_failed,
+    })
 }
 
 /// Verifies that the existing destination prefix matches the source.
@@ -272,7 +291,15 @@ mod tests {
         .expect("determine");
 
         match result {
-            AppendMode::Append(offset) => assert_eq!(offset, 6), // "source" is 6 bytes
+            AppendMode::Append {
+                offset,
+                verify_failed,
+            } => {
+                assert_eq!(offset, 6); // "source" is 6 bytes
+                // Without --append-verify upstream never re-checksums the
+                // prefix, so no redo can be requested.
+                assert!(!verify_failed);
+            }
             AppendMode::Disabled | AppendMode::Skip => panic!("expected Append mode"),
         }
     }
@@ -299,13 +326,19 @@ mod tests {
         .expect("determine");
 
         match result {
-            AppendMode::Append(offset) => assert_eq!(offset, 15),
+            AppendMode::Append {
+                offset,
+                verify_failed,
+            } => {
+                assert_eq!(offset, 15);
+                assert!(!verify_failed);
+            }
             AppendMode::Disabled | AppendMode::Skip => panic!("expected Append mode"),
         }
     }
 
     #[test]
-    fn determine_append_mode_with_verify_disabled_when_prefix_mismatch() {
+    fn determine_append_mode_still_appends_when_verify_will_fail() {
         let temp = tempdir().expect("tempdir");
         let source_path = temp.path().join("source.txt");
         let dest_path = temp.path().join("dest.txt");
@@ -325,7 +358,25 @@ mod tests {
         )
         .expect("determine");
 
-        assert!(matches!(result, AppendMode::Disabled));
+        // Degrading to a plain whole-file copy here is exactly the bug this
+        // encodes against: upstream appends the tail regardless, keeps the
+        // result (--append implies --inplace, options.c:2400-2411), and only
+        // then reports the failed whole-file re-checksum so the generator can
+        // redo the file against the retained partial (generator.c:2175-2217).
+        // Cancelling the append would leave nothing to re-delta and would make
+        // the transfer look like a clean single-pass copy.
+        match result {
+            AppendMode::Append {
+                offset,
+                verify_failed,
+            } => {
+                assert_eq!(offset, 16);
+                assert!(verify_failed);
+            }
+            AppendMode::Disabled | AppendMode::Skip => {
+                panic!("a failed verification must still append, then redo")
+            }
+        }
     }
 
     #[test]

crates/engine/src/local_copy/executor/file/copy/dry_run.rs

@@ -160,8 +160,12 @@ pub(super) fn handle_dry_run(
         ));
         return Ok(());
     }
+    // A dry run never writes, so the phase-2 redo a failed `--append-verify`
+    // would trigger has nothing to report: upstream's `if (!do_xfers)` leg
+    // (receiver.c:805-810) logs the item and `continue`s, never reaching the
+    // `recv_ok` switch that queues the redo.
     let append_offset = match append_mode {
-        AppendMode::Append(offset) => offset,
+        AppendMode::Append { offset, .. } => offset,
         AppendMode::Disabled | AppendMode::Skip => 0,
     };
     let bytes_transferred = file_size.saturating_sub(append_offset);

crates/engine/src/local_copy/executor/file/copy/transfer/execute/clonefile.rs

@@ -156,7 +156,9 @@ pub(super) fn try_clone(
     context.record_hard_link(metadata, destination);
     context
         .summary_mut()
-        .record_file(file_size, file_size, None);
+        // A whole-file clone/copy is all literal: no signature was
+        // consulted, so nothing was matched.
+        .record_file(file_size, file_size, crate::local_copy::MATCHED_NONE, None);
     context
         .summary_mut()
         .record_copy_method(CopyMethodKind::from_platform(clone_method));

crates/engine/src/local_copy/executor/file/copy/transfer/execute/ficlone.rs

@@ -143,7 +143,9 @@ pub(super) fn try_clone(
     context.record_hard_link(metadata, destination);
     context
         .summary_mut()
-        .record_file(file_size, file_size, None);
+        // A whole-file clone/copy is all literal: no signature was
+        // consulted, so nothing was matched.
+        .record_file(file_size, file_size, crate::local_copy::MATCHED_NONE, None);
     context
         .summary_mut()
         .record_copy_method(CopyMethodKind::Ficlone);

crates/engine/src/local_copy/executor/file/copy/transfer/execute/iouring.rs

@@ -118,7 +118,9 @@ pub(super) fn try_dispatch(
     context.record_hard_link(metadata, destination);
     context
         .summary_mut()
-        .record_file(file_size, file_size, None);
+        // A whole-file clone/copy is all literal: no signature was
+        // consulted, so nothing was matched.
+        .record_file(file_size, file_size, crate::local_copy::MATCHED_NONE, None);
     context
         .summary_mut()
         .record_copy_method(CopyMethodKind::IoUring);