Glue together an implementation for the pull responder

Cargo.lock

@@ -43,6 +43,12 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "ambient-authority"
+version = "0.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec8ad6edb4840b78c5c3d88de606b22252d552b55f3a4699fbb10fc070ec3049"
+
 [[package]]
 name = "anyhow"
 version = "1.0.57"
@@ -135,6 +141,50 @@ version = "1.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "14c189c53d098945499cdfa7ecc63567cf3886b3332b312a5b4585d8d3a6a610"
 
+[[package]]
+name = "cap-fs-ext"
+version = "0.24.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e54b86398b5852ddd45784b1d9b196b98beb39171821bad4b8b44534a1e87927"
+dependencies = [
+ "cap-primitives",
+ "cap-std",
+ "io-lifetimes",
+ "winapi",
+]
+
+[[package]]
+name = "cap-primitives"
+version = "0.24.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fb8fca3e81fae1d91a36e9784ca22a39ef623702b5f7904d89dc31f10184a178"
+dependencies = [
+ "ambient-authority",
+ "errno",
+ "fs-set-times",
+ "io-extras",
+ "io-lifetimes",
+ "ipnet",
+ "maybe-owned",
+ "rustix",
+ "winapi",
+ "winapi-util",
+ "winx",
+]
+
+[[package]]
+name = "cap-std"
+version = "0.24.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2247568946095c7765ad2b441a56caffc08027734c634a6d5edda648f04e32eb"
+dependencies = [
+ "cap-primitives",
+ "io-extras",
+ "io-lifetimes",
+ "ipnet",
+ "rustix",
+]
+
 [[package]]
 name = "cc"
 version = "1.0.73"
@@ -369,6 +419,7 @@ dependencies = [
  "hostname",
  "humansize",
  "indicatif",
+ "io-streams",
  "itertools 0.10.3",
  "libc",
  "log",
@@ -414,6 +465,12 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "duplex"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1178dea852a9bad9cb4b8e9111fbf0379a4c288e5cf4e14c36c87b85fe0cbbfa"
+
 [[package]]
 name = "ed25519"
 version = "1.5.2"
@@ -448,6 +505,27 @@ dependencies = [
  "termcolor",
 ]
 
+[[package]]
+name = "errno"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f639046355ee4f37944e44f60642c6f3a7efa3cf6b78c78a0d989a8ce6c396a1"
+dependencies = [
+ "errno-dragonfly",
+ "libc",
+ "winapi",
+]
+
+[[package]]
+name = "errno-dragonfly"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aa68f1b12764fab894d2755d2518754e71b4fd80ecfb822714a1206c2aab39bf"
+dependencies = [
+ "cc",
+ "libc",
+]
+
 [[package]]
 name = "fallible-iterator"
 version = "0.2.0"
@@ -476,6 +554,17 @@ dependencies = [
  "miniz_oxide",
 ]
 
+[[package]]
+name = "fs-set-times"
+version = "0.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7df62ee66ee2d532ea8d567b5a3f0d03ecd64636b98bad5be1e93dcc918b92aa"
+dependencies = [
+ "io-lifetimes",
+ "rustix",
+ "winapi",
+]
+
 [[package]]
 name = "fs2"
 version = "0.4.3"
@@ -622,6 +711,48 @@ dependencies = [
  "regex",
 ]
 
+[[package]]
+name = "io-extras"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0c937cc9891c12eaa8c63ad347e4a288364b1328b924886970b47a14ab8f8f8"
+dependencies = [
+ "io-lifetimes",
+ "os_pipe",
+ "winapi",
+]
+
+[[package]]
+name = "io-lifetimes"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec58677acfea8a15352d42fc87d11d63596ade9239e0a7c9352914417515dbe6"
+dependencies = [
+ "os_pipe",
+]
+
+[[package]]
+name = "io-streams"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fba6685e8e5efa7bd0ce8c4e92ac113b3b059b2e18bd1bf51f7cfab0f61b4b19"
+dependencies = [
+ "duplex",
+ "io-extras",
+ "io-lifetimes",
+ "memchr",
+ "os_pipe",
+ "parking",
+ "rustix",
+ "system-interface",
+]
+
+[[package]]
+name = "ipnet"
+version = "2.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "879d54834c8c76457ef4293a689b2a8c59b076067ad77b15efafbb05f92a592b"
+
 [[package]]
 name = "itertools"
 version = "0.9.0"
@@ -689,6 +820,12 @@ dependencies = [
  "vcpkg",
 ]
 
+[[package]]
+name = "linux-raw-sys"
+version = "0.0.42"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5284f00d480e1c39af34e72f8ad60b94f47007e3481cd3b731c1d67190ddc7b7"
+
 [[package]]
 name = "lock_api"
 version = "0.4.7"
@@ -714,6 +851,12 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ffbee8634e0d45d258acb448e7eaab3fce7a0a467395d4d9f228e3c1f01fb2e4"
 
+[[package]]
+name = "maybe-owned"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4facc753ae494aeb6e3c22f839b158aebd4f9270f55cd3c79906c45476c47ab4"
+
 [[package]]
 name = "memchr"
 version = "2.5.0"
@@ -861,12 +1004,28 @@ version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7709cef83f0c1f58f666e746a08b21e0085f7440fa6a29cc194d68aac97a4225"
 
+[[package]]
+name = "os_pipe"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c92f2b54f081d635c77e7120862d48db8e91f7f21cef23ab1b4fe9971c59f55"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "os_str_bytes"
 version = "6.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "21326818e99cfe6ce1e524c2a805c189a99b5ae555a35d19f9a284b427d86afa"
 
+[[package]]
+name = "parking"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "427c3892f9e783d91cc128285287e70a59e206ca452770ece88a76f7a3eddd72"
+
 [[package]]
 name = "parking_lot"
 version = "0.12.0"
@@ -1071,6 +1230,22 @@ dependencies = [
  "smallvec",
 ]
 
+[[package]]
+name = "rustix"
+version = "0.33.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "938a344304321a9da4973b9ff4f9f8db9caf4597dfd9dda6a60b523340a0fff0"
+dependencies = [
+ "bitflags",
+ "errno",
+ "io-lifetimes",
+ "itoa",
+ "libc",
+ "linux-raw-sys",
+ "once_cell",
+ "winapi",
+]
+
 [[package]]
 name = "rustversion"
 version = "1.0.6"
@@ -1210,6 +1385,19 @@ version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f2dd574626839106c320a323308629dcb1acfc96e32a8cba364ddc61ac23ee83"
 
+[[package]]
+name = "socketpair"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2f0c0b3fc17356799222affc5a40345b7cc25b548c489c5a31eca0888ee2404c"
+dependencies = [
+ "io-extras",
+ "io-lifetimes",
+ "rustix",
+ "uuid",
+ "winapi",
+]
+
 [[package]]
 name = "sodiumoxide"
 version = "0.2.7"
@@ -1264,6 +1452,23 @@ dependencies = [
  "unicode-ident",
 ]
 
+[[package]]
+name = "system-interface"
+version = "0.20.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e09bb3fb4e02ec4b87e182ea9718fadbc0fa3e50085b40a9af9690572b67f9e"
+dependencies = [
+ "atty",
+ "bitflags",
+ "cap-fs-ext",
+ "io-lifetimes",
+ "os_pipe",
+ "rustix",
+ "socketpair",
+ "winapi",
+ "winx",
+]
+
 [[package]]
 name = "temp-dir"
 version = "0.1.11"
@@ -1400,6 +1605,15 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "936e4b492acfd135421d8dca4b1aa80a7bfc26e702ef3af710e0752684df5372"
 
+[[package]]
+name = "uuid"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc5cf98d8186244414c848017f0e2676b3fcb46807f6668a97dfe67359a3c4b7"
+dependencies = [
+ "getrandom",
+]
+
 [[package]]
 name = "vcpkg"
 version = "0.2.15"
@@ -1515,6 +1729,17 @@ version = "0.36.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680"
 
+[[package]]
+name = "winx"
+version = "0.31.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08d5973cb8cd94a77d03ad7e23bbe14889cb29805da1cec0e4aff75e21aebded"
+dependencies = [
+ "bitflags",
+ "io-lifetimes",
+ "winapi",
+]
+
 [[package]]
 name = "yama"
 version = "0.6.0-alpha.3"

datman/Cargo.toml

@@ -35,3 +35,4 @@ metrics = "0.17.1"
 bare-metrics-recorder = { version = "0.1.0" }
 comfy-table = "6.0.0-rc.1"
 libc = "0.2.126"
+io-streams = "0.11.0"

datman/src/bin/datman.rs

@@ -16,6 +16,7 @@ along with Yama.  If not, see <https://www.gnu.org/licenses/>.
 */
 
 use std::fs::File;
+use std::io::{BufReader, BufWriter, Write};
 use std::path::{Path, PathBuf};
 
 use clap::Parser;
@@ -26,7 +27,7 @@ use bare_metrics_recorder::recording::BareMetricsRecorderCore;
 use chrono::{DateTime, Local, NaiveDate, NaiveDateTime, TimeZone, Utc};
 use datman::commands::backup::{backup_all_sources_to_destination, backup_source_to_destination};
 use datman::commands::ilabel::interactive_labelling_session;
-use datman::commands::init_descriptor;
+use datman::commands::{init_descriptor, pushpull};
 use datman::descriptor::{load_descriptor, SourceDescriptor};
 use datman::get_hostname;
 use datman::remote::backup_source_requester::backup_remote_source_to_destination;
@@ -34,6 +35,8 @@ use datman::remote::backup_source_responder;
 use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle};
 use log::info;
 use std::str::FromStr;
+use yama::commands::load_pile_descriptor;
+use yama::operations::legacy_pushpull::{open_pile_with_work_bypass, BypassLevel};
 
 pub const FAILURE_SYMBOL_OBNOXIOUS_FLASHING: &str = "\x1b[5m\x1b[31m⚠️ \x1b[25m\x1b[22m";
 pub const BOLD: &str = "\x1b[1m";
@@ -122,6 +125,18 @@ pub enum DatmanCommand {
 
     #[clap(name = "_backup_source_responder")]
     InternalBackupSourceResponder,
+
+    /// Pulls all pointers from a remote pile to a local pile.
+    /// Does not yet support label filtering, but will do in the future.
+    Pull {
+        /// e.g. 'myserver:main'
+        remote_and_remote_pile: String,
+
+        pile_name: String,
+    },
+
+    #[clap(name = "_pull_responder_offerer")]
+    InternalPullResponderOfferer { pile_name: String },
 }
 
 pub struct HumanDateTime(pub DateTime<Local>);
@@ -329,6 +344,31 @@ fn main() -> anyhow::Result<()> {
             datman::commands::report::print_filesystem_space(&destination.path)?;
             datman::commands::report::print_report(&report)?;
         }
+        DatmanCommand::Pull { .. } => {}
+        DatmanCommand::InternalPullResponderOfferer { pile_name } => {
+            let descriptor = load_descriptor(Path::new(".")).unwrap();
+            let source = &descriptor.piles[&pile_name];
+
+            let pile_desc = load_pile_descriptor(&source.path)?;
+            let (pile, bypass_raw_pile) = open_pile_with_work_bypass(
+                &source.path,
+                &pile_desc,
+                BypassLevel::CompressionBypass,
+            )?;
+
+            let mut stdin = BufReader::new(io_streams::StreamReader::stdin()?);
+            let mut stdout = BufWriter::new(io_streams::StreamWriter::stdout()?);
+
+            pushpull::offering_side(
+                &pile,
+                &bypass_raw_pile,
+                &mut stdin,
+                &mut stdout,
+                Box::new(()),
+            )?;
+
+            stdout.flush()?;
+        }
     }
     Ok(())
 }

datman/src/commands/pushpull.rs

@@ -3,6 +3,7 @@
 use anyhow::{bail, ensure, Context};
 use std::collections::{BTreeMap, BTreeSet};
 use std::io::{Read, Write};
+use std::sync::Arc;
 use yama::chunking::RecursiveUnchunker;
 use yama::commands::retrieve_tree_node;
 use yama::definitions::{ChunkId, PointerData, RecursiveChunkRef, TreeNode};
@@ -57,7 +58,7 @@ pub fn ensure_compatible_bypasses(
 }
 
 pub fn negotiate_bypassed_pile<R: Read, W: Write>(
-    pile: &Pile<Box<dyn RawPile>>,
+    pile: &Pile<Arc<Box<dyn RawPile>>>,
     bypass_pile: &Box<dyn RawPile>,
     reader: &mut R,
     writer: &mut W,
@@ -82,7 +83,7 @@ pub fn negotiate_bypassed_pile<R: Read, W: Write>(
 }
 
 fn collect_chunk_ids(
-    pile: &Pile<Box<dyn RawPile>>,
+    pile: &Pile<Arc<Box<dyn RawPile>>>,
     root: &TreeNode,
     chunk_ids: &mut BTreeSet<ChunkId>,
 ) -> anyhow::Result<()> {
@@ -102,7 +103,7 @@ fn collect_chunk_ids(
 }
 
 fn collect_chunk_ids_from_chunkref(
-    pile: &Pile<Box<dyn RawPile>>,
+    pile: &Pile<Arc<Box<dyn RawPile>>>,
     chunk_ref: &RecursiveChunkRef,
     collection: &mut BTreeSet<ChunkId>,
 ) -> anyhow::Result<()> {
@@ -129,8 +130,8 @@ fn collect_chunk_ids_from_chunkref(
 }
 
 pub fn offering_side<R: Read, W: Write>(
-    pile: &Pile<Box<dyn RawPile>>,
+    pile: &Pile<Arc<Box<dyn RawPile>>>,
-    bypass_pile: Box<dyn RawPile>,
+    bypass_pile: &Box<dyn RawPile>,
     reader: &mut R,
     writer: &mut W,
     mut progress: Box<dyn ProgressTracker>,
@@ -182,7 +183,7 @@ pub fn offering_side<R: Read, W: Write>(
 }
 
 pub fn accepting_side<R: Read, W: Write>(
-    pile: &Pile<Box<dyn RawPile>>,
+    pile: &Pile<Arc<Box<dyn RawPile>>>,
     bypass_pile: Box<dyn RawPile>,
     reader: &mut R,
     writer: &mut W,