fix(transfer): interleave itemize rows in flist order (run_pipelined, step 1)

crates/transfer/src/receiver/context.rs

@@ -5,7 +5,8 @@
 //! loop. The itemize/info-line emission methods live in the sibling
 //! [`super::itemize`] module.
 
-use std::collections::HashMap;
+use std::cell::RefCell;
+use std::collections::{BTreeMap, HashMap};
 use std::num::NonZeroU8;
 use std::sync::Arc;
 use std::sync::atomic::Ordering;
@@ -230,6 +231,24 @@ pub struct ReceiverContext {
     /// - `generator.c:2300-2305` - `preserve_hard_links && inc_recurse` pre-reads
     ///   until `file_total < MIN_FILECNT_LOOKAHEAD / 2`.
     pub(in crate::receiver) hardlink_lookahead_target: usize,
+    /// When set, itemize rows are buffered (keyed by flist index) instead of
+    /// written to the client sink at their emit site, then drained once in
+    /// flist-index order by [`Self::flush_itemize_rows`] just before
+    /// finalization. This reproduces upstream's single flist-index-order walk
+    /// (`generator.c:2329-2344` -> `recv_generator` -> `itemize`), where a
+    /// directory row immediately precedes its children, rather than oc's
+    /// two-phase "all dirs, then all files" emission order.
+    ///
+    /// Defaults to `false`: every non-deferred caller emits immediately, exactly
+    /// as before. Only [`Self::run_pipelined`] opts in; the sync, incremental,
+    /// and async receive paths are converted separately.
+    pub(in crate::receiver) defer_itemize: bool,
+    /// Buffer of rendered itemize rows keyed by flist index, drained in ascending
+    /// key order by [`Self::flush_itemize_rows`]. Populated only while
+    /// [`Self::defer_itemize`] is set. A `Vec` per index tolerates the phase-2
+    /// redo re-itemizing an entry. `RefCell` suffices: every emit site runs on
+    /// the main driver thread (rayon workers never touch this).
+    pub(in crate::receiver) itemize_rows: RefCell<BTreeMap<usize, Vec<String>>>,
 }
 
 impl ReceiverContext {
@@ -296,6 +315,8 @@ impl ReceiverContext {
             flist_eof: false,
             // upstream: generator.c:2304 - MIN_FILECNT_LOOKAHEAD / 2 (1000 / 2).
             hardlink_lookahead_target: 500,
+            defer_itemize: false,
+            itemize_rows: RefCell::new(BTreeMap::new()),
         }
     }
 

crates/transfer/src/receiver/directory/creation.rs

@@ -343,7 +343,10 @@ impl ReceiverContext {
                     // `cd+++++++++ ./` when the pre-flight mkdir created the root.
                     crate::generator::ItemFlags::from_raw(self.existing_dir_iflags(entry, dir_path))
                 };
-                let _ = self.emit_itemize(writer, &iflags, entry);
+                // Deferred on the run_pipelined path so the dir row lands in
+                // flist-index order (immediately before its children) at flush
+                // time; emitted immediately on every other path.
+                let _ = self.emit_or_record_itemize(writer, *idx, &iflags, entry);
             }
         }
 

crates/transfer/src/receiver/itemize.rs

@@ -192,4 +192,77 @@ impl ReceiverContext {
             Ok(())
         }
     }
+
+    /// Emits an itemize row immediately, or buffers it for the deferred
+    /// flist-index-order flush when [`Self::defer_itemize`] is set.
+    ///
+    /// Callers on the deferred path (currently only `run_pipelined`) pass the
+    /// entry's flist index so the buffered row can be re-ordered against rows
+    /// recorded by other passes (directory creation, candidate selection). When
+    /// deferral is off every other receive path emits at the call site exactly
+    /// as before.
+    pub(in crate::receiver) fn emit_or_record_itemize<W: crate::writer::MsgInfoSender + ?Sized>(
+        &self,
+        writer: &mut W,
+        flist_idx: usize,
+        iflags: &crate::generator::ItemFlags,
+        entry: &protocol::flist::FileEntry,
+    ) -> std::io::Result<()> {
+        if self.defer_itemize {
+            self.record_itemize(flist_idx, iflags, entry);
+            Ok(())
+        } else {
+            self.emit_itemize(writer, iflags, entry)
+        }
+    }
+
+    /// Buffers one itemize row under its flist index for the deferred
+    /// flist-index-order flush.
+    ///
+    /// The visibility gate mirrors [`Self::emit_itemize`] exactly: only a
+    /// client-mode receiver (a pull) produces a client-visible row. A server
+    /// receiver's row travels as wire iflags and is printed by the client's
+    /// sender (upstream `log.c:822` gates the `FCLIENT` write on `!am_server`),
+    /// so recording one here would double it. Preserving this gate keeps the
+    /// deferred flush byte-identical to the immediate emit - only the ordering
+    /// changes.
+    pub(in crate::receiver) fn record_itemize(
+        &self,
+        flist_idx: usize,
+        iflags: &crate::generator::ItemFlags,
+        entry: &protocol::flist::FileEntry,
+    ) {
+        if !self.should_emit_itemize() || !self.config.connection.client_mode {
+            return;
+        }
+        if let Some(line) = self.render_itemize_line(iflags, entry) {
+            self.itemize_rows
+                .borrow_mut()
+                .entry(flist_idx)
+                .or_default()
+                .push(line);
+        }
+    }
+
+    /// Drains every buffered itemize row in ascending flist-index order,
+    /// routing each through the client sink.
+    ///
+    /// Mirrors upstream's single flist-index-order walk in `generate_files`
+    /// (`generator.c:2329-2344`), where each entry is itemized as it is reached
+    /// so a directory row immediately precedes its children. Because only
+    /// client-mode rows are ever recorded (see [`Self::record_itemize`]),
+    /// [`Self::emit_info_line`] writes them to the client's stdout. Called once
+    /// by the driver just before finalization.
+    pub(in crate::receiver) fn flush_itemize_rows<W: crate::writer::MsgInfoSender + ?Sized>(
+        &self,
+        writer: &mut W,
+    ) -> std::io::Result<()> {
+        let rows = std::mem::take(&mut *self.itemize_rows.borrow_mut());
+        for (_idx, lines) in rows {
+            for line in lines {
+                self.emit_info_line(writer, &line)?;
+            }
+        }
+        Ok(())
+    }
 }

crates/transfer/src/receiver/transfer/candidates.rs

@@ -285,6 +285,7 @@ impl ReceiverContext {
                     let unchanged_iflags = self.itemize_existing_flags(entry, meta, 0);
                     self.apply_no_change_metadata(
                         writer,
+                        idx,
                         &file_path,
                         entry,
                         meta,
@@ -348,7 +349,10 @@ impl ReceiverContext {
             // (after the transfer completes) to match `log_before_transfer == 0`.
             if emit_itemize && self.config.connection.client_mode {
                 let iflags = crate::generator::ItemFlags::from_raw(base_iflags);
-                let _ = self.emit_itemize(writer, &iflags, entry);
+                // Deferred on the run_pipelined path so this transfer row
+                // interleaves with directory rows in flist-index order at flush
+                // time; emitted immediately on every other path.
+                let _ = self.emit_or_record_itemize(writer, idx, &iflags, entry);
             }
             files_to_transfer.push((idx, entry, file_path, base_iflags));
         }
@@ -446,6 +450,7 @@ impl ReceiverContext {
     fn apply_no_change_metadata<W: Write + crate::writer::MsgInfoSender + ?Sized>(
         &self,
         writer: &mut W,
+        flist_idx: usize,
         file_path: &Path,
         entry: &FileEntry,
         stat_meta: &fs::Metadata,
@@ -465,7 +470,10 @@ impl ReceiverContext {
         // unless the itemize level requests unchanged rows (generator.c:574-576).
         if emit_itemize {
             let iflags = crate::generator::ItemFlags::from_raw(unchanged_iflags);
-            let _ = self.emit_itemize(writer, &iflags, entry);
+            // Deferred on the run_pipelined path so an up-to-date file's
+            // metadata-only row interleaves with directory and transfer rows in
+            // flist-index order at flush time; emitted immediately otherwise.
+            let _ = self.emit_or_record_itemize(writer, flist_idx, &iflags, entry);
         }
 
         // upstream: generator.c:461 unchanged_attrs() - fast-path check avoids
@@ -508,3 +516,153 @@ impl ReceiverContext {
         }
     }
 }
+
+#[cfg(test)]
+mod itemize_order_tests {
+    use std::ffi::OsString;
+
+    use protocol::ProtocolVersion;
+    use protocol::flist::FileEntry;
+
+    use crate::config::ServerConfig;
+    use crate::flags::{InfoFlags, ParsedServerFlags};
+    use crate::handshake::HandshakeResult;
+    use crate::receiver::ReceiverContext;
+    use crate::receiver::stats::TransferStats;
+    use crate::role::ServerRole;
+
+    fn handshake() -> HandshakeResult {
+        HandshakeResult {
+            protocol: ProtocolVersion::try_from(32u8).unwrap(),
+            buffered: Vec::new(),
+            compat_exchanged: false,
+            client_args: None,
+            io_timeout: None,
+            negotiated_algorithms: None,
+            compat_flags: None,
+            checksum_seed: 0,
+        }
+    }
+
+    /// A client-mode pull receiver with `-i` (itemize) requested.
+    fn itemize_client_config() -> ServerConfig {
+        let mut config = ServerConfig {
+            role: ServerRole::Receiver,
+            protocol: ProtocolVersion::try_from(32u8).unwrap(),
+            flag_string: "-ri".to_owned(),
+            flags: ParsedServerFlags {
+                recursive: true,
+                info_flags: InfoFlags {
+                    itemize: true,
+                    ..InfoFlags::default()
+                },
+                ..ParsedServerFlags::default()
+            },
+            args: vec![OsString::from(".")],
+            ..Default::default()
+        };
+        config.connection.client_mode = true;
+        config
+    }
+
+    /// The deferred flush must interleave directory and file itemize rows in
+    /// flist-index order (a dir row immediately precedes its children), not
+    /// batch every directory ahead of every file.
+    ///
+    /// Upstream itemizes in a single flist-index-order walk: `generate_files`
+    /// (generator.c:2329-2344) calls `recv_generator` per `cur_flist->sorted[i]`
+    /// in index order, and `recv_generator` (generator.c:1480-1483) itemizes
+    /// each directory at its own flist position. For the flist
+    /// `a/ a/f1 b/ b/f2` upstream prints `.d a/`, `>f a/f1`, `.d b/`, `>f b/f2`.
+    /// oc's two-phase receiver recorded every `.d` row in the directory-creation
+    /// pass before any `>f` row from the candidate pass, so a raw emission would
+    /// yield `.d a/`, `.d b/`, `>f a/f1`, `>f b/f2`. Keying each row by its flist
+    /// index and draining the BTreeMap in key order restores the interleave.
+    ///
+    /// This test drives the real `run_pipelined` record sites
+    /// (`create_directories` + `build_files_to_transfer`) with deferral on and
+    /// asserts the buffered rows land in index order. It fails if the batch
+    /// emission returns: reverting to an immediate emit leaves the buffer empty
+    /// (0 rows), and recording without the per-index key would order the two
+    /// directory rows ahead of the two file rows.
+    #[test]
+    fn deferred_itemize_rows_interleave_in_flist_index_order() {
+        let dir = test_support::create_tempdir();
+        let dest = dir.path();
+
+        let hs = handshake();
+        let mut ctx = ReceiverContext::new_for_test(&hs, itemize_client_config());
+        ctx.defer_itemize = true;
+        ctx.file_list = vec![
+            FileEntry::new_directory("a".into(), 0o755),  // idx 0
+            FileEntry::new_file("a/f1".into(), 5, 0o644), // idx 1
+            FileEntry::new_directory("b".into(), 0o755),  // idx 2
+            FileEntry::new_file("b/f2".into(), 5, 0o644), // idx 3
+        ];
+
+        let opts = metadata::MetadataOptions::default();
+        let mut writer = crate::writer::ServerWriter::new_plain(Vec::new());
+
+        // Directory-creation pass records the `.d` rows (flist indices 0 and 2).
+        ctx.create_directories(
+            dest,
+            &opts,
+            None,
+            None,
+            &mut writer,
+            #[cfg(unix)]
+            None,
+        )
+        .expect("create_directories succeeds");
+
+        // Candidate pass records the new-file transfer rows (indices 1 and 3).
+        let mut metadata_errors = Vec::new();
+        let mut stats = TransferStats::default();
+        let _ = ctx.build_files_to_transfer(
+            &mut writer,
+            dest,
+            &opts,
+            None,
+            &mut metadata_errors,
+            &mut stats,
+            None,
+            None,
+        );
+
+        let rows: Vec<(usize, String)> = ctx
+            .itemize_rows
+            .borrow()
+            .iter()
+            .map(|(idx, lines)| (*idx, lines[0].clone()))
+            .collect();
+
+        let keys: Vec<usize> = rows.iter().map(|(idx, _)| *idx).collect();
+        assert_eq!(
+            keys,
+            vec![0, 1, 2, 3],
+            "itemize rows must be keyed by flist index and drain in index order"
+        );
+
+        // Interleaved dir/file/dir/file, not batched dir/dir/file/file.
+        assert!(
+            rows[0].1.starts_with("cd") && rows[0].1.contains('a'),
+            "row 0 must be the created directory a/: {:?}",
+            rows[0].1
+        );
+        assert!(
+            rows[1].1.starts_with(">f") && rows[1].1.contains("a/f1"),
+            "row 1 must be the new file a/f1 (before b/), not the b/ directory: {:?}",
+            rows[1].1
+        );
+        assert!(
+            rows[2].1.starts_with("cd") && rows[2].1.contains('b'),
+            "row 2 must be the created directory b/ AFTER a/f1: {:?}",
+            rows[2].1
+        );
+        assert!(
+            rows[3].1.starts_with(">f") && rows[3].1.contains("b/f2"),
+            "row 3 must be the new file b/f2: {:?}",
+            rows[3].1
+        );
+    }
+}

crates/transfer/src/receiver/transfer/pipeline.rs

@@ -397,7 +397,12 @@ impl ReceiverContext {
                     if !self.config.connection.client_mode {
                         use crate::generator::ItemFlags;
                         let iflags = ItemFlags::from_raw(base_iflags);
-                        let _ = self.emit_itemize(writer, &iflags, file_entry);
+                        // Routed through the deferral seam for consistency with
+                        // the other emit sites. A server-mode receiver never
+                        // produces a client-visible row (record_itemize gates on
+                        // client_mode), so this stays the no-op it already was,
+                        // whether or not deferral is active.
+                        let _ = self.emit_or_record_itemize(writer, file_idx, &iflags, file_entry);
                     }
                 }
 

crates/transfer/src/receiver/transfer/pipelined.rs

@@ -37,6 +37,11 @@ impl ReceiverContext {
         pipeline_config: PipelineConfig,
     ) -> io::Result<TransferStats> {
         let _t = PhaseTimer::new("receiver-transfer-pipelined");
+        // Buffer itemize rows and flush them once in flist-index order before
+        // finalization, so a directory row immediately precedes its children
+        // (upstream's single flist-index-order walk, generator.c:2329-2344)
+        // rather than oc's two-phase "all dirs, then all files" emission.
+        self.defer_itemize = true;
         let (mut reader, file_count, mut setup) = self.setup_transfer(reader, writer)?;
         let reader = &mut reader;
 
@@ -264,6 +269,10 @@ impl ReceiverContext {
         // directory mtimes after file writes clobber them.
         self.touch_up_dirs(&setup.dest_dir);
 
+        // Drain the deferred itemize rows in flist-index order before the
+        // goodbye handshake, matching upstream's single-pass emission ordering.
+        self.flush_itemize_rows(writer)?;
+
         self.finalize_transfer(reader, writer)?;
 
         let total_source_bytes: u64 = self.file_list.iter().map(|e| e.size()).sum();