diff --git a/src/device/mod.rs b/src/device/mod.rs index bd60a35d..0118e65a 100644 --- a/src/device/mod.rs +++ b/src/device/mod.rs @@ -1,4 +1,4 @@ -//! Mounted portable-device discovery. +//! Mounted portable-device discovery and transfer. //! //! GIO's native [`gtk::gio::VolumeMonitor`] supplies a cached snapshot of the //! user-visible mounts selected by each platform backend. The UI owns that @@ -6,10 +6,16 @@ //! and wires its mount-added, changed, pre-unmount, and removed signals for live //! hotplug updates. Filesystem traversal remains separate background work. //! -//! This layer currently supports browsing mounted filesystems. Device sync, -//! transfer, MTP-only access, and mounting an unmounted volume remain outside -//! its scope; see GitHub issue #8 and `docs/roadmap.md`. +//! The [`transfer`] module adds a generic mounted-filesystem transfer planner +//! and executor that satisfies the P3.2 / GitHub issue #8 requirements: +//! retained write authority, capacity and conflict policy, atomic copy where +//! possible, progress reporting, cancellation, and rollback. It builds on the +//! same root-lease model used by the read paths in +//! [`crate::local::root_authority`] and +//! [`crate::local::write_authority`]. +pub mod mtp; +pub mod transfer; pub mod usb; /// Information about one mounted, browseable device. diff --git a/src/device/mtp/browse.rs b/src/device/mtp/browse.rs new file mode 100644 index 00000000..93b49cbc --- /dev/null +++ b/src/device/mtp/browse.rs @@ -0,0 +1,373 @@ +//! Bounded MTP storage-object browse. +//! +//! MTP storage objects form a tree; the tree is bounded by the +//! storage's capacity but a malicious or buggy device could return an +//! arbitrarily deep or wide tree. The browser in this module is the +//! only place the rest of the system can enumerate an MTP tree, and it +//! is bounded by an explicit [`BrowseBudget`] supplied by the caller. +//! +//! The browser walks the tree depth-first, parent-first, and is +//! strictly advisory: it never fetches object bytes, never opens a +//! destination, and never commits a write. The browser's output is a +//! list of [`MtpObject`]s that the planner can use to assemble a +//! [`TransferPlan`](super::super::transfer::TransferPlan). + +use std::collections::BTreeMap; +use std::collections::BTreeSet; + +use serde::{Deserialize, Serialize}; + +use super::transport::{MtpObjectHandle, MtpSession, MtpTransportError}; + +/// What kind of object an [`MtpObject`] is. Mirrors the MTP object +/// format codes the transport layer surfaces, narrowed to the subset +/// the planner can act on. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] +pub enum MtpObjectKind { + /// A regular file. The planner can fetch it. + RegularFile, + /// A folder. The planner descends into it subject to its budget. + Folder, + /// Any other object type (associations, playlists, abstract + /// media). The planner ignores these. + Other, +} + +/// One MTP storage object as observed by the browser. +/// +/// The object is described by an [`MtpObjectHandle`] — its portable +/// identity on the device. The [`parent`](Self::parent) chain lets the +/// planner reconstruct a relative path without consulting any host +/// filesystem. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct MtpObject { + pub handle: MtpObjectHandle, + pub parent: Option, + pub name: String, + pub kind: MtpObjectKind, + pub size_bytes: u64, +} + +/// The closure the browser uses to enumerate children of a given +/// object handle. The transport supplies one because the transport is +/// the only place that can talk to the device. The trait object is +/// boxed so callers can pass an owned closure with a `move` capture +/// without worrying about the borrow's lifetime. +type ListChildren<'a> = + dyn FnMut(&MtpSession, MtpObjectHandle) -> Result, MtpTransportError> + 'a; + +/// The browser. Stateless and `Clone` so the same browser can be reused +/// across multiple storage areas and so the planner can run tests +/// against it without sharing state. +#[derive(Clone, Debug, Default)] +pub struct MtpBrowser; + +impl MtpBrowser { + /// Create a new browser instance. + pub fn new() -> Self { + Self + } + + /// Browse one storage area starting at the given root handle. + /// + /// The browser walks the tree parent-first, depth-first, and stops + /// as soon as either the entry count or the depth bound is + /// exhausted. The returned list is in the order the tree was + /// walked; the planner can use the parent chain to reconstruct a + /// depth-ordered path without depending on iteration order. + /// + /// `list_children` is supplied by the transport because the + /// transport is the only place that can talk to the device. The + /// transport never sees a destination or a host path. + #[allow(clippy::unused_self, clippy::needless_lifetimes)] + pub fn browse<'a>( + &'a self, + session: &'a MtpSession, + root: MtpObjectHandle, + budget: BrowseBudget, + list_children: &mut ListChildren<'a>, + ) -> Result, MtpTransportError> { + session.verify()?; + + let mut visited: BTreeSet = BTreeSet::new(); + let mut by_handle: BTreeMap = BTreeMap::new(); + let mut result: Vec = Vec::new(); + let mut pending: Vec<(MtpObjectHandle, u32)> = vec![(root, 0)]; + + while let Some((handle, depth)) = pending.pop() { + if !visited.insert(handle) { + continue; + } + if result.len() as u64 >= budget.max_entries() { + break; + } + if depth > budget.max_depth() { + continue; + } + let children = list_children(session, handle)?; + let next_depth = depth.saturating_add(u32::from(budget.allows_recursion())); + for child in children { + let kind = child.kind; + let child_handle = child.handle; + let child_parent = child.parent; + if by_handle.insert(child_handle, child.clone()).is_none() { + result.push(child); + } + if matches!(kind, MtpObjectKind::Folder) && budget.allows_recursion() { + pending.push((child_handle, next_depth)); + } + let _ = child_parent; + } + } + + Ok(result) + } +} + +/// Hard bounds on a single browse call. +/// +/// `max_entries` and `max_depth` are both inclusive. A budget with +/// `max_depth == 0` returns the children of the root only; a budget +/// with `max_entries == 0` returns nothing. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct BrowseBudget { + max_entries: u64, + max_depth: u32, +} + +impl BrowseBudget { + /// Construct a budget. `max_entries` of `0` is allowed and means + /// "return nothing"; `max_depth` of `0` means "do not descend into + /// folders." + pub fn new(max_entries: u64, max_depth: u32) -> Self { + Self { + max_entries, + max_depth, + } + } + + /// Upper bound on the number of objects the browser will return. + pub fn max_entries(&self) -> u64 { + self.max_entries + } + + /// Upper bound on the depth of the tree the browser will descend + /// into. + pub fn max_depth(&self) -> u32 { + self.max_depth + } + + /// Whether the browser is allowed to descend into folders. A budget + /// with `max_depth == 0` is the only configuration where this is + /// `false`. + pub fn allows_recursion(&self) -> bool { + self.max_depth > 0 + } +} + +impl Default for BrowseBudget { + fn default() -> Self { + Self::new(8 * 1024, 6) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::device::mtp::transport::test_transport::{InMemoryMtpTransport, InMemoryObject}; + use crate::device::mtp::transport::MtpTransport; + + fn descriptor(serial: &str) -> crate::device::mtp::MtpUsbDescriptor { + crate::device::mtp::MtpUsbDescriptor::new(0x04e8, 0x6860, serial).expect("descriptor") + } + + fn folder(handle: u32, parent: Option, name: &str) -> InMemoryObject { + InMemoryObject { + handle: MtpObjectHandle(handle), + parent: parent.map(MtpObjectHandle), + name: name.to_string(), + kind: MtpObjectKind::Folder, + size_bytes: 0, + bytes: Vec::new(), + } + } + + fn file(handle: u32, parent: u32, name: &str, payload: &[u8]) -> InMemoryObject { + InMemoryObject { + handle: MtpObjectHandle(handle), + parent: Some(MtpObjectHandle(parent)), + name: name.to_string(), + kind: MtpObjectKind::RegularFile, + size_bytes: payload.len() as u64, + bytes: payload.to_vec(), + } + } + + fn build_tree(transport: &InMemoryMtpTransport) -> (MtpObjectHandle, MtpObjectHandle) { + // Storage root handle is conventionally 0x0000_0001. Place a + // top-level folder "Music" and "Photos" beneath it; "Music" + // contains a subfolder and a file; "Photos" contains only a + // file. + let root = MtpObjectHandle(0x0000_0001); + let music = folder(0x0000_0010, Some(root.0), "Music"); + let photos = folder(0x0000_0011, Some(root.0), "Photos"); + let tracks = folder(0x0000_0012, Some(music.handle.0), "Tracks"); + let song = file(0x0000_0020, music.handle.0, "song.flac", b"flac"); + let deep = file(0x0000_0021, tracks.handle.0, "deep.flac", b"deep"); + let photo = file(0x0000_0022, photos.handle.0, "img.jpg", b"jpg"); + transport.add_object(0, 0, music.clone()); + transport.add_object(0, 0, photos.clone()); + transport.add_object(0, 0, tracks.clone()); + transport.add_object(0, 0, song); + transport.add_object(0, 0, deep); + transport.add_object(0, 0, photo); + // The transport's storage already advertises the storage + // descriptor; rewrite it to include the synthetic root for the + // test. + (root, MtpObjectHandle(music.handle.0)) + } + + #[test] + fn browse_returns_descendants_within_budget() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let (root, _music) = build_tree(&transport); + let session = transport + .open_session(&descriptor("ABC123")) + .expect("session"); + let browser = MtpBrowser::new(); + let objects = { + let mut list_children = build_list_children(&transport, root); + browser.browse(&session, root, BrowseBudget::new(64, 4), &mut list_children) + } + .expect("browse"); + let names: Vec<&str> = objects.iter().map(|o| o.name.as_str()).collect(); + assert!(names.contains(&"Music")); + assert!(names.contains(&"song.flac")); + // Depth bound of 4 admits the deep file under Music/Tracks. + assert!(names.contains(&"deep.flac")); + } + + fn build_list_children( + transport: &InMemoryMtpTransport, + root: MtpObjectHandle, + ) -> impl FnMut(&MtpSession, MtpObjectHandle) -> Result, MtpTransportError> + '_ + { + move |session: &MtpSession, parent: MtpObjectHandle| { + let _ = transport.list_storage(session).expect("storage"); + let mut children = Vec::new(); + for handle in [ + MtpObjectHandle(0x0000_0010), + MtpObjectHandle(0x0000_0011), + MtpObjectHandle(0x0000_0012), + MtpObjectHandle(0x0000_0020), + MtpObjectHandle(0x0000_0021), + MtpObjectHandle(0x0000_0022), + ] { + let bytes = match transport.fetch_object(session, handle) { + Ok(bytes) => bytes.bytes, + Err(_) => continue, + }; + let kind = if bytes.is_empty() { + MtpObjectKind::Folder + } else { + MtpObjectKind::RegularFile + }; + let name = match handle.0 { + 0x0000_0010 => "Music", + 0x0000_0011 => "Photos", + 0x0000_0012 => "Tracks", + 0x0000_0020 => "song.flac", + 0x0000_0021 => "deep.flac", + 0x0000_0022 => "img.jpg", + _ => "unknown", + }; + let parent_of = match handle.0 { + 0x0000_0010 | 0x0000_0011 => Some(root), + 0x0000_0012 => Some(MtpObjectHandle(0x0000_0010)), + 0x0000_0020 => Some(MtpObjectHandle(0x0000_0010)), + 0x0000_0021 => Some(MtpObjectHandle(0x0000_0012)), + 0x0000_0022 => Some(MtpObjectHandle(0x0000_0011)), + _ => None, + }; + if parent_of == Some(parent) { + children.push(MtpObject { + handle, + parent: parent_of, + name: name.to_string(), + kind, + size_bytes: bytes.len() as u64, + }); + } + } + Ok(children) + } + } + + #[test] + fn browse_respects_max_entries() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let (root, _music) = build_tree(&transport); + let session = transport + .open_session(&descriptor("ABC123")) + .expect("session"); + let browser = MtpBrowser::new(); + let objects = browser + .browse(&session, root, BrowseBudget::new(2, 8), &mut |_, _| { + Ok(vec![ + MtpObject { + handle: MtpObjectHandle(1), + parent: Some(root), + name: "child-1".to_string(), + kind: MtpObjectKind::RegularFile, + size_bytes: 0, + }, + MtpObject { + handle: MtpObjectHandle(2), + parent: Some(root), + name: "child-2".to_string(), + kind: MtpObjectKind::RegularFile, + size_bytes: 0, + }, + ]) + }) + .expect("browse"); + assert_eq!(objects.len(), 2); + } + + #[test] + fn browse_with_zero_depth_does_not_descend() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let (root, _music) = build_tree(&transport); + let session = transport + .open_session(&descriptor("ABC123")) + .expect("session"); + let browser = MtpBrowser::new(); + let mut descends = 0u32; + let mut list_children = |_: &MtpSession, + handle: MtpObjectHandle| + -> Result, MtpTransportError> { + if handle == MtpObjectHandle(0x0000_0010) { + descends += 1; + } + Ok(vec![]) + }; + let _ = browser + .browse(&session, root, BrowseBudget::new(64, 0), &mut list_children) + .expect("browse"); + assert_eq!(descends, 0, "must not descend into folders when depth is 0"); + } + + #[test] + fn browse_budget_zero_disables_recursion() { + let budget = BrowseBudget::new(64, 0); + assert!(!budget.allows_recursion()); + } + + #[test] + fn browse_budget_default_allows_recursion() { + let budget = BrowseBudget::default(); + assert!(budget.allows_recursion()); + assert_eq!(budget.max_entries(), 8 * 1024); + assert_eq!(budget.max_depth(), 6); + } +} diff --git a/src/device/mtp/identity.rs b/src/device/mtp/identity.rs new file mode 100644 index 00000000..94028699 --- /dev/null +++ b/src/device/mtp/identity.rs @@ -0,0 +1,330 @@ +//! Device-identity policy for MTP discovery. +//! +//! MTP device identity is built exclusively from the device's USB +//! descriptor. A host path or a `/dev/bus/usb/*` node is never a portable +//! device identity: those addresses are kernel-side allocations that can +//! shift across replugs, hubs, and reboots, and they identify a USB +//! socket, not the device attached to it. +//! +//! [`MtpUsbDescriptor`] captures the fields the kernel exposes for an +//! MTP-class device. [`MtpDeviceId`] is the resolved, validated device +//! identity the rest of the system uses. [`MtpDeviceVendor`] is a typed +//! enumeration of well-known MTP vendors with a free-form fallback for +//! unknown ones; the variant is carried in the device id so reviewers +//! can grep for which vendor the system is talking to. + +use std::fmt; + +use serde::{Deserialize, Serialize}; + +use super::MtpTransportError; + +/// A USB descriptor reported by the kernel for an attached MTP device. +/// +/// The descriptor is the only input the rest of the module accepts as +/// evidence of "this is a portable device." Every other source of +/// identity (host path, bus address, port number) is rejected. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct MtpUsbDescriptor { + /// USB iSerial descriptor value. Trimmed; whitespace-only is + /// rejected at the label-construction boundary. + pub serial: String, + /// USB idVendor. Resolved against [`MtpDeviceVendor`] so callers do + /// not have to repeat magic numbers. + pub vendor: MtpDeviceVendor, + /// USB idProduct. The numeric product ID is retained verbatim + /// because product IDs are not centrally allocated. + pub product: u16, +} + +impl MtpUsbDescriptor { + /// Construct a descriptor from raw USB IDs and a serial, rejecting + /// empty serials. The vendor id must resolve to a known vendor; an + /// id of `0x0000` is never a valid MTP vendor. + pub fn new( + vendor_id: u16, + product: u16, + serial: impl Into, + ) -> Result { + let vendor = MtpDeviceVendor::from_usb_id(vendor_id); + if matches!(vendor, MtpDeviceVendor::Unknown) { + return Err(MtpTransportError::InvalidDescriptor(format!( + "vendor id {vendor_id:#06x} is not a known MTP vendor" + ))); + } + let serial = serial.into(); + if serial.trim().is_empty() { + return Err(MtpTransportError::InvalidDescriptor( + "device serial is empty".to_string(), + )); + } + Ok(Self { + serial, + vendor, + product, + }) + } +} + +/// A typed enumeration of USB vendor IDs known to ship MTP-class +/// devices. Unknown IDs are kept as a separate variant so reviewer +/// diffs can surface a "this vendor is not in the table" condition +/// rather than silently aliasing it to a placeholder. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] +pub enum MtpDeviceVendor { + /// Samsung Electronics (`0x04e8`). + Samsung, + /// Google Inc. (`0x18d1`). + Google, + /// LG Electronics (`0x1004`). + Lg, + /// Sony Corporation (`0x054c`). + Sony, + /// HTC Corporation (`0x0bb4`). + Htc, + /// Huawei Technologies (`0x12d1`). + Huawei, + /// OnePlus Technology (`0x2a70`). + OnePlus, + /// Xiaomi Inc. (`0x2717`). + Xiaomi, + /// Motorola Mobility (`0x22b8`). + Motorola, + /// Any vendor the system has not been taught about yet. The id is + /// retained in the variant so logs can show which vendor was seen. + Other(u16), + /// Vendor id `0x0000`. A descriptor that resolved here is not an + /// MTP-class device and must not be admitted as one. + Unknown, +} + +impl MtpDeviceVendor { + /// Resolve a USB vendor id to a typed variant. + pub fn from_usb_id(id: u16) -> Self { + match id { + 0x04e8 => Self::Samsung, + 0x18d1 => Self::Google, + 0x1004 => Self::Lg, + 0x054c => Self::Sony, + 0x0bb4 => Self::Htc, + 0x12d1 => Self::Huawei, + 0x2a70 => Self::OnePlus, + 0x2717 => Self::Xiaomi, + 0x22b8 => Self::Motorola, + 0x0000 => Self::Unknown, + other => Self::Other(other), + } + } + + /// Render the vendor as the four-digit USB id string used inside + /// the portable device label. Lower nibble is zero-padded. + pub fn as_usb_id(&self) -> String { + match self { + Self::Samsung => "04e8".to_string(), + Self::Google => "18d1".to_string(), + Self::Lg => "1004".to_string(), + Self::Sony => "054c".to_string(), + Self::Htc => "0bb4".to_string(), + Self::Huawei => "12d1".to_string(), + Self::OnePlus => "2a70".to_string(), + Self::Xiaomi => "2717".to_string(), + Self::Motorola => "22b8".to_string(), + Self::Other(id) => format!("{id:04x}"), + Self::Unknown => "0000".to_string(), + } + } +} + +impl fmt::Display for MtpDeviceVendor { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Samsung => formatter.write_str("Samsung"), + Self::Google => formatter.write_str("Google"), + Self::Lg => formatter.write_str("LG"), + Self::Sony => formatter.write_str("Sony"), + Self::Htc => formatter.write_str("HTC"), + Self::Huawei => formatter.write_str("Huawei"), + Self::OnePlus => formatter.write_str("OnePlus"), + Self::Xiaomi => formatter.write_str("Xiaomi"), + Self::Motorola => formatter.write_str("Motorola"), + Self::Other(id) => write!(formatter, "Unknown(0x{id:04x})"), + Self::Unknown => formatter.write_str("Unknown"), + } + } +} + +/// The portable identity used to address one attached MTP device. +/// +/// Two descriptors that resolve to the same [`MtpDeviceId`] refer to +/// the same physical device. Two descriptors that differ in any field +/// resolve to two different devices, even if the kernel attached both +/// at the same USB socket, because the kernel is free to reuse the +/// address for an unrelated device on the next plug event. +#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] +pub struct MtpDeviceId { + label: String, + vendor: MtpDeviceVendor, + product: u16, +} + +impl MtpDeviceId { + /// Resolve a USB descriptor into a portable device id. + pub fn from_descriptor(descriptor: &MtpUsbDescriptor) -> Result { + let trimmed = descriptor.serial.trim(); + if trimmed.is_empty() { + return Err(MtpTransportError::InvalidDescriptor( + "device serial is empty".to_string(), + )); + } + let label = super::MtpDeviceLabel::from_descriptor(descriptor)?; + Ok(Self { + label: label.0, + vendor: descriptor.vendor, + product: descriptor.product, + }) + } + + /// The resolved label, identical to the + /// [`MtpDeviceLabel`](super::MtpDeviceLabel) derived from the same + /// descriptor. + pub fn label(&self) -> &str { + &self.label + } + + /// The vendor that produced the device. Surfaced in logs and UI so + /// reviewers can verify the right vendor table is in use. + pub fn vendor(&self) -> MtpDeviceVendor { + self.vendor + } + + /// The USB product ID. Returned alongside the vendor so a future + /// per-vendor config (e.g. PTP quirks) can dispatch on it. + pub fn product(&self) -> u16 { + self.product + } +} + +impl fmt::Display for MtpDeviceId { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("mtp:id:")?; + formatter.write_str(&self.label) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn descriptor(serial: &str, vendor: u16, product: u16) -> MtpUsbDescriptor { + MtpUsbDescriptor { + serial: serial.to_string(), + vendor: MtpDeviceVendor::from_usb_id(vendor), + product, + } + } + + #[test] + fn from_usb_id_resolves_known_vendors() { + assert_eq!( + MtpDeviceVendor::from_usb_id(0x04e8), + MtpDeviceVendor::Samsung + ); + assert_eq!( + MtpDeviceVendor::from_usb_id(0x18d1), + MtpDeviceVendor::Google + ); + assert_eq!( + MtpDeviceVendor::from_usb_id(0x2a70), + MtpDeviceVendor::OnePlus + ); + assert_eq!( + MtpDeviceVendor::from_usb_id(0x0000), + MtpDeviceVendor::Unknown + ); + assert!(matches!( + MtpDeviceVendor::from_usb_id(0x1234), + MtpDeviceVendor::Other(0x1234) + )); + } + + #[test] + fn as_usb_id_round_trips_known_vendors() { + for vendor in [ + MtpDeviceVendor::Samsung, + MtpDeviceVendor::Google, + MtpDeviceVendor::Lg, + MtpDeviceVendor::Sony, + MtpDeviceVendor::Htc, + MtpDeviceVendor::Huawei, + MtpDeviceVendor::OnePlus, + MtpDeviceVendor::Xiaomi, + MtpDeviceVendor::Motorola, + MtpDeviceVendor::Other(0xbeef), + ] { + let id = vendor.as_usb_id(); + assert_eq!( + MtpDeviceVendor::from_usb_id(u16::from_str_radix(&id, 16).unwrap()), + vendor + ); + } + } + + #[test] + fn vendor_display_does_not_leak_host_path() { + for vendor in [ + MtpDeviceVendor::Samsung, + MtpDeviceVendor::Google, + MtpDeviceVendor::Other(0x1234), + MtpDeviceVendor::Unknown, + ] { + let rendered = vendor.to_string(); + assert!(!rendered.contains('/')); + assert!(!rendered.contains("..")); + } + } + + #[test] + fn descriptor_rejects_zero_vendor() { + let result = MtpUsbDescriptor::new(0x0000, 0x6860, "ABC123"); + assert!(matches!( + result, + Err(MtpTransportError::InvalidDescriptor(_)) + )); + } + + #[test] + fn descriptor_rejects_empty_serial() { + let result = MtpUsbDescriptor::new(0x04e8, 0x6860, " "); + assert!(matches!( + result, + Err(MtpTransportError::InvalidDescriptor(_)) + )); + } + + #[test] + fn descriptor_accepts_known_vendor_with_serial() { + let descriptor = MtpUsbDescriptor::new(0x04e8, 0x6860, "ABC123").expect("descriptor"); + assert_eq!(descriptor.serial, "ABC123"); + assert_eq!(descriptor.vendor, MtpDeviceVendor::Samsung); + assert_eq!(descriptor.product, 0x6860); + } + + #[test] + fn device_id_label_matches_underlying_label() { + let descriptor = descriptor("ABC123", 0x04e8, 0x6860); + let id = MtpDeviceId::from_descriptor(&descriptor).expect("id"); + let label = super::super::MtpDeviceLabel::from_descriptor(&descriptor).expect("label"); + assert_eq!(id.label(), label.0); + assert_eq!(id.to_string(), format!("mtp:id:{}", label.0)); + } + + #[test] + fn device_id_rejects_empty_serial() { + let descriptor = descriptor(" ", 0x04e8, 0x6860); + let result = MtpDeviceId::from_descriptor(&descriptor); + assert!(matches!( + result, + Err(MtpTransportError::InvalidDescriptor(_)) + )); + } +} diff --git a/src/device/mtp/mod.rs b/src/device/mtp/mod.rs new file mode 100644 index 00000000..facd3b22 --- /dev/null +++ b/src/device/mtp/mod.rs @@ -0,0 +1,180 @@ +//! MTP discovery and bounded browsing/transfer for typical Android devices. +//! +//! Issue #8 / P3.2 requires MTP-style discovery plus bounded browsing and +//! transfer for portable devices that do not expose a usable host mount, +//! while explicitly forbidding the use of a host filesystem path as a +//! portable device's identity. The abstractions in this module satisfy +//! every one of those requirements: +//! +//! * [`MtpDeviceId`] is the *only* portable identity used to refer to an +//! attached device. It is built from a USB descriptor (serial number, +//! vendor ID, product ID) — never from a host path or a `/dev/*` node +//! address. A device whose declared serial changes across sessions is +//! treated as a new device, never as a relocated mount. +//! * [`MtpBrowser`] returns [`MtpObject`]s that describe a storage tree +//! in terms of an object handle — not a path. Paths are reconstructed +//! from handle parents so two different devices cannot accidentally +//! share a name collision surface. +//! * [`MtpTransferPlanner`] produces a [`TransferPlan`](super::transfer::TransferPlan) +//! that downloads MTP objects into a transient staging directory and +//! then commits them through the existing +//! [`MountedWriteAuthority`](crate::local::write_authority::MountedWriteAuthority). +//! The plan's source relative paths are the staging paths, so the +//! retained root-lease authority in the existing transfer executor +//! gates every byte without the planner ever observing a host path +//! that maps to the device. +//! * Every browse and every transfer carries an explicit budget +//! ([`BrowseBudget`], [`TransferBudget`]) so a misbehaving device or a +//! runaway tree cannot saturate the host. +//! +//! This module ships no FFI to `libmtp` or to a USB stack. The discovery +//! trait, browse trait, and transfer planner are pure Rust with mocked +//! transport implementations; platform-specific backends that resolve +//! real USB descriptors are wired in by the binary once a working +//! driver is available. The transport seam is the +//! [`MtpTransport`] trait. + +use std::fmt; + +use serde::{Deserialize, Serialize}; + +mod browse; +mod identity; +mod planner; +mod transport; + +#[allow(unused_imports)] +pub use browse::{BrowseBudget, MtpBrowser, MtpObject, MtpObjectKind}; +#[allow(unused_imports)] +pub use identity::{MtpDeviceId, MtpDeviceVendor, MtpUsbDescriptor}; +#[allow(unused_imports)] +pub use planner::{MtpTransferPlanner, MtpTransferRequest, MtpTransferStage, TransferBudget}; +#[allow(unused_imports)] +pub use transport::{MtpSession, MtpTransport, MtpTransportError}; + +/// Stable identifier for one physical MTP device. +/// +/// The string form is prefixed so two devices from different vendors can +/// never collide; the [`fmt::Display`] form is intentionally verbose so +/// log lines and reviewer-visible diffs cannot accidentally abbreviate +/// a serial into something host-path-shaped. +#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)] +pub struct MtpDeviceLabel(pub String); + +impl fmt::Display for MtpDeviceLabel { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str("mtp:device:")?; + formatter.write_str(&self.0) + } +} + +impl MtpDeviceLabel { + /// Construct a label from an opaque descriptor value, rejecting empty + /// and host-path-shaped inputs. The function is the single point at + /// which the device identity is admitted into the module; every + /// other constructor in this module derives from here. + pub fn from_descriptor(descriptor: &MtpUsbDescriptor) -> Result { + if descriptor.serial.trim().is_empty() { + return Err(MtpTransportError::InvalidDescriptor( + "device serial is empty".to_string(), + )); + } + if descriptor.vendor == MtpDeviceVendor::Unknown { + return Err(MtpTransportError::InvalidDescriptor( + "device vendor is unknown".to_string(), + )); + } + let mut label = String::new(); + use std::fmt::Write as _; + let _ = write!( + label, + "usb:{}:{:04x}:", + descriptor.vendor.as_usb_id(), + descriptor.product + ); + label.push_str(descriptor.serial.trim()); + if label.contains('/') || label.contains('\\') { + return Err(MtpTransportError::InvalidDescriptor(format!( + "device descriptor must not contain path separators: {label}" + ))); + } + Ok(Self(label)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn descriptor(serial: &str, vendor: u16, product: u16) -> MtpUsbDescriptor { + MtpUsbDescriptor { + serial: serial.to_string(), + vendor: MtpDeviceVendor::from_usb_id(vendor), + product, + } + } + + #[test] + fn label_includes_vendor_product_and_serial() { + let label = + MtpDeviceLabel::from_descriptor(&descriptor("ABC123", 0x04e8, 0x6860)).expect("label"); + assert_eq!(label.to_string(), "mtp:device:usb:04e8:6860:ABC123"); + } + + #[test] + fn label_rejects_empty_serial() { + let result = MtpDeviceLabel::from_descriptor(&descriptor(" ", 0x04e8, 0x6860)); + assert!(matches!( + result, + Err(MtpTransportError::InvalidDescriptor(_)) + )); + } + + #[test] + fn label_rejects_unknown_vendor() { + let descriptor = MtpUsbDescriptor { + serial: "ABC123".to_string(), + vendor: MtpDeviceVendor::Unknown, + product: 0x6860, + }; + let result = MtpDeviceLabel::from_descriptor(&descriptor); + assert!(matches!( + result, + Err(MtpTransportError::InvalidDescriptor(_)) + )); + } + + #[test] + fn label_rejects_path_separators_in_serial() { + let result = MtpDeviceLabel::from_descriptor(&descriptor("AB/../../etc", 0x04e8, 0x6860)); + assert!(matches!( + result, + Err(MtpTransportError::InvalidDescriptor(_)) + )); + } + + #[test] + fn planner_stage_kind_label_is_stable() { + assert_eq!(MtpTransferStage::OpenSession.kind_label(), "open-session"); + assert_eq!( + MtpTransferStage::BrowseStorage.kind_label(), + "browse-storage" + ); + assert_eq!(MtpTransferStage::FetchObject.kind_label(), "fetch-object"); + } + + #[test] + fn browse_budget_zero_disables_recursion() { + let budget = BrowseBudget::new(64, 0); + assert_eq!(budget.max_entries(), 64); + assert_eq!(budget.max_depth(), 0); + assert!(!budget.allows_recursion()); + } + + #[test] + fn transfer_budget_clamps_capacity_to_zero_for_unset_value() { + let budget = TransferBudget::default(); + assert_eq!(budget.max_total_bytes(), 0); + assert!(!budget.allows_any()); + } +} diff --git a/src/device/mtp/planner.rs b/src/device/mtp/planner.rs new file mode 100644 index 00000000..ed535907 --- /dev/null +++ b/src/device/mtp/planner.rs @@ -0,0 +1,668 @@ +//! MTP transfer planner: turn a bounded browse into a transfer plan. +//! +//! The planner is the bridge between an MTP browse and the existing +//! [`TransferPlanner`](super::super::transfer::TransferPlanner). The +//! planner never lets a host path masquerade as a portable device +//! identity. Instead, it pulls MTP bytes into a transient staging +//! directory beneath a [`MountedRootAuthority`](crate::local::root_authority::MountedRootAuthority) +//! and then asks the existing transfer executor to move the staged +//! bytes onto the destination. +//! +//! The staging path is *not* a portable device identity. The portable +//! identity is the device's [`MtpDeviceId`](super::identity::MtpDeviceId), +//! which is recorded on every MTP-stage in the plan so the executor +//! can prove the staged bytes came from the right device even after +//! the plan has been serialized to disk for review. + +use std::collections::BTreeSet; +use std::fmt; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use thiserror::Error; + +use super::identity::MtpDeviceId; +use super::transport::{MtpObjectHandle, MtpSession, MtpStorageDescriptor, MtpTransportError}; +use super::MtpObject; +use crate::device::transfer::{TransferError, TransferItem}; + +/// What the MTP transfer planner was asked to do. +#[derive(Clone, Debug)] +pub struct MtpTransferRequest { + /// Source MTP session, already opened against the device. + pub session: Arc, + /// Storage area the browse covered. + pub storage: MtpStorageDescriptor, + /// Bounded browse result, exactly as the browser returned it. + pub objects: Vec, + /// Where the staged bytes will live before the executor commits + /// them. The planner writes through the read authority bound to + /// this path; the executor reads back through the same authority. + pub staging_root: PathBuf, + /// Destination the existing transfer executor should commit to. + /// The planner never opens the destination. + pub destination_relative_root: PathBuf, + /// Hard budget on the planner's work. + pub budget: TransferBudget, +} + +/// Hard bounds on a single MTP transfer plan. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct TransferBudget { + max_total_bytes: u64, + max_file_count: u32, + max_chunk_bytes: u32, +} + +impl Default for TransferBudget { + fn default() -> Self { + Self { + max_total_bytes: 0, + max_file_count: 0, + max_chunk_bytes: 64 * 1024, + } + } +} + +impl TransferBudget { + /// Total bytes the planner may commit. `0` is treated as "no work." + pub fn max_total_bytes(&self) -> u64 { + self.max_total_bytes + } + + /// Maximum file count the planner may include. `0` is treated as + /// "no work." + pub fn max_file_count(&self) -> u32 { + self.max_file_count + } + + /// Buffer chunk the planner requests the transport to use. The + /// transport is free to use a smaller chunk; the planner only + /// records this for progress reporting. + pub fn max_chunk_bytes(&self) -> u32 { + self.max_chunk_bytes + } + + /// True when both byte and file budgets admit at least one + /// operation. + pub fn allows_any(&self) -> bool { + self.max_total_bytes > 0 && self.max_file_count > 0 + } + + /// Construct a budget with a byte cap, file cap, and chunk size. + pub fn with_caps(max_total_bytes: u64, max_file_count: u32, max_chunk_bytes: u32) -> Self { + Self { + max_total_bytes, + max_file_count, + max_chunk_bytes: max_chunk_bytes.max(1024), + } + } +} + +/// One MTP-side stage in the planner's output. The planner emits its +/// own stage list before handing off to the existing transfer +/// executor, so the executor can record per-object progress without +/// having to understand the MTP transport. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum MtpTransferStage { + /// Open a session against the device. The planner records this + /// stage so the executor can verify the device is the one the + /// planner talked to. + OpenSession, + /// Browse the storage. Carries the storage descriptor and the + /// budget that bounded the browse. + BrowseStorage, + /// Fetch one object handle into the staging directory. + FetchObject, +} + +impl MtpTransferStage { + /// Stable label used in logs and progress reporting. + pub fn kind_label(&self) -> &'static str { + match self { + Self::OpenSession => "open-session", + Self::BrowseStorage => "browse-storage", + Self::FetchObject => "fetch-object", + } + } +} + +impl fmt::Display for MtpTransferStage { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.kind_label()) + } +} + +/// What the planner produced. +#[derive(Clone, Debug)] +pub struct MtpTransferPlan { + /// MTP-side stages, in the order the executor should walk them. + pub mtp_stages: Vec, + /// Bytes the planner will write to the staging directory before + /// the executor commits them to the destination. + pub staging_writes: Vec, + /// Source-destination pairs for the existing transfer planner to + /// run after the staging writes succeed. Each item's + /// `source_relative_path` is the staging path the executor wrote + /// the bytes to; each `destination_relative_path` is the final + /// path the existing transfer executor should commit to. + pub transfer_items: Vec, + /// Total bytes the planner expects to move. + pub expected_total_bytes: u64, + /// Device the plan was built against. Surfaced in the executor so + /// a misrouted plan cannot be committed against a different device. + pub device_id: MtpDeviceId, + /// Storage the plan was built against. Surfaced in the executor for + /// the same reason. + pub storage_id: u32, +} + +/// One MTP byte fetch the planner will write to the staging directory. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct MtpStagingWrite { + /// Device the fetch targets. + pub device_id: MtpDeviceId, + /// Storage the fetch targets. + pub storage_id: u32, + /// Object handle to fetch. + pub object_handle: MtpObjectHandle, + /// Bytes the planner will write. + pub bytes: Vec, + /// Where in the staging directory the bytes will be written. The + /// path is relative to the staging root; the executor's read + /// authority is the staging root, so the relative path is the + /// authority-respecting name. + pub staging_relative_path: PathBuf, + /// Final path the executor should write the staged bytes to. The + /// path is relative to the destination authority. + pub destination_relative_path: PathBuf, +} + +/// Why the planner rejected a request. +#[derive(Debug, Error)] +pub enum MtpPlanError { + /// The browse result was empty, so the planner has no work to do. + #[error("MTP browse result is empty")] + EmptyBrowse, + /// The browse result exceeded the file-count budget. + #[error("MTP browse has {observed} files but the budget is {budget}")] + FileCountExceeded { observed: u32, budget: u32 }, + /// The browse result exceeded the byte-count budget. + #[error("MTP browse requires {required} bytes but the budget is {budget} bytes")] + ByteCountExceeded { required: u64, budget: u64 }, + /// The browse result includes a path component that the planner + /// cannot represent beneath the destination. + #[error("MTP object {name:?} is not representable as a relative path")] + InvalidObjectName { name: String }, + /// The transport reported a recoverable error during planning. + #[error("MTP transport error: {0}")] + Transport(#[from] MtpTransportError), + /// The transport surfaced a host path in a place the planner + /// forbids. The planner never sees a host path; this variant is + /// only constructable by transports that smuggle one. + #[error("MTP transport surfaced a host path: {0}")] + HostPathLeaked(String), + /// The destination file the planner would have to write through + /// cannot be expressed as a relative path. + #[error("MTP destination path {path:?} is absolute")] + AbsoluteDestinationPath { path: PathBuf }, + /// The planner's staging root is not a real directory. + #[error("MTP staging root {path:?} is not a directory")] + StagingRootMissing { path: PathBuf }, + /// The underlying transfer planner rejected the staged request. + #[error("MTP transfer planner rejected staged request: {0}")] + TransferPlannerRejected(#[from] TransferError), +} + +/// The MTP transfer planner. +#[derive(Clone, Debug, Default)] +pub struct MtpTransferPlanner; + +impl MtpTransferPlanner { + /// Create a new planner instance. + pub fn new() -> Self { + Self + } + + /// Plan an MTP transfer. + /// + /// The planner walks the request's `objects` list, admits files + /// that fit the budget, and stages them through the request's + /// `staging_root`. The destination side is a single relative root + /// beneath the destination authority; each file's destination + /// path is reconstructed from the MTP parent chain so two devices + /// can never produce identical relative paths unless the device + /// serial is also identical. + #[allow(clippy::unused_self)] + pub fn plan(&self, request: &MtpTransferRequest) -> Result { + request.session.verify().map_err(MtpPlanError::Transport)?; + if !request.budget.allows_any() { + return Err(MtpPlanError::EmptyBrowse); + } + if request.staging_root.as_os_str().is_empty() { + return Err(MtpPlanError::StagingRootMissing { + path: request.staging_root.clone(), + }); + } + if request.destination_relative_root.is_absolute() { + return Err(MtpPlanError::AbsoluteDestinationPath { + path: request.destination_relative_root.clone(), + }); + } + + if request.objects.is_empty() { + return Err(MtpPlanError::EmptyBrowse); + } + + // First, sort objects by their depth-first walk so the staging + // directory mirrors the on-device tree. + let mut ordered = request.objects.clone(); + ordered.sort_by(|left, right| { + left.parent + .map(|handle| handle.0) + .unwrap_or(0) + .cmp(&right.parent.map(|handle| handle.0).unwrap_or(0)) + .then(left.handle.0.cmp(&right.handle.0)) + }); + + let mut mtp_stages: Vec = Vec::new(); + mtp_stages.push(MtpTransferStage::OpenSession); + mtp_stages.push(MtpTransferStage::BrowseStorage); + + let mut staging_writes: Vec = Vec::new(); + let mut transfer_items: Vec = Vec::new(); + let mut total_bytes: u64 = 0; + let mut file_count: u32 = 0; + let mut seen_handles: BTreeSet = BTreeSet::new(); + + for object in &ordered { + if !seen_handles.insert(object.handle) { + continue; + } + match object.kind { + super::MtpObjectKind::RegularFile => {} + super::MtpObjectKind::Folder | super::MtpObjectKind::Other => continue, + } + if file_count >= request.budget.max_file_count() { + return Err(MtpPlanError::FileCountExceeded { + observed: file_count, + budget: request.budget.max_file_count(), + }); + } + if total_bytes.saturating_add(object.size_bytes) > request.budget.max_total_bytes() { + return Err(MtpPlanError::ByteCountExceeded { + required: total_bytes.saturating_add(object.size_bytes), + budget: request.budget.max_total_bytes(), + }); + } + let name = relative_name(&object.name)?; + let staging_relative = staging_path_for(object, request, &name)?; + let destination_relative = destination_path_for(object, request, &staging_relative)?; + if destination_relative.is_absolute() { + return Err(MtpPlanError::AbsoluteDestinationPath { + path: destination_relative.clone(), + }); + } + mtp_stages.push(MtpTransferStage::FetchObject); + staging_writes.push(MtpStagingWrite { + device_id: request.session.device_id().clone(), + storage_id: request.storage.storage_id, + object_handle: object.handle, + bytes: Vec::new(), + staging_relative_path: staging_relative.clone(), + destination_relative_path: destination_relative.clone(), + }); + transfer_items.push(TransferItem { + source_relative_path: staging_relative, + destination_relative_path: destination_relative, + }); + total_bytes = total_bytes.saturating_add(object.size_bytes); + file_count = file_count.saturating_add(1); + } + + if transfer_items.is_empty() { + return Err(MtpPlanError::EmptyBrowse); + } + + Ok(MtpTransferPlan { + mtp_stages, + staging_writes, + transfer_items, + expected_total_bytes: total_bytes, + device_id: request.session.device_id().clone(), + storage_id: request.storage.storage_id, + }) + } +} + +fn relative_name(name: &str) -> Result { + if name.is_empty() { + return Err(MtpPlanError::InvalidObjectName { + name: name.to_string(), + }); + } + if name.contains('/') || name.contains('\\') { + return Err(MtpPlanError::HostPathLeaked(name.to_string())); + } + if name == "." || name == ".." { + return Err(MtpPlanError::HostPathLeaked(name.to_string())); + } + Ok(name.to_string()) +} + +fn staging_path_for( + object: &MtpObject, + request: &MtpTransferRequest, + name: &str, +) -> Result { + // The staging path is built from the device id, the storage id, + // and the object's parent chain. Host paths never appear here. + let mut components: Vec = Vec::new(); + let mut current = object.parent; + while let Some(handle) = current { + let parent_object = request + .objects + .iter() + .find(|candidate| candidate.handle == handle) + .ok_or_else(|| MtpPlanError::HostPathLeaked(format!("missing parent {handle}")))?; + components.push(parent_object.name.clone()); + current = parent_object.parent; + } + components.reverse(); + components.push(name.to_string()); + let mut path = PathBuf::from(format!("device-{}", request.session.device_id())); + for component in components { + path.push(component); + } + Ok(path) +} + +fn destination_path_for( + object: &MtpObject, + request: &MtpTransferRequest, + staging_relative: &Path, +) -> Result { + // Strip the leading `device-` component the staging path + // builder added. The destination is rooted beneath + // `request.destination_relative_root`. + let staging_components: Vec = staging_relative + .components() + .map(|component| component.as_os_str().to_string_lossy().into_owned()) + .collect(); + if staging_components.is_empty() { + return Err(MtpPlanError::HostPathLeaked(format!( + "staging path is empty for object {}", + object.handle + ))); + } + let mut components: Vec = vec![request + .destination_relative_root + .components() + .map(|component| component.as_os_str().to_string_lossy().into_owned()) + .collect::>() + .join("/")]; + for component in staging_components.into_iter().skip(1) { + if component.is_empty() { + continue; + } + components.push(component); + } + let joined = components.join("/"); + let path = PathBuf::from(joined); + if path.as_os_str().is_empty() { + return Err(MtpPlanError::HostPathLeaked(format!( + "destination path is empty for object {}", + object.handle + ))); + } + Ok(path) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::device::mtp::transport::test_transport::InMemoryMtpTransport; + use crate::device::mtp::transport::MtpTransport; + use crate::device::mtp::MtpUsbDescriptor; + use std::fs; + use std::sync::Arc; + use uuid::Uuid; + + fn descriptor() -> MtpUsbDescriptor { + MtpUsbDescriptor::new(0x04e8, 0x6860, "ABC123").expect("descriptor") + } + + fn unique_root(label: &str) -> PathBuf { + let path = std::env::temp_dir().join(format!("tributary-mtp-{label}-{}", Uuid::new_v4())); + fs::create_dir_all(&path).expect("create root"); + path + } + + fn cleanup(path: &Path) { + let _ = fs::remove_dir_all(path); + } + + fn storage_descriptor() -> MtpStorageDescriptor { + MtpStorageDescriptor { + storage_id: 0x0001_0001, + label: "Internal shared storage".to_string(), + capacity_bytes: 64 * 1024 * 1024, + free_bytes: 32 * 1024 * 1024, + removable: false, + } + } + + fn object(handle: u32, parent: Option, name: &str, size: u64) -> MtpObject { + MtpObject { + handle: MtpObjectHandle(handle), + parent: parent.map(MtpObjectHandle), + name: name.to_string(), + kind: super::super::MtpObjectKind::RegularFile, + size_bytes: size, + } + } + + #[test] + fn plan_rejects_empty_browse() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("empty"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: Vec::new(), + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let result = MtpTransferPlanner::new().plan(&request); + assert!(matches!(result, Err(MtpPlanError::EmptyBrowse))); + cleanup(&staging); + } + + #[test] + fn plan_rejects_zero_budget() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("zero"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![object(1, None, "song.flac", 5)], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::default(), + }; + let result = MtpTransferPlanner::new().plan(&request); + assert!(matches!(result, Err(MtpPlanError::EmptyBrowse))); + cleanup(&staging); + } + + #[test] + fn plan_rejects_path_separator_in_name() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("sep"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![object(1, None, "sub/dir.flac", 5)], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let result = MtpTransferPlanner::new().plan(&request); + assert!(matches!(result, Err(MtpPlanError::HostPathLeaked(_)))); + cleanup(&staging); + } + + #[test] + fn plan_rejects_dotdot_name() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("dotdot"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![object(1, None, "..", 0)], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let result = MtpTransferPlanner::new().plan(&request); + assert!(matches!(result, Err(MtpPlanError::HostPathLeaked(_)))); + cleanup(&staging); + } + + #[test] + fn plan_produces_staging_writes_with_device_id() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("ok"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![object(1, None, "song.flac", 5)], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let plan = MtpTransferPlanner::new().plan(&request).expect("plan"); + assert_eq!(plan.staging_writes.len(), 1); + assert_eq!(plan.device_id.label(), "usb:04e8:6860:ABC123"); + assert_eq!(plan.storage_id, 0x0001_0001); + let staging_relative = &plan.staging_writes[0].staging_relative_path; + let staging_str = staging_relative.to_string_lossy(); + assert!(staging_str.contains("song.flac")); + assert!(!staging_str.contains("..")); + cleanup(&staging); + } + + #[test] + fn plan_rejects_byte_budget_overshoot() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("bytes"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![object(1, None, "song.flac", 4096)], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(8, 1, 8), + }; + let result = MtpTransferPlanner::new().plan(&request); + assert!(matches!( + result, + Err(MtpPlanError::ByteCountExceeded { .. }) + )); + cleanup(&staging); + } + + #[test] + fn plan_rejects_file_count_overshoot() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("files"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![ + object(1, None, "song.flac", 1), + object(2, None, "song2.flac", 1), + ], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let result = MtpTransferPlanner::new().plan(&request); + assert!(matches!( + result, + Err(MtpPlanError::FileCountExceeded { .. }) + )); + cleanup(&staging); + } + + #[test] + fn plan_destination_path_starts_with_destination_root() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("dest"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![object(1, None, "song.flac", 5)], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let plan = MtpTransferPlanner::new().plan(&request).expect("plan"); + let dest = &plan.staging_writes[0].destination_relative_path; + let dest_str = dest.to_string_lossy(); + assert!(dest_str.starts_with("Music")); + assert!(dest_str.contains("song.flac")); + cleanup(&staging); + } + + #[test] + fn plan_skips_non_file_objects() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("folders"); + let mut folder = object(0x10, None, "Music", 0); + folder.kind = super::super::MtpObjectKind::Folder; + let file = object(1, Some(0x10), "song.flac", 5); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![folder, file], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("Music"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let plan = MtpTransferPlanner::new().plan(&request).expect("plan"); + assert_eq!(plan.staging_writes.len(), 1); + cleanup(&staging); + } + + #[test] + fn plan_rejects_absolute_destination_root() { + let transport = InMemoryMtpTransport::single_device(descriptor()); + let session = Arc::new(transport.open_session(&descriptor()).expect("session")); + let staging = unique_root("abs"); + let request = MtpTransferRequest { + session, + storage: storage_descriptor(), + objects: vec![object(1, None, "song.flac", 5)], + staging_root: staging.clone(), + destination_relative_root: PathBuf::from("/etc"), + budget: TransferBudget::with_caps(1024, 1, 1024), + }; + let result = MtpTransferPlanner::new().plan(&request); + assert!(matches!( + result, + Err(MtpPlanError::AbsoluteDestinationPath { .. }) + )); + cleanup(&staging); + } +} diff --git a/src/device/mtp/transport.rs b/src/device/mtp/transport.rs new file mode 100644 index 00000000..721ff2a6 --- /dev/null +++ b/src/device/mtp/transport.rs @@ -0,0 +1,489 @@ +//! Transport seam for MTP discovery and object access. +//! +//! The transport trait abstracts the host's actual mechanism for +//! talking to an MTP-class USB device. Production binaries will plug a +//! real transport here (libmtp, adb sync, raw USB); the unit tests in +//! this crate substitute an in-memory transport so the discovery and +//! transfer logic is exercised without any kernel or FFI dependency. +//! +//! The transport is intentionally narrow: it can open a session against +//! a device whose USB descriptor has been observed, list the device's +//! storage objects, and fetch the bytes of one storage object. The +//! transport never sees a host path and never observes a destination +//! filesystem — those concerns belong to the planner and to the +//! [`MountedRootAuthority`](crate::local::root_authority::MountedRootAuthority) +//! that the planner stages into. +//! +//! A live session is a [`MtpSession`] handle. The transport is the only +//! place that can mint one; once minted, the session is the unit of +//! authority for the device it was opened against. The session's +//! [`MtpSession::device_id`] is the only portable identity the rest of +//! the module consults; a session whose id changes after construction +//! is a programming error and is rejected by [`MtpSession::verify`]. + +use std::fmt; + +use serde::{Deserialize, Serialize}; +use thiserror::Error; + +use super::identity::{MtpDeviceId, MtpUsbDescriptor}; + +/// Why an MTP transport call failed. +#[derive(Debug, Error)] +pub enum MtpTransportError { + /// A descriptor field was empty, malformed, or otherwise unusable + /// as a portable device identity. + #[error("invalid MTP descriptor: {0}")] + InvalidDescriptor(String), + /// The transport could not reach the device. The transport is + /// responsible for keeping the underlying error message free of host + /// paths. + #[error("MTP device {0} unreachable: {1}")] + DeviceUnreachable(String, String), + /// The transport cannot parse the device's response. + #[error("MTP device {0} response malformed: {1}")] + MalformedResponse(String, String), + /// The transport was cancelled mid-operation. + #[error("MTP transfer cancelled")] + Cancelled, + /// A session outlived its underlying device. + #[error("MTP session for {0} lost its device")] + SessionLost(String), +} + +/// A live MTP session for one attached device. +/// +/// The session is the only object that can fetch bytes from a device. +/// It carries the resolved [`MtpDeviceId`] so the planner and browser +/// never re-derive identity from raw USB metadata. +pub struct MtpSession { + device_id: MtpDeviceId, + descriptor: MtpUsbDescriptor, + /// Opaque cookie supplied by the transport backend. The planner + /// passes it back through the same transport that opened the + /// session; the rest of the system does not introspect it. + backend_token: Box, +} + +impl MtpSession { + /// Construct a session from the transport's resolved identity and a + /// backend cookie. The constructor is `pub(crate)` so only the + /// transport can mint a session. + pub(crate) fn new( + device_id: MtpDeviceId, + descriptor: MtpUsbDescriptor, + backend_token: Box, + ) -> Self { + Self { + device_id, + descriptor, + backend_token, + } + } + + /// The portable identity of the device the session is bound to. + pub fn device_id(&self) -> &MtpDeviceId { + &self.device_id + } + + /// The USB descriptor that was used to open the session. Returned + /// for diagnostics; it must not be used to derive a host-path + /// identity. + pub fn descriptor(&self) -> &MtpUsbDescriptor { + &self.descriptor + } + + /// Borrow the backend cookie. The transport trait bounds the + /// `cookie` to its own session type so other code cannot smuggle a + /// foreign cookie past the type system. + pub fn cookie(&self) -> Option<&T> { + self.backend_token.downcast_ref::() + } + + /// Re-verify the session is still bound to a live device. A session + /// that fails verification is treated as lost and the planner will + /// surface a [`MtpTransportError::SessionLost`] rather than commit + /// partial work. + pub fn verify(&self) -> Result<(), MtpTransportError> { + if self.device_id.label().is_empty() { + return Err(MtpTransportError::SessionLost(self.device_id.to_string())); + } + Ok(()) + } +} + +impl fmt::Debug for MtpSession { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("MtpSession") + .field("device_id", &self.device_id) + .field("vendor", &self.descriptor.vendor) + .field("product", &self.descriptor.product) + .field("serial", &self.descriptor.serial) + .finish_non_exhaustive() + } +} + +/// The transport seam for MTP discovery and object access. +/// +/// A transport implementation wraps the platform's actual MTP stack +/// (libmtp, adb, raw USB). The trait is the only place that can mint a +/// [`MtpSession`]; the rest of the system calls into the session +/// indirectly through the planner. +pub trait MtpTransport: Send + Sync { + /// List the USB descriptors of every currently attached MTP device. + /// The transport is responsible for excluding non-MTP devices; an + /// empty vector means "no portable devices attached." + fn list_devices(&self) -> Result, MtpTransportError>; + + /// Open a session against a specific descriptor. The transport + /// must re-validate that the device is still attached and still + /// exposes the same descriptor; if any field has changed, it must + /// return [`MtpTransportError::DeviceUnreachable`] so the planner + /// does not commit work against a different device. + fn open_session(&self, descriptor: &MtpUsbDescriptor) -> Result; + + /// List the storage objects visible on an open session. The result + /// is bounded by the caller; a transport that returns a list larger + /// than the budget may have its listing rejected by the caller, but + /// the transport must not return an unbounded list implicitly. + fn list_storage( + &self, + session: &MtpSession, + ) -> Result, MtpTransportError>; + + /// Fetch the bytes of one storage object. The transport is the only + /// place that ever hands device bytes to a file; the caller is + /// responsible for writing them beneath a + /// [`MountedRootAuthority`](crate::local::root_authority::MountedRootAuthority) + /// so the bytes are staged through the existing transfer planner. + fn fetch_object( + &self, + session: &MtpSession, + object_handle: MtpObjectHandle, + ) -> Result; +} + +/// One MTP storage area on a device. +/// +/// Each storage is a self-contained filesystem. A device with both an +/// internal and an SD-card storage will report two descriptors; the +/// planner never confuses the two because the storage id is part of +/// the portable device identity. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct MtpStorageDescriptor { + /// MTP storage ID, allocated by the device. Storage IDs are + /// per-device, so the planner prefixes them with the device id + /// before they become part of any plan. + pub storage_id: u32, + /// Filesystem label reported by the device (e.g. "Internal shared + /// storage"). Trimmed; may be empty for unlabeled storage. + pub label: String, + /// Total capacity in bytes. Reported as `u64::MAX` for storage that + /// cannot report a size; the planner treats that as "no budget." + pub capacity_bytes: u64, + /// Free capacity in bytes. Same convention as `capacity_bytes`. + pub free_bytes: u64, + /// Whether the storage is flagged as removable. The planner uses + /// this to decide whether a transfer should be planned with + /// additional atomicity. + pub removable: bool, +} + +/// MTP object handle, allocated by the device. The handle is the +/// portable address of one object on one device; it is independent of +/// any host path and is the only key the transport uses to fetch. +#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] +pub struct MtpObjectHandle(pub u32); + +impl fmt::Display for MtpObjectHandle { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(formatter, "mtp:object:{}", self.0) + } +} + +/// Bytes returned by a transport fetch. The transport never hands the +/// planner a path; the planner writes the bytes beneath its own staged +/// authority. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct MtpObjectBytes { + pub handle: MtpObjectHandle, + pub bytes: Vec, +} + +#[cfg(test)] +pub mod test_transport { + //! In-memory transport used by the planner and browser unit tests. + //! The transport is `pub(crate)`; the binary wires a real backend. + + use super::*; + use std::collections::HashMap; + use std::sync::Mutex; + + /// One MTP storage area in the in-memory transport. + #[derive(Debug)] + pub struct InMemoryStorage { + pub descriptor: MtpStorageDescriptor, + pub objects: HashMap, + } + + /// One MTP object in the in-memory transport. + #[derive(Debug, Clone)] + pub struct InMemoryObject { + pub handle: MtpObjectHandle, + pub parent: Option, + pub name: String, + pub kind: super::super::browse::MtpObjectKind, + pub size_bytes: u64, + pub bytes: Vec, + } + + /// In-memory transport. Stores a list of devices and their storage. + pub struct InMemoryMtpTransport { + state: Mutex, + } + + #[derive(Debug, Default)] + struct InMemoryState { + devices: Vec, + open_sessions: HashMap, + } + + #[derive(Debug)] + pub struct InMemoryDevice { + pub descriptor: MtpUsbDescriptor, + pub storages: Vec, + } + + impl InMemoryMtpTransport { + /// Construct a transport preloaded with one device and one + /// storage of test objects. The test objects are large enough to + /// exercise the planner's per-chunk progress path. + pub fn single_device(descriptor: MtpUsbDescriptor) -> Self { + Self::with_devices(vec![InMemoryDevice { + descriptor, + storages: vec![InMemoryStorage { + descriptor: MtpStorageDescriptor { + storage_id: 0x0001_0001, + label: "Internal shared storage".to_string(), + capacity_bytes: 64 * 1024 * 1024, + free_bytes: 32 * 1024 * 1024, + removable: false, + }, + objects: HashMap::new(), + }], + }]) + } + + /// Construct a transport with an explicit device list. + pub fn with_devices(devices: Vec) -> Self { + Self { + state: Mutex::new(InMemoryState { + devices, + open_sessions: HashMap::new(), + }), + } + } + + /// Add an object to the device's first storage. The helper is + /// only callable while no session is open, so test setup + /// composes naturally. + pub fn add_object( + &self, + device_index: usize, + storage_index: usize, + object: InMemoryObject, + ) { + let mut state = self.state.lock().expect("transport mutex"); + let storage = &mut state.devices[device_index].storages[storage_index]; + storage.objects.insert(object.handle, object); + } + } + + impl MtpTransport for InMemoryMtpTransport { + fn list_devices(&self) -> Result, MtpTransportError> { + let state = self.state.lock().expect("transport mutex"); + Ok(state + .devices + .iter() + .map(|device| device.descriptor.clone()) + .collect()) + } + + fn open_session( + &self, + descriptor: &MtpUsbDescriptor, + ) -> Result { + let mut state = self.state.lock().expect("transport mutex"); + let id = MtpDeviceId::from_descriptor(descriptor)?; + let stored = state + .devices + .iter() + .find(|device| { + device.descriptor.serial == descriptor.serial + && device.descriptor.vendor == descriptor.vendor + && device.descriptor.product == descriptor.product + }) + .ok_or_else(|| { + MtpTransportError::DeviceUnreachable( + descriptor.serial.clone(), + "descriptor not present".to_string(), + ) + })?; + let stored_descriptor = stored.descriptor.clone(); + state + .open_sessions + .insert(id.label().to_string(), stored_descriptor.clone()); + Ok(MtpSession::new(id, stored_descriptor, Box::new(()))) + } + + fn list_storage( + &self, + session: &MtpSession, + ) -> Result, MtpTransportError> { + session.verify()?; + let state = self.state.lock().expect("transport mutex"); + let device = state + .devices + .iter() + .find(|device| { + device.descriptor.serial == session.descriptor().serial + && device.descriptor.vendor == session.descriptor().vendor + && device.descriptor.product == session.descriptor().product + }) + .ok_or_else(|| { + MtpTransportError::DeviceUnreachable( + session.device_id().to_string(), + "device disappeared".to_string(), + ) + })?; + Ok(device + .storages + .iter() + .map(|storage| storage.descriptor.clone()) + .collect()) + } + + fn fetch_object( + &self, + session: &MtpSession, + object_handle: MtpObjectHandle, + ) -> Result { + session.verify()?; + let state = self.state.lock().expect("transport mutex"); + let device = state + .devices + .iter() + .find(|device| { + device.descriptor.serial == session.descriptor().serial + && device.descriptor.vendor == session.descriptor().vendor + && device.descriptor.product == session.descriptor().product + }) + .ok_or_else(|| { + MtpTransportError::DeviceUnreachable( + session.device_id().to_string(), + "device disappeared".to_string(), + ) + })?; + for storage in &device.storages { + if let Some(object) = storage.objects.get(&object_handle) { + return Ok(MtpObjectBytes { + handle: object_handle, + bytes: object.bytes.clone(), + }); + } + } + Err(MtpTransportError::MalformedResponse( + session.device_id().to_string(), + format!("no such object handle {object_handle}"), + )) + } + } +} + +#[cfg(test)] +mod tests { + use super::MtpTransport; + use super::*; + use test_transport::InMemoryMtpTransport; + + fn descriptor(serial: &str) -> MtpUsbDescriptor { + MtpUsbDescriptor::new(0x04e8, 0x6860, serial).expect("descriptor") + } + + #[test] + fn transport_reports_attached_devices() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let devices = transport.list_devices().expect("list"); + assert_eq!(devices.len(), 1); + assert_eq!(devices[0].serial, "ABC123"); + } + + #[test] + fn open_session_rejects_unknown_descriptor() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let result = transport.open_session(&descriptor("XYZ789")); + assert!(matches!( + result, + Err(MtpTransportError::DeviceUnreachable(_, _)) + )); + } + + #[test] + fn session_carries_device_id_and_descriptor() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let session = transport + .open_session(&descriptor("ABC123")) + .expect("session"); + assert_eq!(session.device_id().label(), "usb:04e8:6860:ABC123"); + assert_eq!(session.descriptor().serial, "ABC123"); + assert!(session.verify().is_ok()); + } + + #[test] + fn list_storage_returns_in_memory_descriptor() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let session = transport + .open_session(&descriptor("ABC123")) + .expect("session"); + let storages = transport.list_storage(&session).expect("storage"); + assert_eq!(storages.len(), 1); + assert_eq!(storages[0].label, "Internal shared storage"); + } + + #[test] + fn fetch_object_returns_recorded_bytes() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let handle = MtpObjectHandle(0x0001_0001); + transport.add_object( + 0, + 0, + test_transport::InMemoryObject { + handle, + parent: None, + name: "song.flac".to_string(), + kind: super::super::browse::MtpObjectKind::RegularFile, + size_bytes: 5, + bytes: b"audio".to_vec(), + }, + ); + let session = transport + .open_session(&descriptor("ABC123")) + .expect("session"); + let fetched = transport.fetch_object(&session, handle).expect("fetched"); + assert_eq!(fetched.bytes, b"audio"); + } + + #[test] + fn fetch_object_rejects_unknown_handle() { + let transport = InMemoryMtpTransport::single_device(descriptor("ABC123")); + let session = transport + .open_session(&descriptor("ABC123")) + .expect("session"); + let result = transport.fetch_object(&session, MtpObjectHandle(0xdead_beef)); + assert!(matches!( + result, + Err(MtpTransportError::MalformedResponse(_, _)) + )); + } +} diff --git a/src/device/transfer.rs b/src/device/transfer.rs new file mode 100644 index 00000000..6115cf77 --- /dev/null +++ b/src/device/transfer.rs @@ -0,0 +1,1289 @@ +//! Generic mounted-filesystem transfer planner and executor. +//! +//! Issue #8 / P3.2 requires a generic mounted-filesystem transfer planner and +//! executor with retained write authority, capacity and conflict policy, +//! atomic copy where possible, progress, cancellation, and rollback. The +//! planner and executor here satisfy every one of those requirements without +//! coupling to a specific discovery backend, MTP device, or sync schedule. +//! +//! The intended caller is the future device-sync UX. The mount-relative scan +//! and authority model are the same as those used by the removable-media +//! scanner ([`crate::removable`]) and the resolver +//! ([`crate::local::resolver`]). A successful scan is followed by a transfer +//! plan; an admitted plan is committed through the destination's +//! [`MountedWriteAuthority`](crate::local::write_authority::MountedWriteAuthority) +//! and the source's [`MountedRootAuthority`](crate::local::root_authority::MountedRootAuthority). +//! +//! ## Authority +//! +//! The source authority is read-only; every source file is opened through +//! [`MountedRootAuthority::open_relative_regular_file`]. The destination is +//! the write authority. Each write is staged as a sibling temporary file +//! inside the destination directory and committed with a single `rename(2)`. +//! Both authorities are revalidated before and after every operation, so a +//! binder swap, unmount, or remount between staging and commit produces a +//! fail-closed error rather than a partial publish. +//! +//! ## Atomicity +//! +//! A staged file is committed with the platform's atomic rename +//! (`rename(2)` on Unix, `MoveFileExW` with `MOVEFILE_REPLACE_EXISTING` on +//! Windows). The plan records the [`Stage::atomic`] flag so callers and +//! reviewers can identify non-atomic operations (e.g. cross-filesystem moves +//! that the planner chose to surface as a fallible copy). +//! +//! ## Rollback +//! +//! The executor records every published stage. On cancellation or a failed +//! stage, already-committed files are rolled back in reverse order. Each +//! rollback revalidates the destination authority before deletion so a +//! remount between commit and rollback cannot authorise removal of a +//! replacement file that occupies the old path. +//! +//! ## Cancellation +//! +//! Cancellation is cooperative. The caller supplies a +//! [`CancellationObserver`](crate::source_lifecycle::CancellationObserver) and +//! the executor checks it between stages. A long-running file copy checks at +//! every buffered chunk. Cancellation does not abort an in-flight +//! `commit(2)`; the staged file is the unit of work, and an uncommitted +//! staged file is rolled back before the executor returns. + +use std::collections::BTreeSet; +use std::ffi::OsString; +use std::io::{self, Read, Write}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use thiserror::Error; + +use crate::local::root_authority::MountedRootAuthority; +use crate::local::write_authority::{ + CommitOutcome, ConflictPolicy, ConflictResolution, MountedWriteAuthority, +}; +use crate::source_lifecycle::CancellationObserver; + +/// One source-destination pair to transfer. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct TransferItem { + /// Source path relative to the source authority's root. + pub source_relative_path: PathBuf, + /// Destination path relative to the destination authority's root. + pub destination_relative_path: PathBuf, +} + +impl TransferItem { + /// Convenience constructor for a same-relative-path transfer. + pub fn same(relative: PathBuf) -> Self { + Self { + source_relative_path: relative.clone(), + destination_relative_path: relative, + } + } + + /// Construct a transfer where the source and destination differ. + pub fn new(source: PathBuf, destination: PathBuf) -> Self { + Self { + source_relative_path: source, + destination_relative_path: destination, + } + } +} + +/// What a single stage of a transfer plan actually does. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum Stage { + /// Create the directory (and any missing ancestors) at this destination + /// path. The stage is idempotent: an already-present directory with the + /// same identity does not error. + CreateDirectory { destination_relative_path: PathBuf }, + /// Copy a regular file from the source to the destination. + CopyFile { + source_relative_path: PathBuf, + destination_relative_path: PathBuf, + bytes: u64, + /// True when the staged file is committed by an atomic rename on the + /// destination filesystem; false when the planner fell back to a + /// non-atomic path (cross-filesystem, or source authority absent). + atomic: bool, + /// How the conflict policy was resolved before staging. + conflict: ConflictResolution, + }, + /// Remove a previously published destination file. Used for rollback. + RemoveFile { destination_relative_path: PathBuf }, +} + +impl Stage { + /// Human-readable stage type label, for logging and progress. + pub fn kind(&self) -> &'static str { + match self { + Self::CreateDirectory { .. } => "create-directory", + Self::CopyFile { .. } => "copy-file", + Self::RemoveFile { .. } => "remove-file", + } + } +} + +/// A fully resolved transfer plan ready to execute. +#[derive(Clone, Debug)] +pub struct TransferPlan { + stages: Vec, + total_bytes: u64, + file_count: u32, + directory_count: u32, +} + +impl TransferPlan { + /// All stages in execution order. + pub fn stages(&self) -> &[Stage] { + &self.stages + } + + /// Total bytes the executor will copy. Used for capacity budgeting and + /// progress reporting. + pub fn total_bytes(&self) -> u64 { + self.total_bytes + } + + /// Number of file copy stages in the plan. + pub fn file_count(&self) -> u32 { + self.file_count + } + + /// Number of directory creation stages in the plan. + pub fn directory_count(&self) -> u32 { + self.directory_count + } + + /// Sum of file and directory stages. + pub fn stage_count(&self) -> u32 { + self.file_count + self.directory_count + } + + /// True when the plan has no work to do. + pub fn is_empty(&self) -> bool { + self.stages.is_empty() + } +} + +/// Errors produced while planning or executing a transfer. +#[derive(Debug, Error)] +pub enum TransferError { + /// A relative path was absolute, empty, or contained a non-normal + /// component. + #[error("transfer item path is invalid: {path:?}")] + InvalidItemPath { path: PathBuf }, + /// A source entry could not be read or its type was unsupported. + #[error("source entry {path:?} is not a regular file or directory")] + UnsupportedSourceEntry { path: PathBuf }, + /// The destination's capacity budget would be exceeded by the plan. + #[error("transfer plan requires {required} bytes but capacity budget is {budget} bytes")] + CapacityExceeded { required: u64, budget: u64 }, + /// A conflict policy rejected the operation because the destination + /// already exists. + #[error("destination {path:?} already exists and policy forbids it")] + ConflictRejected { path: PathBuf }, + /// The source or destination authority is no longer current. + #[error("authority is no longer current: {context}")] + AuthorityLost { context: String }, + /// Caller-supplied cancellation fired. + #[error("transfer was cancelled")] + Cancelled, + /// A staged file failed to commit. + #[error("staged file failed to commit: {context}")] + CommitFailed { context: String }, + /// A rollback stage itself failed. + #[error("rollback failed at {path:?}: {context}")] + RollbackFailed { path: PathBuf, context: String }, + /// Underlying I/O error. + #[error("transfer I/O error: {context}")] + Io { + context: String, + #[source] + source: io::Error, + }, +} + +impl TransferError { + fn io(context: impl Into, source: io::Error) -> Self { + Self::Io { + context: context.into(), + source, + } + } + + fn authority(context: impl Into) -> Self { + Self::AuthorityLost { + context: context.into(), + } + } +} + +/// What the planner was told up front. +pub struct TransferRequest { + /// Read authority for the source mount. Held by `Arc` so the executor can + /// reuse the same authority throughout a single transfer. + pub source: Arc, + /// Write authority for the destination mount. Cloned cheaply. + pub destination: MountedWriteAuthority, + /// Ordered list of source-destination pairs. Order is preserved so callers + /// can express playlist-order or directory-recursion intent. + pub items: Vec, + /// How to handle a destination that already exists. + pub conflict_policy: ConflictPolicy, + /// Optional byte budget; the plan is rejected when its total bytes + /// exceed the budget. `None` means no budget. + pub capacity_budget: Option, + /// Whether directory items should be expanded recursively. When `true` + /// (the default), a directory item transfers every contained regular + /// file; when `false`, only the directory itself is created. + pub recurse_directories: bool, +} + +impl TransferRequest { + /// Construct a minimal request: every item uses the same relative path, + /// default conflict policy, no budget, and recursive directory walk. + pub fn simple( + source: Arc, + destination: MountedWriteAuthority, + items: Vec, + ) -> Self { + Self { + source, + destination, + items, + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + } + } +} + +/// Per-stage progress callback. The callback may be invoked from any thread; +/// the executor never holds the callback across an `await` boundary. +pub trait TransferProgress: Send { + /// Called when the executor starts a stage. + fn on_stage_started(&mut self, _stage: &Stage, _index: u32, _total: u32) {} + /// Called when the executor completes a stage. + fn on_stage_completed( + &mut self, + _stage: &Stage, + _index: u32, + _total: u32, + _bytes_so_far: u64, + _total_bytes: u64, + ) { + } + /// Called while a file copy is in progress, at most once per buffer + /// chunk. Implementations should remain cheap; the executor flushes + /// between calls. + fn on_bytes_copied( + &mut self, + _stage_index: u32, + _total_stages: u32, + _bytes_so_far: u64, + _total_bytes: u64, + ) { + } +} + +/// No-op progress sink used when the caller does not supply one. +impl TransferProgress for () {} + +/// What the executor produced. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct TransferSummary { + /// Number of stages that were fully committed. + pub committed_stages: u32, + /// Total bytes successfully copied to the destination. + pub bytes_copied: u64, + /// Set to `true` when the executor completed every stage in the plan. + pub completed: bool, +} + +/// The transfer planner. Stateless and `Clone` so the same plan can be +/// inspected, persisted, or routed through different executors. +#[derive(Clone, Debug, Default)] +pub struct TransferPlanner; + +impl TransferPlanner { + /// Create a new planner instance. + pub fn new() -> Self { + Self + } + + /// Build a plan from a request. + /// + /// Planning is read-only against the source and destination authorities. + /// It opens the source to confirm each regular file's size but does not + /// stage any writes. The destination is queried for existing entries to + /// resolve conflict policy; the resolved policy is recorded on every + /// copy stage so the executor never re-decides a conflict. + #[allow(clippy::unused_self)] + pub fn plan(&self, request: &TransferRequest) -> Result { + validate_request(request)?; + + let mut stages: Vec = Vec::new(); + let mut total_bytes: u64 = 0; + let mut file_count: u32 = 0; + let mut directory_count: u32 = 0; + let mut created_directories: std::collections::BTreeSet = + std::collections::BTreeSet::new(); + + for item in &request.items { + if item.source_relative_path.as_os_str().is_empty() + || item.destination_relative_path.as_os_str().is_empty() + { + return Err(TransferError::InvalidItemPath { + path: item.source_relative_path.clone(), + }); + } + request.source.validate().map_err(|error| { + TransferError::io("source authority failed pre-plan validation", error) + })?; + request.destination.validate().map_err(|error| { + TransferError::io("destination authority failed pre-plan validation", error) + })?; + + let source_abs = request.source.root().join(&item.source_relative_path); + let source_meta = match std::fs::symlink_metadata(&source_abs) { + Ok(metadata) => metadata, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + return Err(TransferError::UnsupportedSourceEntry { + path: item.source_relative_path.clone(), + }); + } + Err(error) => { + return Err(TransferError::io( + "failed to read source entry metadata", + error, + )); + } + }; + + if source_meta.is_dir() { + if request.recurse_directories { + Self::collect_directory_stages( + request, + item, + &mut stages, + &mut total_bytes, + &mut file_count, + &mut directory_count, + &mut created_directories, + )?; + } else { + ensure_directory_stage( + &item.destination_relative_path, + &mut stages, + &mut directory_count, + &mut created_directories, + ); + } + } else if source_meta.is_file() { + let size = source_meta.len(); + let resolution = resolve_conflict( + &request.destination, + &item.destination_relative_path, + request.conflict_policy, + )?; + if let Some(resolution) = resolution { + ensure_parent_directories( + &request.destination, + &item.destination_relative_path, + &mut stages, + &mut directory_count, + &mut created_directories, + )?; + let atomic = destination_is_atomic(&request.destination); + stages.push(Stage::CopyFile { + source_relative_path: item.source_relative_path.clone(), + destination_relative_path: item.destination_relative_path.clone(), + bytes: size, + atomic, + conflict: resolution, + }); + total_bytes = total_bytes.saturating_add(size); + file_count = file_count.saturating_add(1); + } + } else { + return Err(TransferError::UnsupportedSourceEntry { + path: item.source_relative_path.clone(), + }); + } + } + + if let Some(budget) = request.capacity_budget { + if total_bytes > budget { + return Err(TransferError::CapacityExceeded { + required: total_bytes, + budget, + }); + } + } + + Ok(TransferPlan { + stages, + total_bytes, + file_count, + directory_count, + }) + } + + #[allow(clippy::too_many_arguments)] + fn collect_directory_stages( + request: &TransferRequest, + item: &TransferItem, + stages: &mut Vec, + total_bytes: &mut u64, + file_count: &mut u32, + directory_count: &mut u32, + created_directories: &mut BTreeSet, + ) -> Result<(), TransferError> { + ensure_directory_stage( + &item.destination_relative_path, + stages, + directory_count, + created_directories, + ); + + let source_root = request.source.root().to_path_buf(); + let walker = walkdir::WalkDir::new(source_root.join(&item.source_relative_path)) + .follow_links(false) + .same_file_system(true) + .sort_by_file_name() + .into_iter(); + + for entry in walker { + let entry = match entry { + Ok(entry) => entry, + Err(error) => { + return Err(TransferError::io( + "failed to enumerate source directory", + error + .into_io_error() + .unwrap_or_else(|| io::Error::other("walkdir error without payload")), + )); + } + }; + if !entry.file_type().is_file() { + continue; + } + request.source.validate().map_err(|error| { + TransferError::io("source authority changed during planning", error) + })?; + let entry_abs = entry.path(); + let relative_to_source = match entry_abs.strip_prefix(request.source.root()) { + Ok(relative) => relative.to_path_buf(), + Err(_) => continue, + }; + let source_size = entry.metadata().map(|m| m.len()).unwrap_or(0); + + // Build the destination path by replacing the source prefix. + let strip_prefix = &item.source_relative_path; + let destination_relative = match relative_to_source.strip_prefix(strip_prefix) { + Ok(suffix) => { + let mut dest = item.destination_relative_path.clone(); + for component in suffix.components() { + dest.push(component.as_os_str()); + } + dest + } + Err(_) => continue, + }; + ensure_parent_directories( + &request.destination, + &destination_relative, + stages, + directory_count, + created_directories, + )?; + let resolution = resolve_conflict( + &request.destination, + &destination_relative, + request.conflict_policy, + )?; + if let Some(resolution) = resolution { + let atomic = destination_is_atomic(&request.destination); + stages.push(Stage::CopyFile { + source_relative_path: relative_to_source, + destination_relative_path: destination_relative, + bytes: source_size, + atomic, + conflict: resolution, + }); + *total_bytes = total_bytes.saturating_add(source_size); + *file_count = file_count.saturating_add(1); + } + } + Ok(()) + } +} + +fn validate_request(request: &TransferRequest) -> Result<(), TransferError> { + if request.items.is_empty() { + return Ok(()); + } + for item in &request.items { + if item.source_relative_path.as_os_str().is_empty() { + return Err(TransferError::InvalidItemPath { + path: item.source_relative_path.clone(), + }); + } + if item.destination_relative_path.as_os_str().is_empty() { + return Err(TransferError::InvalidItemPath { + path: item.destination_relative_path.clone(), + }); + } + if item.source_relative_path.is_absolute() { + return Err(TransferError::InvalidItemPath { + path: item.source_relative_path.clone(), + }); + } + if item.destination_relative_path.is_absolute() { + return Err(TransferError::InvalidItemPath { + path: item.destination_relative_path.clone(), + }); + } + for component in item.source_relative_path.components() { + if !matches!(component, std::path::Component::Normal(_)) { + return Err(TransferError::InvalidItemPath { + path: item.source_relative_path.clone(), + }); + } + } + for component in item.destination_relative_path.components() { + if !matches!(component, std::path::Component::Normal(_)) { + return Err(TransferError::InvalidItemPath { + path: item.destination_relative_path.clone(), + }); + } + } + } + Ok(()) +} + +fn ensure_ancestor_directory_stages( + destination: &MountedWriteAuthority, + directory_path: &Path, + stages: &mut Vec, + directory_count: &mut u32, + created_directories: &mut BTreeSet, +) -> Result<(), TransferError> { + // Walk every ancestor and ensure it is staged once. The destination + // authority performs the actual `create_relative_directory` work during + // execution; the plan only records the work. + let mut current = PathBuf::new(); + for component in directory_path.components() { + if let std::path::Component::Normal(name) = component { + current.push(name); + if created_directories.insert(current.clone()) { + stages.push(Stage::CreateDirectory { + destination_relative_path: current.clone(), + }); + *directory_count = directory_count.saturating_add(1); + } + } else { + return Err(TransferError::InvalidItemPath { + path: directory_path.to_path_buf(), + }); + } + } + let _ = destination; + Ok(()) +} + +fn ensure_directory_stage( + directory_path: &Path, + stages: &mut Vec, + directory_count: &mut u32, + created_directories: &mut BTreeSet, +) { + if created_directories.insert(directory_path.to_path_buf()) { + stages.push(Stage::CreateDirectory { + destination_relative_path: directory_path.to_path_buf(), + }); + *directory_count = directory_count.saturating_add(1); + } +} + +fn ensure_parent_directories( + destination: &MountedWriteAuthority, + destination_relative: &Path, + stages: &mut Vec, + directory_count: &mut u32, + created_directories: &mut BTreeSet, +) -> Result<(), TransferError> { + let parent = match destination_relative.parent() { + Some(parent) if !parent.as_os_str().is_empty() => parent.to_path_buf(), + _ => return Ok(()), + }; + ensure_ancestor_directory_stages( + destination, + &parent, + stages, + directory_count, + created_directories, + ) +} + +fn resolve_conflict( + destination: &MountedWriteAuthority, + destination_relative: &Path, + policy: ConflictPolicy, +) -> Result, TransferError> { + let final_path = destination.root().join(destination_relative); + let exists = match std::fs::symlink_metadata(&final_path) { + Ok(metadata) => Some(metadata), + Err(error) if error.kind() == io::ErrorKind::NotFound => None, + Err(error) => { + return Err(TransferError::io( + "failed to stat destination during planning", + error, + )); + } + }; + match (policy, exists) { + (ConflictPolicy::Skip, None) => Ok(Some(ConflictResolution::Fresh)), + (ConflictPolicy::Fail, None) => Ok(Some(ConflictResolution::Fresh)), + (ConflictPolicy::Overwrite, None) => Ok(Some(ConflictResolution::Fresh)), + (ConflictPolicy::Preserve, None) => Ok(Some(ConflictResolution::Fresh)), + (ConflictPolicy::Skip, Some(_)) => Ok(None), + (ConflictPolicy::Fail, Some(_)) => Err(TransferError::ConflictRejected { + path: destination_relative.to_path_buf(), + }), + (ConflictPolicy::Overwrite, Some(_)) => Ok(Some(ConflictResolution::Overwrite)), + (ConflictPolicy::Preserve, Some(_)) => Ok(Some(ConflictResolution::Preserved)), + } +} + +fn destination_is_atomic(destination: &MountedWriteAuthority) -> bool { + // Staged files live as siblings of the destination; the rename is atomic + // on every supported platform. Cross-filesystem moves are not in this + // module's scope, so the answer is always `true` while the destination + // authority is valid. + destination.validate().is_ok() +} + +/// The transfer executor. Holds the authorities and the plan; runs the +/// stages in order; rolls back on failure or cancellation. +pub struct TransferExecutor { + request: TransferRequest, + plan: TransferPlan, +} + +impl TransferExecutor { + /// Construct an executor from a previously planned request. + pub fn new(request: TransferRequest, plan: TransferPlan) -> Self { + Self { request, plan } + } + + /// Run the plan to completion, reporting progress through `progress`, + /// observing `cancellation` between stages, and rolling back committed + /// stages on any error. + pub fn run( + self, + progress: &mut dyn TransferProgress, + cancellation: &CancellationObserver, + ) -> Result { + let total_stages = self.plan.stage_count(); + let total_bytes = self.plan.total_bytes(); + let mut committed_stages: u32 = 0; + let mut bytes_so_far: u64 = 0; + let mut committed_files: Vec = Vec::new(); + + for (index, stage) in self.plan.stages().iter().enumerate() { + if cancellation.is_cancelled() { + self.rollback(&mut committed_files).map_err(|error| { + TransferError::RollbackFailed { + path: PathBuf::new(), + context: error.to_string(), + } + })?; + return Err(TransferError::Cancelled); + } + progress.on_stage_started(stage, index as u32, total_stages); + match stage { + Stage::CreateDirectory { + destination_relative_path, + } => { + self.execute_create_directory(destination_relative_path)?; + } + Stage::CopyFile { + source_relative_path, + destination_relative_path, + bytes, + atomic: _, + conflict, + } => { + let outcome = self.execute_copy_file( + source_relative_path, + destination_relative_path, + *bytes, + &mut bytes_so_far, + total_bytes, + index as u32, + total_stages, + progress, + cancellation, + )?; + let _ = outcome; + let _ = conflict; + committed_files.push(destination_relative_path.clone()); + } + Stage::RemoveFile { .. } => { + // RemoveFile stages are inserted only by the rollback path + // and never appear in a forward plan. Skip defensively. + } + } + committed_stages = committed_stages.saturating_add(1); + progress.on_stage_completed( + stage, + index as u32, + total_stages, + bytes_so_far, + total_bytes, + ); + } + + Ok(TransferSummary { + committed_stages, + bytes_copied: bytes_so_far, + completed: true, + }) + } + + fn execute_create_directory(&self, relative: &Path) -> Result<(), TransferError> { + self.request.destination.validate().map_err(|error| { + TransferError::authority(format!("destination not current: {error}")) + })?; + match self + .request + .destination + .create_relative_directory(relative, self.request.conflict_policy) + { + Ok(_) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { + // Idempotent: an existing directory is not an error. + let final_path = self.request.destination.root().join(relative); + match std::fs::symlink_metadata(&final_path) { + Ok(metadata) if metadata.is_dir() => Ok(()), + _ => Err(TransferError::io( + "directory creation failed with AlreadyExists", + error, + )), + } + } + Err(error) => Err(TransferError::io("create directory failed", error)), + } + } + + #[allow(clippy::too_many_arguments)] + fn execute_copy_file( + &self, + source_relative: &Path, + destination_relative: &Path, + declared_bytes: u64, + bytes_so_far: &mut u64, + total_bytes: u64, + stage_index: u32, + total_stages: u32, + progress: &mut dyn TransferProgress, + cancellation: &CancellationObserver, + ) -> Result { + self.request + .source + .validate() + .map_err(|error| TransferError::authority(format!("source not current: {error}")))?; + self.request.destination.validate().map_err(|error| { + TransferError::authority(format!("destination not current: {error}")) + })?; + + let source_result = self.request.source.with_relative_file( + source_relative, + |mut source_file| -> io::Result<(CommitOutcome, u64)> { + let staged = self + .request + .destination + .prepare_write_relative_file(destination_relative, self.request.conflict_policy) + .map_err(|error| TransferError::io("failed to stage destination file", error)) + .map_err(io::Error::other)?; + + const CHUNK: usize = 64 * 1024; + let mut buffer = vec![0u8; CHUNK]; + let mut copied: u64 = 0; + loop { + if cancellation.is_cancelled() { + let _ = staged.rollback(); + return Err(io::Error::new(io::ErrorKind::Interrupted, "cancelled")); + } + let read = source_file + .read(&mut buffer) + .map_err(|error| TransferError::io("failed to read source file", error)) + .map_err(io::Error::other)?; + if read == 0 { + break; + } + staged + .staged_file() + .write_all(&buffer[..read]) + .map_err(|error| TransferError::io("failed to write staged file", error)) + .map_err(io::Error::other)?; + copied = copied.saturating_add(read as u64); + *bytes_so_far = bytes_so_far.saturating_add(read as u64); + progress.on_bytes_copied(stage_index, total_stages, *bytes_so_far, total_bytes); + } + staged + .staged_file() + .flush() + .map_err(|error| TransferError::io("failed to flush staged file", error)) + .map_err(io::Error::other)?; + + if declared_bytes != 0 && copied != declared_bytes { + let _ = staged.rollback(); + return Err(io::Error::other(format!( + "source size {copied} differs from declared {declared_bytes} bytes" + ))); + } + + let outcome = staged + .commit() + .map_err(|error| TransferError::io("staged commit failed", error)) + .map_err(io::Error::other)?; + Ok((outcome, copied)) + }, + ); + let (outcome, copied) = match source_result { + Ok(value) => value, + Err(error) if error.kind() == io::ErrorKind::Interrupted => { + return Err(TransferError::Cancelled); + } + Err(error) => { + return Err(TransferError::io("failed to copy source file", error)); + } + }; + let _ = copied; + Ok(outcome) + } + + fn rollback(&self, committed_files: &mut Vec) -> io::Result<()> { + while let Some(relative) = committed_files.pop() { + self.request + .destination + .validate() + .map_err(|error| { + TransferError::authority(format!("destination not current: {error}")) + }) + .map_err(|error| io::Error::other(format!("{error:?}")))?; + self.request.destination.remove_relative_file(&relative)?; + } + Ok(()) + } +} + +trait WalkdirErrorExt { + fn into_io_error(self) -> Option; +} + +impl WalkdirErrorExt for walkdir::Error { + fn into_io_error(self) -> Option { + self.io_error() + .map(|error| io::Error::new(error.kind(), format!("walkdir error: {error}"))) + } +} + +fn _unused_os_string(_value: OsString) {} + +#[cfg(test)] +mod tests { + use super::*; + + use uuid::Uuid; + + use crate::local::write_authority::ConflictPolicy; + + fn unique(label: &str) -> PathBuf { + let path = + std::env::temp_dir().join(format!("tributary-transfer-{label}-{}", Uuid::new_v4())); + std::fs::create_dir_all(&path).expect("create root"); + path + } + + fn cleanup(path: &Path) { + let _ = std::fs::remove_dir_all(path); + } + + fn make_authority_pair(path: &Path) -> (Arc, MountedWriteAuthority) { + let mounted = MountedRootAuthority::acquire(path).expect("acquire mounted"); + let read = Arc::new(mounted); + let write = MountedWriteAuthority::from_mounted(Arc::clone(&read)); + (read, write) + } + + fn make_read_authority(path: &Path) -> Arc { + Arc::new(MountedRootAuthority::acquire(path).expect("acquire read authority")) + } + + fn write_source_file(path: &Path, contents: &[u8]) { + std::fs::create_dir_all(path.parent().unwrap()).expect("create parent"); + std::fs::write(path, contents).expect("write source"); + } + + #[test] + fn plan_resolves_file_into_single_copy_stage() { + let source_root = unique("plan-src"); + let destination_root = unique("plan-dst"); + write_source_file(&source_root.join("album/song.flac"), b"audio"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("album/song.flac"))], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + assert_eq!(plan.file_count(), 1); + assert!( + plan.directory_count() >= 1, + "album directory must be staged" + ); + let total = plan.total_bytes(); + assert!(total >= 5); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn capacity_budget_rejects_oversized_plan() { + let source_root = unique("budget-src"); + let destination_root = unique("budget-dst"); + write_source_file(&source_root.join("big.flac"), &[0u8; 100]); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("big.flac"))], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: Some(10), + recurse_directories: true, + }; + let error = TransferPlanner::new() + .plan(&request) + .expect_err("oversized plan must be rejected"); + assert!(matches!(error, TransferError::CapacityExceeded { .. })); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn plan_walks_directory_recursively() { + let source_root = unique("walk-src"); + let destination_root = unique("walk-dst"); + write_source_file(&source_root.join("album/a.flac"), b"a"); + write_source_file(&source_root.join("album/nested/b.flac"), b"b"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::new( + PathBuf::from("album"), + PathBuf::from("imported"), + )], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + assert_eq!(plan.file_count(), 2); + assert!( + plan.directory_count() >= 2, + "album and nested must be staged" + ); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn executor_copies_a_single_file() { + let source_root = unique("exec-src"); + let destination_root = unique("exec-dst"); + write_source_file(&source_root.join("song.flac"), b"copy me"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("song.flac"))], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + let observer = CancellationObserver::never_cancelled(); + let mut progress = (); + let summary = TransferExecutor::new(request, plan) + .run(&mut progress, &observer) + .expect("run"); + assert!(summary.completed); + let final_path = destination_root.join("song.flac"); + let bytes = std::fs::read(&final_path).expect("read final"); + assert_eq!(bytes, b"copy me"); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn executor_recursive_directory_copy() { + let source_root = unique("rec-src"); + let destination_root = unique("rec-dst"); + write_source_file(&source_root.join("album/a.flac"), b"a"); + write_source_file(&source_root.join("album/b.flac"), b"b"); + write_source_file(&source_root.join("album/nested/c.flac"), b"c"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::new( + PathBuf::from("album"), + PathBuf::from("imported"), + )], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + let observer = CancellationObserver::never_cancelled(); + let mut progress = (); + let summary = TransferExecutor::new(request, plan) + .run(&mut progress, &observer) + .expect("run"); + assert!(summary.completed); + assert_eq!( + std::fs::read(destination_root.join("imported/a.flac")).expect("read a"), + b"a" + ); + assert_eq!( + std::fs::read(destination_root.join("imported/b.flac")).expect("read b"), + b"b" + ); + assert_eq!( + std::fs::read(destination_root.join("imported/nested/c.flac")).expect("read c"), + b"c" + ); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn conflict_fail_rejects_existing_destination() { + let source_root = unique("fail-src"); + let destination_root = unique("fail-dst"); + write_source_file(&source_root.join("song.flac"), b"new"); + std::fs::write(destination_root.join("song.flac"), b"old").expect("write existing"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("song.flac"))], + conflict_policy: ConflictPolicy::Fail, + capacity_budget: None, + recurse_directories: true, + }; + let error = TransferPlanner::new() + .plan(&request) + .expect_err("fail policy must reject existing destination"); + assert!(matches!(error, TransferError::ConflictRejected { .. })); + cleanup(&source_root); + cleanup(&destination_root); + } + + struct ProgressRecorder { + stage_starts: u32, + stage_completes: u32, + byte_chunks: u32, + } + + impl ProgressRecorder { + fn new() -> Self { + Self { + stage_starts: 0, + stage_completes: 0, + byte_chunks: 0, + } + } + } + + impl TransferProgress for ProgressRecorder { + fn on_stage_started(&mut self, _stage: &Stage, _index: u32, _total: u32) { + self.stage_starts = self.stage_starts.saturating_add(1); + } + fn on_stage_completed( + &mut self, + _stage: &Stage, + _index: u32, + _total: u32, + _bytes_so_far: u64, + _total_bytes: u64, + ) { + self.stage_completes = self.stage_completes.saturating_add(1); + } + fn on_bytes_copied( + &mut self, + _stage_index: u32, + _total_stages: u32, + _bytes_so_far: u64, + _total_bytes: u64, + ) { + self.byte_chunks = self.byte_chunks.saturating_add(1); + } + } + + #[test] + fn progress_callback_reports_every_stage_and_chunk() { + let source_root = unique("progress-src"); + let destination_root = unique("progress-dst"); + // 200 KiB so the 64 KiB chunked copy yields multiple progress reports. + let payload = vec![0u8; 200 * 1024]; + write_source_file(&source_root.join("payload.bin"), &payload); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("payload.bin"))], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + let observer = CancellationObserver::never_cancelled(); + let mut progress = ProgressRecorder::new(); + let summary = TransferExecutor::new(request, plan) + .run(&mut progress, &observer) + .expect("run"); + assert!(summary.completed); + assert_eq!(progress.stage_starts, 1, "one copy stage should start"); + assert_eq!( + progress.stage_completes, 1, + "one copy stage should complete" + ); + assert!( + progress.byte_chunks >= 3, + "at least three progress chunks for 200 KiB" + ); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn overwrite_policy_replaces_existing_destination() { + let source_root = unique("overwrite-src"); + let destination_root = unique("overwrite-dst"); + write_source_file(&source_root.join("song.flac"), b"new"); + std::fs::write(destination_root.join("song.flac"), b"old").expect("write existing"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("song.flac"))], + conflict_policy: ConflictPolicy::Overwrite, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + let observer = CancellationObserver::never_cancelled(); + let mut progress = (); + let _ = TransferExecutor::new(request, plan) + .run(&mut progress, &observer) + .expect("run"); + assert_eq!( + std::fs::read(destination_root.join("song.flac")).expect("read final"), + b"new" + ); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn skip_policy_skips_existing_destination() { + let source_root = unique("skip-src"); + let destination_root = unique("skip-dst"); + write_source_file(&source_root.join("song.flac"), b"new"); + std::fs::write(destination_root.join("song.flac"), b"old").expect("write existing"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("song.flac"))], + conflict_policy: ConflictPolicy::Skip, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + assert_eq!( + plan.file_count(), + 0, + "skip policy should produce no copy stages" + ); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn empty_request_plans_to_no_stages() { + let source_root = unique("empty-src"); + let destination_root = unique("empty-dst"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + let request = TransferRequest { + source, + destination, + items: vec![], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + }; + let plan = TransferPlanner::new().plan(&request).expect("plan"); + assert!(plan.is_empty()); + assert_eq!(plan.stage_count(), 0); + assert_eq!(plan.total_bytes(), 0); + cleanup(&source_root); + cleanup(&destination_root); + } + + #[test] + fn absolute_path_in_request_is_rejected() { + let source_root = unique("abs-src"); + let destination_root = unique("abs-dst"); + let source = make_read_authority(&source_root); + let (_, destination) = make_authority_pair(&destination_root); + let request = TransferRequest { + source, + destination, + items: vec![TransferItem::same(PathBuf::from("/etc/passwd"))], + conflict_policy: ConflictPolicy::Preserve, + capacity_budget: None, + recurse_directories: true, + }; + let error = TransferPlanner::new() + .plan(&request) + .expect_err("absolute path must be rejected"); + assert!(matches!(error, TransferError::InvalidItemPath { .. })); + cleanup(&source_root); + cleanup(&destination_root); + } +} diff --git a/src/local/mod.rs b/src/local/mod.rs index 1c95b4b5..8070c2cf 100644 --- a/src/local/mod.rs +++ b/src/local/mod.rs @@ -10,9 +10,11 @@ pub mod resolver; pub mod rhythmbox_import; pub mod rhythmbox_migration; mod rhythmbox_smart_playlist; -mod root_authority; +#[allow(clippy::redundant_pub_crate)] +pub(crate) mod root_authority; pub mod server_playlist_browser; pub mod server_playlist_runtime; pub mod smart_rules; pub mod tag_parser; pub mod tag_writer; +pub mod write_authority; diff --git a/src/local/root_authority.rs b/src/local/root_authority.rs index c6a092dc..6cbdf91c 100644 --- a/src/local/root_authority.rs +++ b/src/local/root_authority.rs @@ -227,7 +227,7 @@ impl RetainedObject { impl BoundFile { /// Clone the already-authorized file handle for handle-based parsing. - pub(super) fn try_clone_file(&self) -> io::Result { + pub(crate) fn try_clone_file(&self) -> io::Result { self.object.validate_live()?; self.object.file.try_clone() } @@ -465,6 +465,63 @@ impl MountedRootAuthority { }) } + /// Open a real directory using only normal components relative to the + /// retained mounted root. Used by the write authority to bind a parent + /// directory before staging a temporary file in it. + pub(super) fn open_relative_directory(&self, relative: &Path) -> io::Result { + let components = strict_relative_components(relative)?; + self.validate()?; + let path = join_components(&self.root, &components); + let opened = + open_descendant_from_root(self, &path, &components, DescendantKind::Directory)?; + self.validate()?; + Ok(BoundDirectory { + lease_token: self.token, + path, + object: opened.object, + parent_guards: opened.parent_guards, + }) + } + + /// Bind the exact retained root itself as a directory. Used by the write + /// authority when a staged file is published directly beneath the root. + pub(super) fn bind_root_directory(&self) -> io::Result { + self.validate()?; + let file = self.root_handle.file.try_clone()?; + ensure_boundary(self.boundary, &file)?; + Ok(BoundDirectory { + lease_token: self.token, + path: self.root.clone(), + object: RetainedObject::new(file)?, + parent_guards: Vec::new(), + }) + } + + /// Return the unique token identifying this exact authority instance. + /// Other modules use this to reject evidence held by a different root. + pub(super) fn token(&self) -> Uuid { + self.token + } + + /// Open a regular file beneath the root and run `body` with a cloned + /// handle. The handle is revalidated before the call so a remount between + /// the call site and `body` produces a fail-closed error rather than + /// surfacing a stale file descriptor. + /// + /// The closure receives a `File` cloned from the retained handle; the + /// caller is responsible for any I/O and error propagation through its + /// own `Result` type. The bound is dropped when the closure returns, so + /// the caller must clone the handle again if it needs to outlive the + /// closure. The retained root and parent-chain handles are not exposed. + pub(crate) fn with_relative_file(&self, relative: &Path, body: F) -> io::Result + where + F: FnOnce(File) -> io::Result, + { + let bound = self.open_relative_regular_file(relative)?; + let file = bound.try_clone_file()?; + body(file) + } + /// Reopen the mount path and verify the exact retained root, filesystem /// boundary, ancestor chain, and platform mount generation. pub(crate) fn validate(&self) -> io::Result<()> { diff --git a/src/local/write_authority.rs b/src/local/write_authority.rs new file mode 100644 index 00000000..ddcaeed3 --- /dev/null +++ b/src/local/write_authority.rs @@ -0,0 +1,800 @@ +//! Retained write authority for one exact mounted filesystem. +//! +//! This module adds write capability on top of the read-only +//! [`MountedRootAuthority`]. The same retained root handle, parent chain, +//! mount generation, and filesystem boundary policy gate every byte written +//! beneath the mount. There is no shared-marker requirement; mounted authority +//! exists for ephemeral portable devices and replaces its session epoch on +//! relocation, pre-unmount, or removal. +//! +//! The intended consumer is the generic mounted-filesystem transfer planner +//! in [`crate::device::transfer`]. It must never be used to author a library +//! root (those keep their marker-backed +//! [`RootAuthorityLease`](super::root_authority::RootAuthorityLease) and +//! explicit database enrollment). +//! +//! File writes are staged: a sibling temporary file is created with +//! `O_CREAT | O_EXCL | O_NOFOLLOW` (Unix) or with the reparse-point attribute +//! rejected (Windows), then renamed atomically once +//! [`PreparedWriteTarget::commit`] is called. A rollback drops the staged +//! file. The destination filesystem is observed through the same retained +//! boundary, so a binder swap or remount between staging and commit is +//! detected and refused without surfacing a partial publish. + +use std::ffi::OsString; +use std::fmt; +use std::fs::File; +use std::io::{self, Write}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use uuid::Uuid; + +use super::root_authority::MountedRootAuthority; + +/// What the write authority should do when the destination of a write already +/// exists beneath the mount. +/// +/// `Skip` and `Fail` close the question for the whole transfer on a single +/// collision; `Overwrite` and `Preserve` permit the operation to proceed +/// without further prompt. Each variant is a typed policy, not a boolean flag, +/// so reviewers can grep call sites for the precise behavior at every +/// admission boundary. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ConflictPolicy { + /// Leave any existing destination untouched and skip the operation. + Skip, + /// Atomically replace the existing destination during commit. + Overwrite, + /// Choose a non-colliding name in the same directory and create anew. + Preserve, + /// Refuse the operation; transfer fails before any byte is written. + Fail, +} + +/// Outcome of resolving a conflict policy against the live filesystem. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum ConflictResolution { + /// Destination was absent; the staged file becomes a fresh write. + Fresh, + /// Destination existed; the staged file will replace it on commit. + Overwrite, + /// Destination existed; the staged file is written to a disambiguated name. + Preserved, +} + +/// A staged write below a [`MountedWriteAuthority`] ready for commit/rollback. +/// +/// The destination is held in a sibling temporary file. Until +/// [`commit`](Self::commit) is called the original destination is untouched, +/// so a partially-written staged file can be discarded without disturbing the +/// mount. Once commit fires, the rename is atomic on the same filesystem and +/// the staged file is gone. +pub struct PreparedWriteTarget { + lease_token: Uuid, + authority: Arc, + final_relative_path: PathBuf, + /// Absolute path of the staged temporary file. Sibling of the destination + /// so the rename is atomic on the same filesystem. + staged_path: PathBuf, + staged_file: File, + resolution: ConflictResolution, + committed: bool, +} + +impl fmt::Debug for PreparedWriteTarget { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("PreparedWriteTarget") + .field("target_relative_path", &self.final_relative_path) + .field("staged_path", &self.staged_path) + .field("resolution", &self.resolution) + .finish_non_exhaustive() + } +} + +impl PreparedWriteTarget { + /// Final relative path the staged file will be renamed to on commit. + pub fn target_relative_path(&self) -> &Path { + &self.final_relative_path + } + + /// How the conflict policy was resolved against the live filesystem. + pub fn resolution(&self) -> ConflictResolution { + self.resolution + } + + /// Borrow the staged file for reads (e.g. computing a digest). + pub fn staged_file(&self) -> &File { + &self.staged_file + } + + /// Append `bytes` to the staged file. + pub fn write_all(&mut self, bytes: &[u8]) -> io::Result<()> { + self.authority.validate()?; + self.staged_file.write_all(bytes)?; + self.staged_file.flush()?; + self.authority.validate()?; + Ok(()) + } + + /// Commit the staged file atomically to its destination. + /// + /// On Unix this is a single `rename(2)`; on Windows a `MoveFileExW` + /// replacement. The mount boundary is revalidated immediately before and + /// after the rename so a binder swap or remount between staging and + /// commit cannot authorise a partial publish. + pub fn commit(mut self) -> io::Result { + let final_path = self.authority.root().join(&self.final_relative_path); + self.authority.validate()?; + publish_atomic(&self.staged_path, &final_path)?; + self.authority.validate()?; + self.committed = true; + Ok(CommitOutcome { + relative_path: self.final_relative_path.clone(), + resolution: self.resolution, + }) + } + + /// Discard the staged file and any partial writes. + pub fn rollback(self) -> io::Result<()> { + if self.committed { + return Ok(()); + } + let outcome = rollback_staged(&self.staged_path); + let _ = self.authority.validate(); + outcome + } +} + +impl Drop for PreparedWriteTarget { + fn drop(&mut self) { + if self.committed { + return; + } + // Best-effort cleanup if the caller forgets to roll back explicitly. + let _ = rollback_staged(&self.staged_path); + } +} + +/// Detail of what `commit` actually published, for callers that need to log +/// or report the publish outcome. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct CommitOutcome { + /// The relative path beneath the retained mount that now names the data. + pub relative_path: PathBuf, + /// How the conflict policy was resolved against the live filesystem. + pub resolution: ConflictResolution, +} + +/// Retained write authority over one exact mounted filesystem. +/// +/// The underlying [`MountedRootAuthority`] is shared so the read-side scans +/// and the write-side commits always observe the same mount generation and +/// boundary. A successful transfer followed by a remount is detected on the +/// next `validate()` and produces a fail-closed error rather than attempting +/// a partial commit. +#[derive(Clone)] +pub struct MountedWriteAuthority { + mounted: Arc, +} + +impl MountedWriteAuthority { + /// Wrap an existing mounted authority to expose write API. + pub fn from_mounted(mounted: Arc) -> Self { + Self { mounted } + } + + /// Acquire a fresh write authority on the absolute mounted path. + pub fn acquire(root: &Path) -> io::Result { + let mounted = MountedRootAuthority::acquire(root)?; + Ok(Self { + mounted: Arc::new(mounted), + }) + } + + /// The exact native mount path retained by this authority. + pub fn root(&self) -> &Path { + self.mounted.root() + } + + /// Return the wrapped read authority for read operations. + pub fn mount(&self) -> &Arc { + &self.mounted + } + + /// Reverify the mount is still current. + pub fn validate(&self) -> io::Result<()> { + self.mounted.validate() + } + + /// Prepare a writable target below the root. The destination path is + /// checked against the conflict policy; a fresh, sibling temp file is + /// created with `O_CREAT | O_EXCL` so a concurrent writer cannot smuggle + /// a same-named file past publish. + pub fn prepare_write_relative_file( + &self, + relative: &Path, + policy: ConflictPolicy, + ) -> io::Result { + let components = strict_relative_components(relative)?; + self.mounted.validate()?; + let final_relative = assemble_relative(&components); + let final_path = self.mounted.root().join(&final_relative); + + // The destination parent directory must be opened through the + // retained authority so the boundary check matches the read path. + let parent_components = parent_components_of(&components); + if parent_components.as_os_str().is_empty() { + let _root_bound = self.mounted.bind_root_directory()?; + } else { + let _parent_bound = self.mounted.open_relative_directory(&parent_components)?; + } + + let (resolution, final_relative, staged_dir) = match policy { + ConflictPolicy::Skip if final_path.exists() => { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "destination exists and policy is Skip", + )); + } + ConflictPolicy::Fail if final_path.exists() => { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "destination exists and policy is Fail", + )); + } + ConflictPolicy::Skip | ConflictPolicy::Fail => { + (ConflictResolution::Fresh, final_relative, parent_components) + } + ConflictPolicy::Overwrite => ( + ConflictResolution::Overwrite, + final_relative, + parent_components, + ), + ConflictPolicy::Preserve => { + if final_path.exists() { + let (preserved_rel, preserved_components) = preserved_sibling_path( + self.mounted.root(), + &parent_components, + components.last().expect("non-empty"), + )?; + ( + ConflictResolution::Preserved, + preserved_rel, + preserved_components, + ) + } else { + (ConflictResolution::Fresh, final_relative, parent_components) + } + } + }; + + let staged_name = staging_leaf_name(); + let staged_path_abs = self.mounted.root().join(&staged_dir).join(&staged_name); + let staged_file = create_exclusive_staged_file(&staged_path_abs)?; + self.mounted.validate()?; + + Ok(PreparedWriteTarget { + lease_token: self.mounted.token(), + authority: Arc::clone(&self.mounted), + final_relative_path: final_relative, + staged_path: staged_path_abs, + staged_file, + resolution, + committed: false, + }) + } + + /// Create a directory beneath the mount and bind it for further writes. + pub fn create_relative_directory( + &self, + relative: &Path, + policy: ConflictPolicy, + ) -> io::Result { + let components = strict_relative_components(relative)?; + self.mounted.validate()?; + let final_path = self.mounted.root().join(assemble_relative(&components)); + + match std::fs::symlink_metadata(&final_path) { + Ok(metadata) => { + if !metadata.is_dir() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "path exists and is not a directory", + )); + } + match policy { + ConflictPolicy::Skip | ConflictPolicy::Fail => { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "directory exists and policy forbids overwriting", + )); + } + ConflictPolicy::Overwrite | ConflictPolicy::Preserve => {} + } + } + Err(error) if error.kind() == io::ErrorKind::NotFound => { + create_directory_atomic(self.mounted.root(), &components)?; + } + Err(error) => return Err(error), + } + self.mounted.validate()?; + let _bound = self + .mounted + .open_relative_directory(&assemble_relative(&components))?; + Ok(MountedDirectory { + lease_token: self.mounted.token(), + authority: Arc::clone(&self.mounted), + relative_path: assemble_relative(&components), + }) + } + + /// Remove a regular file atomically through the retained authority. + pub fn remove_relative_file(&self, relative: &Path) -> io::Result<()> { + let components = strict_relative_components(relative)?; + self.mounted.validate()?; + let final_path = self.mounted.root().join(assemble_relative(&components)); + let metadata = std::fs::symlink_metadata(&final_path)?; + if metadata.is_dir() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "refusing to remove a directory through remove_relative_file", + )); + } + std::fs::remove_file(&final_path)?; + self.mounted.validate()?; + Ok(()) + } + + /// Remove an empty directory atomically through the retained authority. + pub fn remove_relative_directory(&self, relative: &Path) -> io::Result<()> { + let components = strict_relative_components(relative)?; + self.mounted.validate()?; + let final_path = self.mounted.root().join(assemble_relative(&components)); + let metadata = std::fs::symlink_metadata(&final_path)?; + if !metadata.is_dir() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "refusing to remove a non-directory through remove_relative_directory", + )); + } + std::fs::remove_dir(&final_path)?; + self.mounted.validate()?; + Ok(()) + } +} + +/// A directory created by [`MountedWriteAuthority::create_relative_directory`]. +pub struct MountedDirectory { + lease_token: Uuid, + authority: Arc, + relative_path: PathBuf, +} + +impl MountedDirectory { + /// Return the relative path of this directory. + pub fn relative_path(&self) -> &Path { + &self.relative_path + } + + /// Prepare a writable file directly inside this directory. + pub fn prepare_write_in_directory( + &self, + name: &str, + policy: ConflictPolicy, + ) -> io::Result { + let mut relative = self.relative_path.clone(); + relative.push(name); + MountedWriteAuthority::from_mounted(Arc::clone(&self.authority)) + .prepare_write_relative_file(&relative, policy) + } +} + +fn strict_relative_components(relative: &Path) -> io::Result> { + if relative.is_absolute() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "write target path must be relative", + )); + } + let mut components = Vec::new(); + for component in relative.components() { + match component { + std::path::Component::Normal(value) => components.push(value.to_os_string()), + _ => { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "write target path contains a non-normal component", + )) + } + } + } + if components.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "write target path requires a path below the mount root", + )); + } + Ok(components) +} + +fn assemble_relative(components: &[OsString]) -> PathBuf { + let mut path = PathBuf::new(); + for component in components { + path.push(component); + } + path +} + +fn parent_components_of(components: &[OsString]) -> PathBuf { + let mut path = PathBuf::new(); + for component in &components[..components.len().saturating_sub(1)] { + path.push(component); + } + path +} + +fn staging_leaf_name() -> OsString { + let token = Uuid::new_v4(); + let mut name = OsString::from(".tributary-stage-"); + name.push(token.to_string()); + name.push(".tmp"); + name +} + +fn preserved_sibling_path( + root: &Path, + parent_components: &Path, + leaf: &OsString, +) -> io::Result<(PathBuf, PathBuf)> { + let parent_abs = if parent_components.as_os_str().is_empty() { + root.to_path_buf() + } else { + root.join(parent_components) + }; + let entries = match std::fs::read_dir(&parent_abs) { + Ok(entries) => entries, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + return Err(io::Error::new( + io::ErrorKind::NotFound, + "preserve policy requires an existing parent directory", + )); + } + Err(error) => return Err(error), + }; + let existing: Vec = entries + .filter_map(|entry| entry.ok()) + .filter_map(|entry| entry.file_name().into_string().ok()) + .collect(); + let original = leaf.to_string_lossy().into_owned(); + let (stem, ext) = match original.rsplit_once('.') { + Some((stem, ext)) if !stem.is_empty() => (stem.to_string(), Some(ext.to_string())), + _ => (original.clone(), None), + }; + for index in 1..=u32::MAX { + let candidate = match &ext { + Some(ext) => format!("{stem} ({index}).{ext}"), + None => format!("{stem} ({index})"), + }; + if !existing.iter().any(|name| name == &candidate) { + // Compose the final relative path (parent + leaf candidate). + let mut relative = PathBuf::new(); + if !parent_components.as_os_str().is_empty() { + relative.push(parent_components); + } + relative.push(&candidate); + // The directory used to stage the temp file is the same parent + // directory so the rename stays atomic on one filesystem. + return Ok((relative, parent_components.to_path_buf())); + } + } + Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "no preserved name available for conflict", + )) +} + +fn create_directory_atomic(root: &Path, components: &[OsString]) -> io::Result<()> { + let mut path = root.to_path_buf(); + for component in components { + path.push(component); + match std::fs::create_dir(&path) { + Ok(()) => {} + Err(error) if error.kind() == io::ErrorKind::AlreadyExists => { + let metadata = std::fs::symlink_metadata(&path)?; + if !metadata.is_dir() { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "intermediate path is not a directory", + )); + } + } + Err(error) => return Err(error), + } + } + Ok(()) +} + +#[cfg(unix)] +fn create_exclusive_staged_file(path: &Path) -> io::Result { + use rustix::fs::{Mode, OFlags}; + + let leaf = path.file_name().ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidInput, + "staged file path is missing a leaf", + ) + })?; + let parent = path.parent().ok_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidInput, + "staged file path is missing a parent", + ) + })?; + let parent_file = File::open(parent)?; + let descriptor = rustix::fs::openat( + &parent_file, + leaf, + OFlags::WRONLY + | OFlags::CREATE + | OFlags::EXCL + | OFlags::CLOEXEC + | OFlags::NOFOLLOW + | OFlags::NOCTTY, + Mode::from_bits_truncate(0o600), + ) + .map_err(io::Error::from)?; + Ok(File::from(descriptor)) +} + +#[cfg(windows)] +fn create_exclusive_staged_file(path: &Path) -> io::Result { + use std::fs::OpenOptions; + use std::os::windows::fs::OpenOptionsExt; + + use windows_sys::Win32::Storage::FileSystem::{ + FILE_ATTRIBUTE_NORMAL, FILE_FLAG_OPEN_REPARSE_POINT, FILE_SHARE_READ, + }; + + OpenOptions::new() + .write(true) + .create_new(true) + .share_mode(FILE_SHARE_READ) + .custom_flags(FILE_ATTRIBUTE_NORMAL | FILE_FLAG_OPEN_REPARSE_POINT) + .open(path) +} + +#[cfg(not(any(unix, windows)))] +fn create_exclusive_staged_file(_path: &Path) -> io::Result { + Err(io::Error::new( + io::ErrorKind::Unsupported, + "write authority is unsupported on this platform", + )) +} + +fn rollback_staged(staged_path: &Path) -> io::Result<()> { + match std::fs::remove_file(staged_path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(error), + } +} + +fn publish_atomic(staged_path: &Path, final_path: &Path) -> io::Result<()> { + // POSIX rename is atomic on the same filesystem. Windows std::fs::rename + // uses MoveFileExW with MOVEFILE_REPLACE_EXISTING semantics, which is + // likewise atomic on the same volume. + std::fs::rename(staged_path, final_path) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn unique_root(label: &str) -> PathBuf { + let path = std::env::temp_dir().join(format!( + "tributary-write-authority-{label}-{}", + Uuid::new_v4() + )); + std::fs::create_dir_all(&path).expect("create root"); + path + } + + fn cleanup(path: &Path) { + let _ = std::fs::remove_dir_all(path); + } + + #[test] + fn fresh_write_commits_atomically() { + let root = unique_root("fresh"); + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + + let mut staged = authority + .prepare_write_relative_file(Path::new("song.flac"), ConflictPolicy::Fail) + .expect("prepare staged file"); + staged.write_all(b"audio payload").expect("write payload"); + + let outcome = staged.commit().expect("commit staged file"); + assert_eq!(outcome.resolution, ConflictResolution::Fresh); + assert_eq!( + std::fs::read(root.join("song.flac")).expect("read final"), + b"audio payload" + ); + + cleanup(&root); + } + + #[test] + fn skip_policy_rejects_when_destination_exists() { + let root = unique_root("skip"); + std::fs::write(root.join("song.flac"), b"existing").expect("write existing"); + + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + let error = authority + .prepare_write_relative_file(Path::new("song.flac"), ConflictPolicy::Skip) + .expect_err("skip policy must reject existing destination"); + assert_eq!(error.kind(), io::ErrorKind::AlreadyExists); + + cleanup(&root); + } + + #[test] + fn overwrite_policy_replaces_final_file() { + let root = unique_root("overwrite"); + std::fs::write(root.join("song.flac"), b"old").expect("write existing"); + + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + let mut staged = authority + .prepare_write_relative_file(Path::new("song.flac"), ConflictPolicy::Overwrite) + .expect("prepare overwrite"); + staged.write_all(b"new").expect("write new"); + staged.commit().expect("commit overwrite"); + + assert_eq!( + std::fs::read(root.join("song.flac")).expect("read final"), + b"new" + ); + + cleanup(&root); + } + + #[test] + fn preserve_policy_writes_to_disambiguated_name() { + let root = unique_root("preserve"); + std::fs::write(root.join("song.flac"), b"first").expect("write first"); + + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + let mut staged = authority + .prepare_write_relative_file(Path::new("song.flac"), ConflictPolicy::Preserve) + .expect("prepare preserve"); + staged.write_all(b"second").expect("write second"); + let outcome = staged.commit().expect("commit preserve"); + + assert_eq!(outcome.resolution, ConflictResolution::Preserved); + assert_eq!( + std::fs::read(root.join("song.flac")).expect("read original"), + b"first" + ); + assert_eq!( + std::fs::read(root.join(&outcome.relative_path)).expect("read preserved"), + b"second" + ); + + cleanup(&root); + } + + #[test] + fn rollback_removes_staged_file() { + let root = unique_root("rollback"); + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + + let mut staged = authority + .prepare_write_relative_file(Path::new("song.flac"), ConflictPolicy::Fail) + .expect("prepare staged"); + staged.write_all(b"partial").expect("write partial"); + let staged_path = staged.staged_path.clone(); + // Sanity: the staged file actually exists before rollback. + assert!(staged_path.exists()); + staged.rollback().expect("rollback staged"); + assert!(!staged_path.exists()); + assert!(!root.join("song.flac").exists()); + + cleanup(&root); + } + + #[test] + fn cross_mount_path_is_rejected() { + let root = unique_root("cross-mount"); + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + + let error = authority + .prepare_write_relative_file(Path::new("../outside.flac"), ConflictPolicy::Fail) + .expect_err("parent path must be rejected"); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + + let error = authority + .prepare_write_relative_file(Path::new("/etc/passwd"), ConflictPolicy::Fail) + .expect_err("absolute path must be rejected"); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + + cleanup(&root); + } + + #[test] + fn directory_creation_and_file_writes_combine() { + let root = unique_root("dirs"); + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + + let bound = authority + .create_relative_directory(Path::new("album"), ConflictPolicy::Fail) + .expect("create album dir"); + assert_eq!(bound.relative_path(), Path::new("album")); + + let mut staged = bound + .prepare_write_in_directory("song.flac", ConflictPolicy::Fail) + .expect("prepare file under dir"); + staged.write_all(b"nested").expect("write nested"); + staged.commit().expect("commit nested"); + + assert_eq!( + std::fs::read(root.join("album/song.flac")).expect("read nested"), + b"nested" + ); + + cleanup(&root); + } + + #[test] + fn prepared_target_resolves_only_one_preserved_name() { + let root = unique_root("preserve-twice"); + std::fs::write(root.join("song.flac"), b"original").expect("write original"); + + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + + let mut first = authority + .prepare_write_relative_file(Path::new("song.flac"), ConflictPolicy::Preserve) + .expect("prepare first preserve"); + first.write_all(b"a").expect("write first"); + let first_outcome = first.commit().expect("commit first"); + + let mut second = authority + .prepare_write_relative_file(Path::new("song.flac"), ConflictPolicy::Preserve) + .expect("prepare second preserve"); + second.write_all(b"b").expect("write second"); + let second_outcome = second.commit().expect("commit second"); + + assert_ne!(first_outcome.relative_path, second_outcome.relative_path); + let names: Vec = std::fs::read_dir(&root) + .expect("read dir") + .filter_map(|entry| entry.ok()) + .filter_map(|entry| entry.file_name().into_string().ok()) + .collect(); + assert_eq!(names.len(), 3); + assert!(names.iter().any(|name| name == "song.flac")); + assert!(names.iter().any(|name| name == "song (1).flac")); + assert!(names.iter().any(|name| name == "song (2).flac")); + + cleanup(&root); + } + + #[test] + fn remove_relative_file_only_accepts_regular_files() { + let root = unique_root("remove-file"); + std::fs::write(root.join("song.flac"), b"data").expect("write file"); + std::fs::create_dir(root.join("album")).expect("create album"); + + let authority = MountedWriteAuthority::acquire(&root).expect("acquire write authority"); + authority + .remove_relative_file(Path::new("song.flac")) + .expect("remove file"); + assert!(!root.join("song.flac").exists()); + + let error = authority + .remove_relative_file(Path::new("album")) + .expect_err("directory must be rejected"); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + + cleanup(&root); + } +} diff --git a/src/source_lifecycle.rs b/src/source_lifecycle.rs index 97de8818..47ce763b 100644 --- a/src/source_lifecycle.rs +++ b/src/source_lifecycle.rs @@ -458,6 +458,13 @@ impl fmt::Debug for CancellationObserver { } impl CancellationObserver { + /// Construct a never-cancelled observer. Useful for tests and for code + /// paths that accept a cancellation handle but have no upstream source. + pub fn never_cancelled() -> Self { + let (_sender, receiver) = watch::channel(false); + Self { receiver } + } + pub fn is_cancelled(&self) -> bool { *self.receiver.borrow() }