Enable fallback delegation for batch workflows

crates/cli/src/frontend/arguments/parsed_args.rs

@@ -87,6 +87,9 @@ pub(crate) struct ParsedArgs {
     pub(crate) temp_dir: Option<PathBuf>,
     pub(crate) log_file: Option<OsString>,
     pub(crate) log_file_format: Option<OsString>,
+    pub(crate) write_batch: Option<OsString>,
+    pub(crate) only_write_batch: Option<OsString>,
+    pub(crate) read_batch: Option<OsString>,
     pub(crate) link_dests: Vec<PathBuf>,
     pub(crate) remove_source_files: bool,
     pub(crate) inplace: Option<bool>,

crates/cli/src/frontend/arguments/parser.rs

@@ -309,6 +309,9 @@ where
         .map(PathBuf::from);
     let log_file = matches.remove_one::<OsString>("log-file");
     let log_file_format = matches.remove_one::<OsString>("log-file-format");
+    let write_batch = matches.remove_one::<OsString>("write-batch");
+    let only_write_batch = matches.remove_one::<OsString>("only-write-batch");
+    let read_batch = matches.remove_one::<OsString>("read-batch");
     let link_dest_args: Vec<OsString> = matches
         .remove_many::<OsString>("link-dest")
         .map(|values| values.collect())
@@ -540,6 +543,9 @@ where
         temp_dir,
         log_file,
         log_file_format,
+        write_batch,
+        only_write_batch,
+        read_batch,
         link_dests,
         remove_source_files,
         inplace,

crates/cli/src/frontend/command_builder/sections/section_02.rs

@@ -32,6 +32,30 @@ pub(crate) fn section_02(command: ClapCommand) -> ClapCommand {
                     .help("Customise the format used when appending to --log-file.")
                     .value_parser(OsStringValueParser::new()),
             )
+            .arg(
+                Arg::new("write-batch")
+                    .long("write-batch")
+                    .value_name("PREFIX")
+                    .help("Store updated data in batch files named PREFIX for later replay.")
+                    .value_parser(OsStringValueParser::new())
+                    .conflicts_with_all(["read-batch", "only-write-batch"]),
+            )
+            .arg(
+                Arg::new("only-write-batch")
+                    .long("only-write-batch")
+                    .value_name("PREFIX")
+                    .help("Write batch files named PREFIX without applying the updates locally.")
+                    .value_parser(OsStringValueParser::new())
+                    .conflicts_with_all(["read-batch", "write-batch"]),
+            )
+            .arg(
+                Arg::new("read-batch")
+                    .long("read-batch")
+                    .value_name("PREFIX")
+                    .help("Apply updates stored in batch files named PREFIX.")
+                    .value_parser(OsStringValueParser::new())
+                    .conflicts_with_all(["write-batch", "only-write-batch"]),
+            )
             .arg(
                 Arg::new("whole-file")
                     .long("whole-file")

crates/cli/src/frontend/execution/drive/config.rs

@@ -84,6 +84,7 @@ pub(crate) struct ConfigInputs {
     pub(crate) append: bool,
     pub(crate) append_verify: bool,
     pub(crate) whole_file: bool,
+    pub(crate) force_fallback: bool,
     pub(crate) timeout: TransferTimeout,
     pub(crate) connect_timeout: TransferTimeout,
     pub(crate) stop_deadline: Option<SystemTime>,
@@ -175,6 +176,7 @@ pub(crate) fn build_base_config(mut inputs: ConfigInputs) -> ClientConfigBuilder
         .append(inputs.append)
         .append_verify(inputs.append_verify)
         .whole_file(inputs.whole_file)
+        .force_fallback(inputs.force_fallback)
         .timeout(inputs.timeout)
         .connect_timeout(inputs.connect_timeout)
         .stop_at(inputs.stop_deadline);

crates/cli/src/frontend/execution/drive/fallback.rs

@@ -126,6 +126,9 @@ pub(crate) struct FallbackInputs {
     #[cfg(feature = "xattr")]
     pub(crate) xattrs: Option<bool>,
     pub(crate) itemize_changes: bool,
+    pub(crate) write_batch: Option<OsString>,
+    pub(crate) only_write_batch: Option<OsString>,
+    pub(crate) read_batch: Option<OsString>,
 }
 
 /// Builds the remote fallback arguments when required.
@@ -265,6 +268,9 @@ where
         fallback_binary: None,
         rsync_path: inputs.rsync_path,
         remainder: inputs.remainder,
+        write_batch: inputs.write_batch,
+        only_write_batch: inputs.only_write_batch,
+        read_batch: inputs.read_batch,
         #[cfg(feature = "acl")]
         acls: inputs.acls,
         #[cfg(feature = "xattr")]

crates/cli/src/frontend/execution/drive/workflow/fallback_plan.rs

@@ -121,6 +121,9 @@ pub(crate) struct FallbackArgumentsContext<'a> {
     pub(crate) itemize_changes: bool,
     pub(crate) log_file_path: Option<&'a PathBuf>,
     pub(crate) log_file_format: Option<&'a OsString>,
+    pub(crate) write_batch: Option<&'a OsString>,
+    pub(crate) only_write_batch: Option<&'a OsString>,
+    pub(crate) read_batch: Option<&'a OsString>,
 }
 
 pub(crate) fn build_fallback_arguments<Err>(
@@ -256,6 +259,9 @@ where
         address_mode: context.address_mode,
         rsync_path: context.rsync_path.cloned(),
         remainder: context.remainder.to_vec(),
+        write_batch: context.write_batch.cloned(),
+        only_write_batch: context.only_write_batch.cloned(),
+        read_batch: context.read_batch.cloned(),
         #[cfg(feature = "acl")]
         acls: context.acls,
         #[cfg(feature = "xattr")]

crates/cli/src/frontend/execution/drive/workflow/mod.rs

@@ -138,6 +138,9 @@ where
         temp_dir,
         log_file,
         log_file_format,
+        write_batch,
+        only_write_batch,
+        read_batch,
         link_dests,
         remove_source_files,
         inplace,
@@ -330,8 +333,10 @@ where
     let files_from_used = !files_from.is_empty();
     let implied_dirs_option = implied_dirs;
     let implied_dirs = implied_dirs_option.unwrap_or(true);
+    let batch_mode_requested =
+        write_batch.is_some() || only_write_batch.is_some() || read_batch.is_some();
     let requires_remote_fallback = transfer_requires_remote(&remainder, &file_list_operands);
-    let fallback_required = requires_remote_fallback;
+    let fallback_required = requires_remote_fallback || batch_mode_requested;
 
     let fallback_context = FallbackArgumentsContext {
         required: fallback_required,
@@ -440,6 +445,9 @@ where
         itemize_changes,
         log_file_path: log_file_path_buf.as_ref(),
         log_file_format: log_file_format_cli.as_ref(),
+        write_batch: write_batch.as_ref(),
+        only_write_batch: only_write_batch.as_ref(),
+        read_batch: read_batch.as_ref(),
     };
     let fallback_args = match build_fallback_arguments(fallback_context, stderr) {
         Ok(args) => args,
@@ -609,6 +617,7 @@ where
         append: append_enabled,
         append_verify,
         whole_file: whole_file_enabled,
+        force_fallback: batch_mode_requested,
         timeout: timeout_setting,
         connect_timeout: connect_timeout_setting,
         stop_deadline: stop_request.as_ref().map(StopRequest::deadline),

crates/cli/src/frontend/tests/local.rs

@@ -45,3 +45,45 @@ exit 0
     let recorded = std::fs::read_to_string(&args_path).expect("read args file");
     assert_eq!(recorded, "untouched");
 }
+
+#[cfg(unix)]
+#[test]
+fn local_write_batch_forces_fallback_path() {
+    use tempfile::tempdir;
+
+    let _env_lock = ENV_LOCK.lock().expect("env lock");
+    let _rsh_guard = clear_rsync_rsh();
+    let temp = tempdir().expect("tempdir");
+    let script_path = temp.path().join("fallback.sh");
+    let args_path = temp.path().join("args.txt");
+
+    let script = r#"#!/bin/sh
+printf "%s\n" "$@" > "$ARGS_FILE"
+exit 0
+"#;
+    write_executable_script(&script_path, script);
+
+    let _fallback_guard = EnvGuard::set(CLIENT_FALLBACK_ENV, script_path.as_os_str());
+    let _args_guard = EnvGuard::set("ARGS_FILE", args_path.as_os_str());
+
+    std::fs::write(&args_path, b"").expect("truncate args file");
+
+    let source = temp.path().join("source.txt");
+    let destination = temp.path().join("dest.txt");
+    std::fs::write(&source, b"contents").expect("write source");
+
+    let (code, stdout, stderr) = run_with_args([
+        OsString::from(RSYNC),
+        OsString::from("--write-batch=batch"),
+        source.into_os_string(),
+        destination.into_os_string(),
+    ]);
+
+    assert_eq!(code, 0);
+    assert!(stdout.is_empty());
+    assert!(stderr.is_empty());
+
+    let recorded = std::fs::read_to_string(&args_path).expect("read args file");
+    let args: Vec<&str> = recorded.lines().collect();
+    assert!(args.contains(&"--write-batch=batch"));
+}

crates/cli/src/frontend/tests/mod.rs

@@ -104,6 +104,8 @@ mod parse_args_reads_tests;
 mod parse_args_recognises_append_tests;
 #[path = "parse_args_recognises_archive.rs"]
 mod parse_args_recognises_archive_tests;
+#[path = "parse_args_recognises_batch.rs"]
+mod parse_args_recognises_batch_tests;
 #[path = "parse_args_recognises_block_size.rs"]
 mod parse_args_recognises_block_size_tests;
 #[path = "parse_args_recognises_checksum.rs"]
@@ -232,6 +234,8 @@ mod remote_fallback_forwards_acls_tests;
 mod remote_fallback_forwards_append_tests;
 #[path = "remote_fallback_forwards_backup.rs"]
 mod remote_fallback_forwards_backup_tests;
+#[path = "remote_fallback_forwards_batch.rs"]
+mod remote_fallback_forwards_batch_tests;
 #[path = "remote_fallback_forwards_compress.rs"]
 mod remote_fallback_forwards_compress_tests;
 #[path = "remote_fallback_forwards_connect.rs"]

crates/cli/src/frontend/tests/parse_args_recognises_batch.rs

@@ -0,0 +1,47 @@
+use super::common::*;
+use super::*;
+
+#[test]
+fn parse_args_recognises_write_batch_prefix() {
+    let parsed = parse_args([
+        OsString::from(RSYNC),
+        OsString::from("--write-batch=updates"),
+        OsString::from("source"),
+        OsString::from("dest"),
+    ])
+    .expect("parse");
+
+    assert_eq!(parsed.write_batch, Some(OsString::from("updates")));
+    assert!(parsed.only_write_batch.is_none());
+    assert!(parsed.read_batch.is_none());
+}
+
+#[test]
+fn parse_args_recognises_only_write_batch_prefix() {
+    let parsed = parse_args([
+        OsString::from(RSYNC),
+        OsString::from("--only-write-batch=batch"),
+        OsString::from("source"),
+        OsString::from("dest"),
+    ])
+    .expect("parse");
+
+    assert_eq!(parsed.only_write_batch, Some(OsString::from("batch")));
+    assert!(parsed.write_batch.is_none());
+    assert!(parsed.read_batch.is_none());
+}
+
+#[test]
+fn parse_args_recognises_read_batch_prefix() {
+    let parsed = parse_args([
+        OsString::from(RSYNC),
+        OsString::from("--read-batch=replay"),
+        OsString::from("source"),
+        OsString::from("dest"),
+    ])
+    .expect("parse");
+
+    assert_eq!(parsed.read_batch, Some(OsString::from("replay")));
+    assert!(parsed.write_batch.is_none());
+    assert!(parsed.only_write_batch.is_none());
+}

crates/cli/src/frontend/tests/remote_fallback_forwards_batch.rs

@@ -0,0 +1,74 @@
+use super::common::*;
+use super::*;
+
+#[cfg(unix)]
+#[test]
+fn remote_fallback_forwards_batch_arguments() {
+    use tempfile::tempdir;
+
+    let _env_lock = ENV_LOCK.lock().expect("env lock");
+    let _rsh_guard = clear_rsync_rsh();
+    let temp = tempdir().expect("tempdir");
+    let script_path = temp.path().join("fallback.sh");
+    let args_path = temp.path().join("args.txt");
+
+    let script = r#"#!/bin/sh
+printf "%s\n" "$@" > "$ARGS_FILE"
+exit 0
+"#;
+    write_executable_script(&script_path, script);
+
+    let _fallback_guard = EnvGuard::set(CLIENT_FALLBACK_ENV, script_path.as_os_str());
+    let _args_guard = EnvGuard::set("ARGS_FILE", args_path.as_os_str());
+
+    let dest = temp.path().join("dest");
+
+    let (code, stdout, stderr) = run_with_args([
+        OsString::from(RSYNC),
+        OsString::from("--write-batch=batch"),
+        OsString::from("remote::module"),
+        dest.clone().into_os_string(),
+    ]);
+
+    assert_eq!(code, 0);
+    assert!(stdout.is_empty());
+    assert!(stderr.is_empty());
+
+    let recorded = std::fs::read_to_string(&args_path).expect("read args file");
+    let args: Vec<&str> = recorded.lines().collect();
+    assert!(args.contains(&"--write-batch=batch"));
+
+    std::fs::write(&args_path, b"").expect("truncate args file");
+
+    let (code, stdout, stderr) = run_with_args([
+        OsString::from(RSYNC),
+        OsString::from("--only-write-batch=batch"),
+        OsString::from("remote::module"),
+        dest.clone().into_os_string(),
+    ]);
+
+    assert_eq!(code, 0);
+    assert!(stdout.is_empty());
+    assert!(stderr.is_empty());
+
+    let recorded = std::fs::read_to_string(&args_path).expect("read args file");
+    let args: Vec<&str> = recorded.lines().collect();
+    assert!(args.contains(&"--only-write-batch=batch"));
+
+    std::fs::write(&args_path, b"").expect("truncate args file");
+
+    let (code, stdout, stderr) = run_with_args([
+        OsString::from(RSYNC),
+        OsString::from("--read-batch=batch"),
+        OsString::from("remote::module"),
+        dest.into_os_string(),
+    ]);
+
+    assert_eq!(code, 0);
+    assert!(stdout.is_empty());
+    assert!(stderr.is_empty());
+
+    let recorded = std::fs::read_to_string(&args_path).expect("read args file");
+    let args: Vec<&str> = recorded.lines().collect();
+    assert!(args.contains(&"--read-batch=batch"));
+}

crates/core/src/client/config/builder/fallback.rs

@@ -0,0 +1,17 @@
+use super::*;
+
+impl ClientConfigBuilder {
+    /// Forces the client orchestration to delegate to the legacy rsync binary.
+    ///
+    /// The native engine does not yet support batch file generation or replay,
+    /// so the CLI triggers delegation when `--write-batch`,
+    /// `--only-write-batch`, or `--read-batch` is supplied. Setting this flag
+    /// ensures [`run_client_or_fallback`](crate::client::run_client_or_fallback)
+    /// invokes the fallback even when the local plan would otherwise be
+    /// executable.
+    #[must_use]
+    pub const fn force_fallback(mut self, force: bool) -> Self {
+        self.force_fallback = force;
+        self
+    }
+}

crates/core/src/client/config/builder/mod.rs

@@ -80,6 +80,7 @@ pub struct ClientConfigBuilder {
     append: bool,
     append_verify: bool,
     force_event_collection: bool,
+    force_fallback: bool,
     preserve_devices: bool,
     preserve_specials: bool,
     list_only: bool,
@@ -166,6 +167,7 @@ impl ClientConfigBuilder {
             append: self.append,
             append_verify: self.append_verify,
             force_event_collection: self.force_event_collection,
+            force_fallback: self.force_fallback,
             preserve_devices: self.preserve_devices,
             preserve_specials: self.preserve_specials,
             list_only: self.list_only,
@@ -187,6 +189,7 @@ impl ClientConfigBuilder {
 
 mod arguments;
 mod deletion;
+mod fallback;
 mod filters;
 mod metadata;
 mod network;

crates/core/src/client/config/client/fallback.rs

@@ -0,0 +1,9 @@
+use super::*;
+
+impl ClientConfig {
+    /// Reports whether the client must delegate to the legacy rsync binary.
+    #[must_use]
+    pub const fn force_fallback(&self) -> bool {
+        self.force_fallback
+    }
+}