From 57a37576a5e7957774ead9c85321adca80e1f7c4 Mon Sep 17 00:00:00 2001 From: Ofer Chen Date: Mon, 13 Jul 2026 14:21:07 +0300 Subject: [PATCH] fix(transfer): interleave itemize rows in flist order (run_pipelined, step 1) The two-phase receiver emitted every directory (.d) itemize row in the directory-creation pass before any file (.f) row from the candidate pass, so `a/ a/f1 b/ b/f2` printed .d a/, .d b/, >f a/f1, >f b/f2. Upstream itemizes in a single flist-index-order walk (generator.c:2329-2344 -> recv_generator -> itemize, log.c:788 rwrite), where a directory row immediately precedes its children: .d a/, >f a/f1, .d b/, >f b/f2. Display-only: transferred data, wire bytes, and metadata application are unchanged; directories are still created before their files. Only the itemize emission moves to a deferred, flist-index-keyed flush. Adds a per-transfer buffer (itemize_rows: RefCell>>) plus record_itemize / flush_itemize_rows / a emit_or_record_itemize dispatcher, gated by an opt-in defer_itemize toggle. The dir/file/no-change emit sites in the shared create_directories, build_files_to_transfer, and pipeline loop now route through the dispatcher, which buffers when deferral is on and emits immediately otherwise. record_itemize preserves emit_itemize's client-mode visibility gate exactly (a server receiver's row still travels as wire iflags). Only run_pipelined opts in (sets defer_itemize and flushes before finalize); the sync, incremental, and async receive paths keep emitting immediately and are converted in follow-ups. Requires an lxhost -ii seal before merge. --- crates/transfer/src/receiver/context.rs | 23 ++- .../src/receiver/directory/creation.rs | 5 +- crates/transfer/src/receiver/itemize.rs | 73 ++++++++ .../src/receiver/transfer/candidates.rs | 162 +++++++++++++++++- .../src/receiver/transfer/pipeline.rs | 7 +- .../src/receiver/transfer/pipelined.rs | 9 + 6 files changed, 274 insertions(+), 5 deletions(-) diff --git a/crates/transfer/src/receiver/context.rs b/crates/transfer/src/receiver/context.rs index 6fe783387..4993d4504 100644 --- a/crates/transfer/src/receiver/context.rs +++ b/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>>, } 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()), } } diff --git a/crates/transfer/src/receiver/directory/creation.rs b/crates/transfer/src/receiver/directory/creation.rs index 18c6b3276..49158aae0 100644 --- a/crates/transfer/src/receiver/directory/creation.rs +++ b/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); } } diff --git a/crates/transfer/src/receiver/itemize.rs b/crates/transfer/src/receiver/itemize.rs index 5d4c38bc7..372d6b85e 100644 --- a/crates/transfer/src/receiver/itemize.rs +++ b/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( + &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( + &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(()) + } } diff --git a/crates/transfer/src/receiver/transfer/candidates.rs b/crates/transfer/src/receiver/transfer/candidates.rs index 7b00e2e8b..a18d1e910 100644 --- a/crates/transfer/src/receiver/transfer/candidates.rs +++ b/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( &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 = 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 + ); + } +} diff --git a/crates/transfer/src/receiver/transfer/pipeline.rs b/crates/transfer/src/receiver/transfer/pipeline.rs index c41b39ec2..83e728a80 100644 --- a/crates/transfer/src/receiver/transfer/pipeline.rs +++ b/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); } } diff --git a/crates/transfer/src/receiver/transfer/pipelined.rs b/crates/transfer/src/receiver/transfer/pipelined.rs index bb971bf17..fe3e192f0 100644 --- a/crates/transfer/src/receiver/transfer/pipelined.rs +++ b/crates/transfer/src/receiver/transfer/pipelined.rs @@ -37,6 +37,11 @@ impl ReceiverContext { pipeline_config: PipelineConfig, ) -> io::Result { 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();