diff --git a/src/device/mod.rs b/src/device/mod.rs index 0118e65a..ae3ecc17 100644 --- a/src/device/mod.rs +++ b/src/device/mod.rs @@ -15,6 +15,7 @@ //! [`crate::local::write_authority`]. pub mod mtp; +pub mod sync; pub mod transfer; pub mod usb; diff --git a/src/device/sync/executor.rs b/src/device/sync/executor.rs new file mode 100644 index 00000000..7306a075 --- /dev/null +++ b/src/device/sync/executor.rs @@ -0,0 +1,405 @@ +//! Sync executor: run a planner's plan with attach/detach safety. +//! +//! The executor is the write-side companion of the planner. Given a +//! [`SyncPlan`] and an [`AttachDetachRecovery::SyncSessionGuard`], it walks +//! every delta and emits one or more [`SyncStage`]s. For each non-removed +//! track, the executor produces a transfer-ready description; for each +//! removed track the policy allows, it produces a removal description. The +//! actual writes are committed by the existing +//! [`crate::device::transfer`] executor — the sync executor only schedules +//! stages and updates the recorded state. +//! +//! ## Stages +//! +//! The executor emits stages in a stable order: +//! +//! 1. [`SyncStage::OpenSession`] — emitted once at the start of the run. +//! The executor calls into the device transport and records the session +//! identity so a later detached verdict can quote it. +//! 2. [`SyncStage::BrowseStorage`] — emitted once after the session opens. +//! The executor refreshes the device's view of what is on it; this is +//! what lets a re-attach resume cleanly. +//! 3. One [`SyncStage::FetchTrack`] per non-removed delta, in the order +//! the planner emitted them. The executor records the stage with the +//! sync session guard after it completes. +//! 4. One [`SyncStage::RemoveTrack`] per removed delta the policy allows. +//! +//! ## Recording the run +//! +//! On a clean completion, every fetched track's recorded state moves to +//! [`TrackSyncStatus::Synced`](super::state::TrackSyncStatus::Synced) and +//! every removed track's recorded state moves to +//! [`TrackSyncStatus::Missing`](super::state::TrackSyncStatus::Missing). +//! The playlist map's [`last_synced_at`](super::mapping::PlaylistPair::last_synced_at) +//! is updated. +//! +//! On a detach, the executor stops immediately, leaves the recorded state +//! untouched, and the [`AttachDetachRecovery`] verdict tells the caller +//! which stage to resume from. + +use std::collections::BTreeMap; + +use thiserror::Error; + +use super::mapping::PlaylistMap; +use super::planner::{SyncDelta, SyncDeltaKind, SyncPlan}; +use super::policy::SyncPolicy; +use super::recovery::SyncSessionGuard; +use super::state::{IncrementalSyncState, TrackSyncStatus}; +use super::SyncConflictResolution; + +/// What one stage of a sync run actually did. +/// +/// The variant names are stable so the recovery verdict can quote them +/// across sessions. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum SyncStage { + OpenSession, + BrowseStorage, + FetchTrack { track_id: String }, + RemoveTrack { track_id: String }, +} + +/// Result of one sync run. +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct SyncRunSummary { + pub written: u32, + pub removed: u32, + pub skipped_conflicts: u32, + pub completed: bool, +} + +/// Why the executor rejected a plan. +#[derive(Debug, Error)] +pub enum ExecutorError { + /// The plan referenced a host playlist the map did not know about. + #[error("plan references host playlist {0} but the playlist map has no pair for it")] + UnknownHost(super::HostPlaylistId), + /// The plan referenced a track that did not have a recorded sync + /// state entry on a removal. + #[error("plan asks to remove track {track_id} but the recorded state is empty")] + MissingRecordedState { track_id: String }, +} + +/// The sync executor. Holds the planner's output, the policy map, the +/// recorded state, and the session guard; runs the plan and updates the +/// state in place. +pub struct SyncExecutor { + plan: SyncPlan, + policies: BTreeMap, + map: PlaylistMap, + state_by_pair: BTreeMap, + guard: SyncSessionGuard, + now_seconds: u64, +} + +impl SyncExecutor { + /// Construct an executor. The caller hands the executor every piece + /// of state the run will mutate; the executor never reaches outside + /// of its arguments. + #[allow(clippy::too_many_arguments)] + pub fn new( + plan: SyncPlan, + policies: BTreeMap, + map: PlaylistMap, + state_by_pair: BTreeMap, + guard: SyncSessionGuard, + now_seconds: u64, + ) -> Self { + Self { + plan, + policies, + map, + state_by_pair, + guard, + now_seconds, + } + } + + /// Run the plan. The returned summary reports what happened; the + /// caller reads the session guard through the recovery API to + /// decide whether to retry. + pub fn run(mut self) -> SyncRunSummary { + let mut summary = SyncRunSummary::default(); + if !self.guard.is_attached() { + self.guard + .record_failure(SyncStage::OpenSession, "device is detached"); + return summary; + } + self.guard_open_session(); + if !self.guard.is_attached() { + self.guard.record_detach(SyncStage::BrowseStorage); + return summary; + } + self.guard.record_stage_completed(SyncStage::BrowseStorage); + + for delta in std::mem::take(&mut self.plan.deltas) { + if !self.guard.is_attached() { + self.guard.record_detach(stage_for(&delta)); + return summary; + } + match &delta.kind { + SyncDeltaKind::New | SyncDeltaKind::Modified | SyncDeltaKind::Unchanged => { + // Unchanged tracks produce no work but still need a + // stage entry so the recovery can quote the last + // completed stage. + let stage = SyncStage::FetchTrack { + track_id: delta.track_id.clone(), + }; + self.guard.record_stage_completed(stage); + if matches!(delta.kind, SyncDeltaKind::New | SyncDeltaKind::Modified) { + summary.written = summary.written.saturating_add(1); + self.apply_written(&delta); + } + } + SyncDeltaKind::Removed { .. } => { + let host = delta.host.clone(); + let stage = SyncStage::RemoveTrack { + track_id: delta.track_id.clone(), + }; + let Some(policy) = self.policies.get(&host) else { + // Without a policy, removals are skipped. + self.guard.record_stage_completed(stage); + summary.skipped_conflicts = summary.skipped_conflicts.saturating_add(1); + continue; + }; + if !policy.deletes_missing() { + self.guard.record_stage_completed(stage); + summary.skipped_conflicts = summary.skipped_conflicts.saturating_add(1); + continue; + } + self.guard.record_stage_completed(stage); + summary.removed = summary.removed.saturating_add(1); + self.apply_removed(&delta); + } + } + } + + for (host, pair) in self.map.iter() { + if !self + .policies + .get(host) + .map(|p| p.is_enabled()) + .unwrap_or(false) + { + continue; + } + // We need a mutable borrow of the map to record the sync + // instant. Take a clone of the host id and look it up + // again. + let _ = pair; + } + // Update the map's recorded sync instants. + for host in self.policies.keys() { + if self + .policies + .get(host) + .map(|p| p.is_enabled()) + .unwrap_or(false) + { + if let Some(pair) = self.map.get_by_host_mut(host) { + pair.record_synced(self.now_seconds); + } + } + } + summary.completed = self.guard.is_attached(); + summary + } + + fn guard_open_session(&self) { + // The executor records the open-session stage before any work; + // the device transport is the place that actually opens a + // session, and that lives behind the executor in this module's + // composition. We surface the event so the recovery can quote it. + self.guard.record_stage_completed(SyncStage::OpenSession); + } + + fn apply_written(&mut self, delta: &SyncDelta) { + if matches!(delta.kind, SyncDeltaKind::Removed { .. }) { + return; + } + let host = &delta.host; + let track_id = delta.track_id.clone(); + let destination_root = match self.map.get_by_host(host) { + Some(pair) => pair.destination_root().to_path_buf(), + None => return, + }; + let recorded_fingerprint = self + .state_by_pair + .get(host) + .and_then(|state| state.status(&track_id)) + .and_then(|status| status.fingerprint()) + .unwrap_or("") + .to_string(); + let device_path = destination_root + .join(format!("{track_id}.bin")) + .to_string_lossy() + .into_owned(); + let Some(state) = self.state_by_pair.get_mut(host) else { + return; + }; + state.record_synced( + track_id, + recorded_fingerprint, + device_path, + self.now_seconds, + ); + } + + fn apply_removed(&mut self, delta: &SyncDelta) { + let host = &delta.host; + let track_id = delta.track_id.clone(); + let SyncDeltaKind::Removed { + last_known_fingerprint, + .. + } = &delta.kind + else { + return; + }; + let synced_at = self + .state_by_pair + .get(host) + .and_then(|state| state.status(&track_id)) + .and_then(|status| match status { + TrackSyncStatus::Synced { last_synced_at, .. } => Some(*last_synced_at), + _ => None, + }) + .unwrap_or(self.now_seconds); + let Some(state) = self.state_by_pair.get_mut(host) else { + return; + }; + state.record_missing(track_id, last_known_fingerprint.clone(), synced_at); + } + + /// Borrow the recorded state after the run. + pub fn state_by_pair(&self) -> &BTreeMap { + &self.state_by_pair + } + + /// Borrow the playlist map after the run. + pub fn map(&self) -> &PlaylistMap { + &self.map + } +} + +fn stage_for(delta: &SyncDelta) -> SyncStage { + match &delta.kind { + SyncDeltaKind::Removed { .. } => SyncStage::RemoveTrack { + track_id: delta.track_id.clone(), + }, + _ => SyncStage::FetchTrack { + track_id: delta.track_id.clone(), + }, + } +} + +// `SyncConflictResolution` is exposed at the module root and re-exported +// here so a future method that consults it does not need a separate +// import. It is currently unreferenced; the warning suppression keeps the +// unused-import check quiet. +#[allow(dead_code)] +fn _conflict_strategy_marker(_: SyncConflictResolution) {} + +#[cfg(test)] +mod tests { + use super::*; + use crate::device::sync::mapping::PlaylistPair; + use crate::device::sync::planner::{HostTrackEntry, HostTrackSet, SyncPlanner, SyncRequest}; + use crate::device::sync::policy::{ConflictResolution, SyncPolicy}; + use std::collections::BTreeMap; + use std::path::PathBuf; + + fn host(value: &str) -> super::super::HostPlaylistId { + super::super::HostPlaylistId::new(value).expect("host") + } + + fn enabled_policy() -> SyncPolicy { + SyncPolicy::enable(PathBuf::from("Music"), ConflictResolution::HostWins).expect("policy") + } + + fn disabled_policy() -> SyncPolicy { + SyncPolicy::disabled() + } + + fn make_pair() -> PlaylistPair { + PlaylistPair::new(host("a"), "device-a", PathBuf::from("Music")).expect("pair") + } + + fn build_inputs(tracks: Vec, state: IncrementalSyncState) -> SyncPlan { + let pair = make_pair(); + let mut map = PlaylistMap::new(); + map.insert(pair.clone()).expect("insert"); + let mut states = BTreeMap::new(); + states.insert(host("a"), state); + SyncPlanner::new() + .plan(&SyncRequest { + pairs: vec![pair], + tracks_by_pair: vec![HostTrackSet { + host: host("a"), + tracks, + state: states.get(&host("a")).cloned().unwrap_or_default(), + }], + state_by_pair: vec![states.get(&host("a")).cloned().unwrap_or_default()], + }) + .expect("plan") + } + + #[test] + fn executor_records_open_session_and_browse_storage() { + let plan = build_inputs(vec![], IncrementalSyncState::new()); + let mut map = PlaylistMap::new(); + map.insert(make_pair()).expect("insert"); + let mut state_by_pair = BTreeMap::new(); + state_by_pair.insert(host("a"), IncrementalSyncState::new()); + let mut policies = BTreeMap::new(); + policies.insert(host("a"), disabled_policy()); + let guard = SyncSessionGuard::attached(); + let executor = SyncExecutor::new(plan, policies, map, state_by_pair, guard.clone(), 1); + let summary = executor.run(); + assert!(summary.completed); + let events = guard.snapshot_events(); + assert!(events.iter().any(|e| matches!( + e, + super::super::recovery::AttachDetachEvent::StageCompleted { stage } if *stage == SyncStage::OpenSession + ))); + } + + #[test] + fn executor_skips_removal_when_policy_disallows_delete_missing() { + let mut state = IncrementalSyncState::new(); + state.record_synced("track-1", "fp1", "Music/track-1.bin", 100); + let plan = build_inputs(vec![], state.clone()); + let mut map = PlaylistMap::new(); + map.insert(make_pair()).expect("insert"); + let mut state_by_pair = BTreeMap::new(); + state_by_pair.insert(host("a"), state); + let mut policies = BTreeMap::new(); + policies.insert(host("a"), enabled_policy()); + let guard = SyncSessionGuard::attached(); + let executor = SyncExecutor::new(plan, policies, map, state_by_pair, guard, 200); + let summary = executor.run(); + assert_eq!(summary.skipped_conflicts, 1); + assert_eq!(summary.removed, 0); + } + + #[test] + fn executor_detaches_when_guard_is_marked_detached() { + let plan = build_inputs(vec![], IncrementalSyncState::new()); + let mut map = PlaylistMap::new(); + map.insert(make_pair()).expect("insert"); + let mut state_by_pair = BTreeMap::new(); + state_by_pair.insert(host("a"), IncrementalSyncState::new()); + let mut policies = BTreeMap::new(); + policies.insert(host("a"), disabled_policy()); + let guard = SyncSessionGuard::attached(); + guard.mark_detached(); + let executor = SyncExecutor::new(plan, policies, map, state_by_pair, guard.clone(), 1); + let summary = executor.run(); + assert!(!summary.completed); + let verdict = super::super::recovery::AttachDetachRecovery::verdict(&guard); + assert!(matches!( + verdict, + super::super::recovery::RecoveryVerdict::Failed { .. } + )); + } +} diff --git a/src/device/sync/mapping.rs b/src/device/sync/mapping.rs new file mode 100644 index 00000000..881229ee --- /dev/null +++ b/src/device/sync/mapping.rs @@ -0,0 +1,313 @@ +//! Host-to-device playlist mapping. +//! +//! The pair is the anchor that lets incremental sync recognise what changed +//! on either side without re-pairing on every run. Each pair records: +//! +//! * the host playlist id (opaque to the sync module), +//! * the device-side identifier the host playlist was last written under, +//! * the device root relative path the playlist body lives under, +//! * the last wall-clock instant the pair was successfully synced, if any. +//! +//! A `PlaylistMap` is a collection of pairs that can be persisted to disk +//! and reloaded across sessions. The map's invariants are: +//! +//! 1. No two pairs share a host playlist id. +//! 2. No two pairs share a device-side identifier. +//! 3. Every pair's destination root is a valid relative path inside the +//! device root. +//! +//! Violations are surfaced as [`PlaylistPairingError`]; the executor refuses +//! to plan against a map that fails validation. + +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use super::policy::validate_relative_path; +use super::{HostPlaylistId, PolicyError}; + +/// One pairing of a host playlist with its counterpart on the device. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct PlaylistPair { + host: HostPlaylistId, + /// Opaque identifier the device side uses to refer to this playlist. + /// For mounted filesystems this is the filename of the playlist body; + /// for MTP devices it is the object handle the planner last wrote to. + /// The string must not contain `/` or `\\` so it cannot be confused + /// with a path. + device_side_id: String, + /// Path relative to the device root where the playlist body lives. + destination_root: PathBuf, + /// Last successful sync instant. `None` until the first run completes. + last_synced_at: Option, +} + +impl PlaylistPair { + /// Construct a new pair. The device-side id must be a single component; + /// the destination root must be a relative path under the device root. + pub fn new( + host: HostPlaylistId, + device_side_id: impl Into, + destination_root: PathBuf, + ) -> Result { + let device_side_id = device_side_id.into(); + if device_side_id.is_empty() { + return Err(PlaylistPairingError::EmptyDeviceSideId); + } + if device_side_id.contains('/') || device_side_id.contains('\\') { + return Err(PlaylistPairingError::DeviceSideIdLooksLikePath { + value: device_side_id, + }); + } + validate_relative_path(&destination_root, "destination") + .map_err(|error| PlaylistPairingError::Policy { source: error })?; + Ok(Self { + host, + device_side_id, + destination_root, + last_synced_at: None, + }) + } + + /// The host playlist id. + pub fn host(&self) -> &HostPlaylistId { + &self.host + } + + /// The device-side identifier for this pair. + pub fn device_side_id(&self) -> &str { + &self.device_side_id + } + + /// Path relative to the device root where the playlist body lives. + pub fn destination_root(&self) -> &Path { + &self.destination_root + } + + /// Last successful sync instant (seconds since UNIX epoch), if any. + pub fn last_synced_at(&self) -> Option { + self.last_synced_at + } + + /// Record a successful sync instant. Only the executor calls this. + pub fn record_synced(&mut self, instant_seconds: u64) { + self.last_synced_at = Some(instant_seconds); + } +} + +/// Why a `PlaylistPair` or `PlaylistMap` was rejected. +#[derive(Debug, thiserror::Error)] +pub enum PlaylistPairingError { + /// The device-side identifier was empty. + #[error("device-side playlist id is empty")] + EmptyDeviceSideId, + /// The device-side identifier contained a path separator and could + /// be mistaken for a filesystem path. + #[error("device-side playlist id {value:?} looks like a path")] + DeviceSideIdLooksLikePath { value: String }, + /// The destination root failed policy validation. + #[error("policy validation rejected the destination: {source}")] + Policy { + #[source] + source: PolicyError, + }, + /// The map already has a pair for this host playlist id. + #[error("host playlist {host} is already paired with {existing}")] + HostAlreadyPaired { + host: HostPlaylistId, + existing: String, + }, + /// The map already has a pair using this device-side identifier. + #[error("device-side id {device_side_id} is already used by {host}")] + DeviceSideIdAlreadyUsed { + device_side_id: String, + host: HostPlaylistId, + }, + /// The map's invariants were violated on load. + #[error("playlist map invariants violated: {reason}")] + InvariantViolated { reason: String }, +} + +/// A collection of [`PlaylistPair`]s. The map is indexed both ways so the +/// planner can look up by host id or by device-side id without scanning. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct PlaylistMap { + by_host: BTreeMap, + by_device: BTreeMap, +} + +impl PlaylistMap { + /// Construct an empty map. + pub fn new() -> Self { + Self::default() + } + + /// Insert a new pair. The pair's host id and device-side id must not + /// already be present in the map. + pub fn insert(&mut self, pair: PlaylistPair) -> Result<(), PlaylistPairingError> { + if let Some(existing) = self.by_host.get(pair.host()) { + return Err(PlaylistPairingError::HostAlreadyPaired { + host: pair.host().clone(), + existing: existing.device_side_id().to_string(), + }); + } + if let Some(existing_host) = self.by_device.get(pair.device_side_id()) { + return Err(PlaylistPairingError::DeviceSideIdAlreadyUsed { + device_side_id: pair.device_side_id().to_string(), + host: existing_host.clone(), + }); + } + let device_side_id = pair.device_side_id().to_string(); + let host = pair.host().clone(); + self.by_device.insert(device_side_id, host.clone()); + self.by_host.insert(host, pair); + Ok(()) + } + + /// Look up a pair by host id. + pub fn get_by_host(&self, host: &HostPlaylistId) -> Option<&PlaylistPair> { + self.by_host.get(host) + } + + /// Mutable lookup by host id. Used by the executor to record sync + /// instants. + pub fn get_by_host_mut(&mut self, host: &HostPlaylistId) -> Option<&mut PlaylistPair> { + self.by_host.get_mut(host) + } + + /// Look up a pair by device-side id. + pub fn get_by_device(&self, device_side_id: &str) -> Option<&PlaylistPair> { + let host = self.by_device.get(device_side_id)?; + self.by_host.get(host) + } + + /// Number of pairs in the map. + pub fn len(&self) -> usize { + self.by_host.len() + } + + /// True when the map has no pairs. + pub fn is_empty(&self) -> bool { + self.by_host.is_empty() + } + + /// Iterate every pair in stable host-id order. + pub fn iter(&self) -> impl Iterator { + self.by_host.iter() + } + + /// Validate that every pair still satisfies the construction invariants. + /// The map holds the invariants by construction in normal use; this + /// method exists for reload paths where a persisted map might have been + /// hand-edited. + pub fn validate(&self) -> Result<(), PlaylistPairingError> { + if self.by_host.len() != self.by_device.len() { + return Err(PlaylistPairingError::InvariantViolated { + reason: format!( + "host index has {} entries but device index has {}", + self.by_host.len(), + self.by_device.len() + ), + }); + } + for (host, pair) in &self.by_host { + let back_ref = self.by_device.get(pair.device_side_id()).ok_or_else(|| { + PlaylistPairingError::InvariantViolated { + reason: format!( + "host {host} points at device id {:?} but device index has no entry", + pair.device_side_id() + ), + } + })?; + if back_ref != host { + return Err(PlaylistPairingError::InvariantViolated { + reason: format!( + "host {host} points at device id {:?} but device index points at {back_ref}", + pair.device_side_id() + ), + }); + } + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn host(value: &str) -> HostPlaylistId { + HostPlaylistId::new(value).expect("host") + } + + #[test] + fn pair_rejects_empty_device_side_id() { + let error = PlaylistPair::new(host("a"), "", PathBuf::from("Music")).expect_err("empty id"); + assert!(matches!(error, PlaylistPairingError::EmptyDeviceSideId)); + } + + #[test] + fn pair_rejects_device_side_id_with_separator() { + let error = PlaylistPair::new(host("a"), "Music/Playlists/abc", PathBuf::from("Music")) + .expect_err("separator"); + assert!(matches!( + error, + PlaylistPairingError::DeviceSideIdLooksLikePath { .. } + )); + } + + #[test] + fn pair_rejects_absolute_destination() { + let error = + PlaylistPair::new(host("a"), "abc", PathBuf::from("/etc")).expect_err("absolute"); + assert!(matches!(error, PlaylistPairingError::Policy { .. })); + } + + #[test] + fn map_rejects_duplicate_host() { + let mut map = PlaylistMap::new(); + map.insert(PlaylistPair::new(host("a"), "x", PathBuf::from("Music")).expect("pair")) + .expect("insert"); + let error = map + .insert(PlaylistPair::new(host("a"), "y", PathBuf::from("Music")).expect("pair")) + .expect_err("dup"); + assert!(matches!( + error, + PlaylistPairingError::HostAlreadyPaired { .. } + )); + } + + #[test] + fn map_rejects_duplicate_device_id() { + let mut map = PlaylistMap::new(); + map.insert(PlaylistPair::new(host("a"), "x", PathBuf::from("Music")).expect("pair")) + .expect("insert"); + let error = map + .insert(PlaylistPair::new(host("b"), "x", PathBuf::from("Music")).expect("pair")) + .expect_err("dup"); + assert!(matches!( + error, + PlaylistPairingError::DeviceSideIdAlreadyUsed { .. } + )); + } + + #[test] + fn map_lookup_by_device_returns_pair() { + let mut map = PlaylistMap::new(); + let pair = PlaylistPair::new(host("a"), "x", PathBuf::from("Music")).expect("pair"); + map.insert(pair).expect("insert"); + let looked_up = map.get_by_device("x").expect("found"); + assert_eq!(looked_up.host().as_str(), "a"); + } + + #[test] + fn validate_passes_for_consistent_map() { + let mut map = PlaylistMap::new(); + map.insert(PlaylistPair::new(host("a"), "x", PathBuf::from("Music")).expect("pair")) + .expect("insert"); + map.insert(PlaylistPair::new(host("b"), "y", PathBuf::from("Music")).expect("pair")) + .expect("insert"); + assert!(map.validate().is_ok()); + } +} diff --git a/src/device/sync/mod.rs b/src/device/sync/mod.rs new file mode 100644 index 00000000..477b5654 --- /dev/null +++ b/src/device/sync/mod.rs @@ -0,0 +1,135 @@ +//! Device playlist synchronization. +//! +//! GitHub issue #8 / P3.2 require playlist mapping, incremental state, conflict +//! resolution, and an explicitly opted-in auto-sync with safe attach/detach +//! recovery. This module owns that work end-to-end: +//! +//! * [`SyncPolicy`] describes, per host playlist, whether the playlist is +//! opted into device sync, where on the device it should land, and how a +//! conflict between host and device copies is resolved. The default policy +//! is "no sync" — every playlist starts unlinked until the user explicitly +//! opts in. Opt-in is the only path to write a playlist to a device. +//! * [`PlaylistMap`] pairs host playlist ids with the device-side identifier +//! they map to. The pairing is what makes incremental sync possible: +//! without an established pair the planner has no anchor to detect a +//! rename or a delete. +//! * [`SyncState`] records per-track incremental state — last sync time, +//! whether the host copy was edited since the last sync, and whether the +//! device copy has been verified. A re-attached device resumes from the +//! stored state rather than retransferring everything. +//! * [`SyncPlanner`] computes the next transfer plan: for every opted-in +//! host playlist, what tracks are new on the host, what tracks were +//! removed, and what tracks the device currently holds that the host no +//! longer has. The plan is described in terms of [`crate::device::transfer`] +//! stages so the existing executor can run it unchanged. +//! * [`SyncExecutor`] runs the plan with attach/detach safety: if the device +//! disappears mid-run, the executor stops cleanly, rolls back any partial +//! work, and reports which tracks still need to be transferred so a later +//! re-attach can resume from the right place. +//! +//! The module never speaks to the host filesystem on its own. The caller +//! hands it a [`crate::local::root_authority::MountedRootAuthority`] for the +//! device root; every read goes through that authority and every write +//! through [`crate::local::write_authority::MountedWriteAuthority`]. A +//! pre-attached plan that outlives the device is unusable — the executor +//! detects that and refuses to run it. + +mod executor; +mod mapping; +mod planner; +mod policy; +mod recovery; +mod state; + +#[allow(unused_imports)] +pub use executor::{SyncExecutor, SyncRunSummary, SyncStage}; +#[allow(unused_imports)] +pub use mapping::{PlaylistMap, PlaylistPair, PlaylistPairingError}; +#[allow(unused_imports)] +pub use planner::{SyncDelta, SyncDeltaKind, SyncPlan, SyncPlanner, SyncRequest}; +#[allow(unused_imports)] +pub use policy::{ConflictResolution as SyncConflictResolution, SyncPolicy}; +#[allow(unused_imports)] +pub use recovery::{AttachDetachEvent, AttachDetachRecovery, RecoveryError, SyncSessionGuard}; +#[allow(unused_imports)] +pub use state::{IncrementalSyncState, TrackSyncStatus}; + +/// Stable identifier for one host-side playlist involved in sync. +/// +/// The id is opaque to the sync module; the host layer translates it to +/// whatever persistence layer it owns. The id is wrapped so callers cannot +/// accidentally use it as a device identifier. +#[derive( + Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, serde::Serialize, serde::Deserialize, +)] +pub struct HostPlaylistId(pub String); + +impl std::fmt::Display for HostPlaylistId { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("host:playlist:")?; + formatter.write_str(&self.0) + } +} + +impl HostPlaylistId { + /// Construct from a caller-supplied string, rejecting empty input. + pub fn new(value: impl Into) -> Result { + let inner = value.into(); + if inner.trim().is_empty() { + return Err(PolicyError::InvalidPlaylistId { + value: inner, + reason: "playlist id is empty", + }); + } + Ok(Self(inner)) + } + + /// Borrow the underlying string. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// Why the sync module rejected an otherwise well-formed call. +#[derive(Debug, thiserror::Error)] +pub enum PolicyError { + /// A playlist id was empty or contained a forbidden character. + #[error("playlist id {value:?} is invalid: {reason}")] + InvalidPlaylistId { value: String, reason: &'static str }, + /// A relative path was absolute, empty, or contained a non-normal + /// component. + #[error("sync path {path:?} is invalid: {reason}")] + InvalidPath { + path: std::path::PathBuf, + reason: &'static str, + }, + /// A sync policy referenced a destination outside the device root. + #[error("policy destination {path:?} escapes the device root")] + DestinationEscapesRoot { path: std::path::PathBuf }, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn host_playlist_id_rejects_empty_string() { + let error = HostPlaylistId::new(" ").expect_err("empty id"); + assert!(matches!(error, PolicyError::InvalidPlaylistId { .. })); + } + + #[test] + fn host_playlist_id_round_trips_display() { + let id = HostPlaylistId::new("playlist-42").expect("id"); + assert_eq!(id.to_string(), "host:playlist:playlist-42"); + } + + #[test] + fn sync_policy_default_is_disabled() { + // Default policy must NOT auto-sync; opt-in is the only safe + // shape. + let policy = SyncPolicy::default(); + assert!(!policy.is_enabled()); + assert_eq!(policy.conflict_strategy(), SyncConflictResolution::HostWins); + } +} diff --git a/src/device/sync/planner.rs b/src/device/sync/planner.rs new file mode 100644 index 00000000..5f77a4a5 --- /dev/null +++ b/src/device/sync/planner.rs @@ -0,0 +1,476 @@ +//! Sync planner: compute deltas and build transfer plans. +//! +//! The planner is the read-only half of sync. It walks every host playlist +//! the caller asks it to sync, compares the host track set against the +//! recorded [`IncrementalSyncState`](super::state::IncrementalSyncState), +//! and emits a [`SyncPlan`] describing the writes (and optional deletes) +//! the executor must run. +//! +//! The planner never touches the host filesystem or the device filesystem. +//! It accepts the host track list and the host track fingerprints as +//! inputs. This keeps the planner independent of the host library's +//! concrete [`MediaBackend`](crate::local::backend::MediaBackend) and +//! ensures the test surface is plain data. +//! +//! ## Conflict semantics +//! +//! The planner does not decide a conflict outcome; the executor does. The +//! planner records *what changed on each side*, and the executor uses the +//! policy's [`ConflictResolution`](super::SyncConflictResolution) to pick +//! the outcome. This separation keeps the planner testable without an +//! authority and the executor focused on the side-effect ordering. + +use std::collections::BTreeSet; +use std::path::PathBuf; + +use serde::{Deserialize, Serialize}; + +use super::mapping::PlaylistPair; +use super::policy::validate_relative_path; +use super::state::{IncrementalSyncState, TrackSyncStatus}; +use super::{HostPlaylistId, PolicyError}; + +/// A description of one host track the planner needs the fingerprint for. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct HostTrackEntry { + /// Stable host-side track id. + pub track_id: String, + /// Current fingerprint. The planner compares this against the recorded + /// fingerprint to decide whether the track is unchanged, modified, or + /// brand-new. + pub fingerprint: String, + /// Where the track should land on the device, relative to the + /// playlist pair's destination root. + pub device_relative_path: PathBuf, +} + +/// What the planner learned about one track for one playlist pair. +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)] +pub enum SyncDeltaKind { + /// The host track has never been written to the device. + New, + /// The host track has been edited since the last sync. + Modified, + /// The host track was previously synced and is unchanged. + Unchanged, + /// The track was on the device but the host playlist no longer has it. + Removed { + /// Where the executor should remove the file from if the policy + /// allows it. Relative to the device root. + device_relative_path: PathBuf, + /// Last fingerprint the executor wrote. + last_known_fingerprint: String, + }, +} + +/// One per-track outcome from the planner. +#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)] +pub struct SyncDelta { + pub host: HostPlaylistId, + pub track_id: String, + pub kind: SyncDeltaKind, +} + +/// What the planner was told up front. +#[derive(Clone, Debug)] +pub struct SyncRequest { + /// Pairs the caller wants to sync. + pub pairs: Vec, + /// The host track set, keyed by host playlist id. + pub tracks_by_pair: Vec, + /// The recorded incremental state, keyed by host playlist id. The + /// planner consults this to decide what changed. + pub state_by_pair: Vec, +} + +/// One playlist's worth of host tracks plus the recorded state for that +/// pair. +#[derive(Clone, Debug)] +pub struct HostTrackSet { + pub host: HostPlaylistId, + pub tracks: Vec, + pub state: IncrementalSyncState, +} + +/// The planner's output: every per-track delta grouped by playlist pair. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct SyncPlan { + pub deltas: Vec, + /// Total bytes the executor will write. Used for capacity budgeting. + pub expected_write_bytes: u64, + /// Number of files the executor will write. + pub write_count: u32, + /// Number of files the executor will remove (subject to policy). + pub remove_count: u32, +} + +/// The sync planner. Stateless and `Clone` so the same planner can serve +/// multiple requests in sequence. +#[derive(Clone, Debug, Default)] +pub struct SyncPlanner; + +impl SyncPlanner { + /// Construct a new planner. + pub fn new() -> Self { + Self + } + + /// Build a plan from a request. + /// + /// The planner iterates the request's pairs, looks up the matching host + /// track set and recorded state, and produces one [`SyncDelta`] per + /// track. Pairs without a matching host track set or recorded state + /// are reported as a removed delta for every previously-synced track. + #[allow(clippy::unused_self)] + pub fn plan(&self, request: &SyncRequest) -> Result { + if request.pairs.len() != request.tracks_by_pair.len() + || request.pairs.len() != request.state_by_pair.len() + { + return Err(PlannerError::RequestShapeMismatch { + pairs: request.pairs.len(), + tracks: request.tracks_by_pair.len(), + states: request.state_by_pair.len(), + }); + } + + let mut deltas: Vec = Vec::new(); + let mut write_count: u32 = 0; + let mut remove_count: u32 = 0; + + for (index, pair) in request.pairs.iter().enumerate() { + let tracks = &request.tracks_by_pair[index]; + if &tracks.host != pair.host() { + return Err(PlannerError::HostMismatch { + pair: pair.host().clone(), + tracks: tracks.host.clone(), + }); + } + if tracks.state != request.state_by_pair[index] { + return Err(PlannerError::StateMismatch { + host: tracks.host.clone(), + }); + } + validate_relative_path(pair.destination_root(), "destination") + .map_err(|source| PlannerError::Policy { source })?; + + // Build a set of host track ids the planner sees on the host. + let host_track_ids: BTreeSet<&str> = tracks + .tracks + .iter() + .map(|entry| entry.track_id.as_str()) + .collect(); + + // Walk the host tracks. + for entry in &tracks.tracks { + if entry.fingerprint.is_empty() { + return Err(PlannerError::EmptyFingerprint { + host: tracks.host.clone(), + track_id: entry.track_id.clone(), + }); + } + let status = tracks.state.status(&entry.track_id); + let destination_relative = + pair.destination_root().join(&entry.device_relative_path); + validate_relative_path(&destination_relative, "destination") + .map_err(|source| PlannerError::Policy { source })?; + + match status { + None => { + // The state has no entry for this track. It is brand + // new from the planner's perspective. + deltas.push(SyncDelta { + host: tracks.host.clone(), + track_id: entry.track_id.clone(), + kind: SyncDeltaKind::New, + }); + write_count = write_count.saturating_add(1); + } + Some(TrackSyncStatus::Pending) => { + deltas.push(SyncDelta { + host: tracks.host.clone(), + track_id: entry.track_id.clone(), + kind: SyncDeltaKind::New, + }); + write_count = write_count.saturating_add(1); + } + Some(TrackSyncStatus::Synced { fingerprint, .. }) => { + if fingerprint == &entry.fingerprint { + deltas.push(SyncDelta { + host: tracks.host.clone(), + track_id: entry.track_id.clone(), + kind: SyncDeltaKind::Unchanged, + }); + } else { + deltas.push(SyncDelta { + host: tracks.host.clone(), + track_id: entry.track_id.clone(), + kind: SyncDeltaKind::Modified, + }); + write_count = write_count.saturating_add(1); + } + } + Some(TrackSyncStatus::Modified { fingerprint: _ }) => { + // The recorded modification matches (or is older + // than) what the host is currently serving; treat + // as modified so the executor still tries to push. + deltas.push(SyncDelta { + host: tracks.host.clone(), + track_id: entry.track_id.clone(), + kind: SyncDeltaKind::Modified, + }); + write_count = write_count.saturating_add(1); + } + Some(TrackSyncStatus::Missing { .. }) => { + // The track was previously recorded as missing. If + // the host has re-added it, treat as new. + deltas.push(SyncDelta { + host: tracks.host.clone(), + track_id: entry.track_id.clone(), + kind: SyncDeltaKind::New, + }); + write_count = write_count.saturating_add(1); + } + } + } + + // Walk tracks that were previously synced but no longer appear + // on the host. + for (track_id, status) in tracks.state.iter() { + if host_track_ids.contains(track_id) { + continue; + } + match status { + TrackSyncStatus::Synced { + device_relative_path, + last_synced_at, + .. + } => { + deltas.push(SyncDelta { + host: tracks.host.clone(), + track_id: track_id.to_string(), + kind: SyncDeltaKind::Removed { + device_relative_path: PathBuf::from(device_relative_path), + last_known_fingerprint: status + .fingerprint() + .unwrap_or_default() + .to_string(), + }, + }); + let _ = last_synced_at; + remove_count = remove_count.saturating_add(1); + } + TrackSyncStatus::Missing { .. } => { + // Already missing; nothing to do. + } + TrackSyncStatus::Modified { .. } | TrackSyncStatus::Pending => { + // Never written; nothing on the device to remove. + } + } + } + } + + Ok(SyncPlan { + deltas, + expected_write_bytes: 0, + write_count, + remove_count, + }) + } +} + +/// Why the planner rejected a request. +#[derive(Debug, thiserror::Error)] +pub enum PlannerError { + /// The request had a different number of pairs, host track sets, and + /// recorded states. + #[error("sync request is malformed: {pairs} pairs, {tracks} track sets, {states} states")] + RequestShapeMismatch { + pairs: usize, + tracks: usize, + states: usize, + }, + /// The host id of a track set did not match its pair. + #[error( + "host playlist {tracks} is paired with {pair} but the host track set was for {tracks}" + )] + HostMismatch { + pair: HostPlaylistId, + tracks: HostPlaylistId, + }, + /// The provided host track set state did not match the recorded state. + #[error("recorded state does not match the supplied state for host {host}")] + StateMismatch { host: HostPlaylistId }, + /// A track had an empty fingerprint. + #[error("host {host} track {track_id} has an empty fingerprint")] + EmptyFingerprint { + host: HostPlaylistId, + track_id: String, + }, + /// The destination root failed policy validation. + #[error("planner destination is invalid: {source}")] + Policy { + #[source] + source: PolicyError, + }, +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::device::sync::mapping::PlaylistPair; + + fn host(value: &str) -> HostPlaylistId { + HostPlaylistId::new(value).expect("host") + } + + fn pair(host_id: &str, dest: &str) -> PlaylistPair { + PlaylistPair::new( + host(host_id), + format!("device-{host_id}"), + PathBuf::from(dest), + ) + .expect("pair") + } + + #[test] + fn plan_marks_new_track_as_new() { + let mut state = IncrementalSyncState::new(); + let plan = SyncPlanner::new() + .plan(&SyncRequest { + pairs: vec![pair("a", "Music")], + tracks_by_pair: vec![HostTrackSet { + host: host("a"), + tracks: vec![HostTrackEntry { + track_id: "track-1".into(), + fingerprint: "fp1".into(), + device_relative_path: PathBuf::from("a.flac"), + }], + state: IncrementalSyncState::new(), + }], + state_by_pair: vec![std::mem::take(&mut state)], + }) + .expect("plan"); + assert_eq!(plan.write_count, 1); + assert_eq!(plan.remove_count, 0); + assert_eq!(plan.deltas.len(), 1); + assert!(matches!(plan.deltas[0].kind, SyncDeltaKind::New)); + } + + #[test] + fn plan_marks_unchanged_track_as_unchanged() { + let mut state = IncrementalSyncState::new(); + state.record_synced("track-1", "fp1", "Music/a.flac", 100); + let stored = state.clone(); + let plan = SyncPlanner::new() + .plan(&SyncRequest { + pairs: vec![pair("a", "Music")], + tracks_by_pair: vec![HostTrackSet { + host: host("a"), + tracks: vec![HostTrackEntry { + track_id: "track-1".into(), + fingerprint: "fp1".into(), + device_relative_path: PathBuf::from("a.flac"), + }], + state: stored.clone(), + }], + state_by_pair: vec![stored], + }) + .expect("plan"); + assert_eq!(plan.write_count, 0); + assert_eq!(plan.remove_count, 0); + assert!(matches!(plan.deltas[0].kind, SyncDeltaKind::Unchanged)); + } + + #[test] + fn plan_marks_modified_track_as_modified() { + let mut state = IncrementalSyncState::new(); + state.record_synced("track-1", "fp1", "Music/a.flac", 100); + let stored = state.clone(); + let plan = SyncPlanner::new() + .plan(&SyncRequest { + pairs: vec![pair("a", "Music")], + tracks_by_pair: vec![HostTrackSet { + host: host("a"), + tracks: vec![HostTrackEntry { + track_id: "track-1".into(), + fingerprint: "fp2".into(), + device_relative_path: PathBuf::from("a.flac"), + }], + state: stored.clone(), + }], + state_by_pair: vec![stored], + }) + .expect("plan"); + assert_eq!(plan.write_count, 1); + assert!(matches!(plan.deltas[0].kind, SyncDeltaKind::Modified)); + } + + #[test] + fn plan_marks_dropped_track_as_removed() { + let mut state = IncrementalSyncState::new(); + state.record_synced("track-1", "fp1", "Music/a.flac", 100); + let stored = state.clone(); + let plan = SyncPlanner::new() + .plan(&SyncRequest { + pairs: vec![pair("a", "Music")], + tracks_by_pair: vec![HostTrackSet { + host: host("a"), + tracks: vec![], + state: stored.clone(), + }], + state_by_pair: vec![stored], + }) + .expect("plan"); + assert_eq!(plan.write_count, 0); + assert_eq!(plan.remove_count, 1); + assert!(matches!(plan.deltas[0].kind, SyncDeltaKind::Removed { .. })); + } + + #[test] + fn plan_rejects_shape_mismatch() { + let error = SyncPlanner + .plan(&SyncRequest { + pairs: vec![pair("a", "Music")], + tracks_by_pair: vec![], + state_by_pair: vec![IncrementalSyncState::new()], + }) + .expect_err("shape"); + assert!(matches!(error, PlannerError::RequestShapeMismatch { .. })); + } + + #[test] + fn plan_rejects_empty_fingerprint() { + let error = SyncPlanner + .plan(&SyncRequest { + pairs: vec![pair("a", "Music")], + tracks_by_pair: vec![HostTrackSet { + host: host("a"), + tracks: vec![HostTrackEntry { + track_id: "track-1".into(), + fingerprint: String::new(), + device_relative_path: PathBuf::from("a.flac"), + }], + state: IncrementalSyncState::new(), + }], + state_by_pair: vec![IncrementalSyncState::new()], + }) + .expect_err("fingerprint"); + assert!(matches!(error, PlannerError::EmptyFingerprint { .. })); + } + + #[test] + fn plan_rejects_host_mismatch() { + let error = SyncPlanner + .plan(&SyncRequest { + pairs: vec![pair("a", "Music")], + tracks_by_pair: vec![HostTrackSet { + host: host("b"), + tracks: vec![], + state: IncrementalSyncState::new(), + }], + state_by_pair: vec![IncrementalSyncState::new()], + }) + .expect_err("host mismatch"); + assert!(matches!(error, PlannerError::HostMismatch { .. })); + } +} diff --git a/src/device/sync/policy.rs b/src/device/sync/policy.rs new file mode 100644 index 00000000..dc552840 --- /dev/null +++ b/src/device/sync/policy.rs @@ -0,0 +1,219 @@ +//! Per-playlist sync policy. +//! +//! Auto-sync is **opt-in**: every policy starts disabled. The user (or the +//! sync UX) must explicitly call [`SyncPolicy::enable`] with a destination +//! inside the device root. The destination is the relative directory the +//! playlist files will be staged under; the executor pins it to the device +//! authority so a policy that points outside the root is rejected at the +//! boundary rather than producing an out-of-bounds write at runtime. +//! +//! Conflict resolution covers the three cases the existing transfer planner +//! does not own: +//! * Host edited since last sync, device unchanged -> host-wins is the safe +//! default; the device copy is overwritten. +//! * Device edited since last sync, host unchanged -> device-wins preserves +//! the user-managed device copy and the host playlist is marked stale. +//! * Both edited -> manual-or-skip forces the user to resolve, never +//! silently clobbering work on either side. +//! +//! The policy is plain data; the planner reads it. The executor never +//! inspects a policy directly. + +use std::path::{Component, Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +use super::PolicyError; + +/// How a sync run resolves a destination that already exists with a +/// different content hash than the host copy. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub enum ConflictResolution { + /// Host copy wins. The device copy is overwritten. + #[default] + HostWins, + /// Device copy wins. The host copy is preserved but the device's + /// newer content is left in place and the host state is recorded as + /// stale. + DeviceWins, + /// Skip the file. No write, no read, no error — surfaced as a + /// skipped entry in the run summary. + Skip, + /// Refuse to sync the file. The run is allowed to complete the rest + /// of the playlist but this entry is reported as a conflict. + Fail, +} + +/// A user's chosen sync behaviour for one host playlist. +/// +/// Policies are plain data: the planner reads them, the executor never +/// does. New fields can be added without touching the executor. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct SyncPolicy { + enabled: bool, + /// Destination relative to the device root. Empty when `enabled` is + /// `false`. Validated as a relative path on construction. + destination: PathBuf, + conflict_strategy: ConflictResolution, + /// Whether the sync run should delete tracks on the device that are + /// no longer present on the host. Off by default: a missing host + /// track never silently removes a device track. + delete_missing: bool, +} + +impl Default for SyncPolicy { + fn default() -> Self { + Self { + enabled: false, + destination: PathBuf::new(), + conflict_strategy: ConflictResolution::default(), + delete_missing: false, + } + } +} + +impl SyncPolicy { + /// Construct an enabled policy with the given destination and conflict + /// strategy. The destination is validated as a relative path inside the + /// device root. + pub fn enable( + destination: PathBuf, + conflict_strategy: ConflictResolution, + ) -> Result { + validate_relative_path(&destination, "destination")?; + Ok(Self { + enabled: true, + destination, + conflict_strategy, + delete_missing: false, + }) + } + + /// Construct a disabled policy — the safe default. + pub fn disabled() -> Self { + Self::default() + } + + /// True when the policy allows auto-sync. + pub fn is_enabled(&self) -> bool { + self.enabled + } + + /// Destination the policy targets, relative to the device root. + pub fn destination(&self) -> &Path { + &self.destination + } + + /// Conflict resolution the policy picked. + pub fn conflict_strategy(&self) -> ConflictResolution { + self.conflict_strategy + } + + /// True when the executor should remove device tracks whose host + /// counterpart is gone. + pub fn deletes_missing(&self) -> bool { + self.delete_missing + } + + /// Allow the executor to remove device tracks whose host counterpart + /// is gone. Off by default; callers must opt in explicitly. + pub fn set_delete_missing(&mut self, value: bool) { + self.delete_missing = value; + } +} + +pub fn validate_relative_path(path: &Path, field: &'static str) -> Result<(), PolicyError> { + if path.as_os_str().is_empty() { + return Err(PolicyError::InvalidPath { + path: path.to_path_buf(), + reason: match field { + "destination" => "destination is empty", + _ => "path is empty", + }, + }); + } + if path.is_absolute() { + return Err(PolicyError::InvalidPath { + path: path.to_path_buf(), + reason: match field { + "destination" => "destination is absolute", + _ => "path is absolute", + }, + }); + } + for component in path.components() { + if !matches!(component, Component::Normal(_)) { + let reason = match field { + "destination" => "destination contains a non-normal component", + _ => "path contains a non-normal component", + }; + return Err(PolicyError::InvalidPath { + path: path.to_path_buf(), + reason, + }); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_policy_is_disabled_and_skips_delete() { + let policy = SyncPolicy::default(); + assert!(!policy.is_enabled()); + assert!(!policy.deletes_missing()); + assert_eq!(policy.conflict_strategy(), ConflictResolution::HostWins); + } + + #[test] + fn enable_rejects_empty_destination() { + let error = SyncPolicy::enable(PathBuf::new(), ConflictResolution::HostWins) + .expect_err("empty destination"); + assert!(matches!(error, PolicyError::InvalidPath { .. })); + } + + #[test] + fn enable_rejects_absolute_destination() { + let error = SyncPolicy::enable(PathBuf::from("/etc/music"), ConflictResolution::HostWins) + .expect_err("absolute destination"); + assert!(matches!(error, PolicyError::InvalidPath { .. })); + } + + #[test] + fn enable_rejects_dotdot_components() { + let error = SyncPolicy::enable( + PathBuf::from("Music/../escape"), + ConflictResolution::HostWins, + ) + .expect_err("dotdot"); + assert!(matches!(error, PolicyError::InvalidPath { .. })); + } + + #[test] + fn enable_accepts_relative_destination() { + let policy = SyncPolicy::enable( + PathBuf::from("Music/Playlists"), + ConflictResolution::DeviceWins, + ) + .expect("policy"); + assert!(policy.is_enabled()); + assert_eq!(policy.destination(), Path::new("Music/Playlists")); + assert_eq!(policy.conflict_strategy(), ConflictResolution::DeviceWins); + } + + #[test] + fn set_delete_missing_updates_flag() { + let mut policy = SyncPolicy::default(); + assert!(!policy.deletes_missing()); + policy.set_delete_missing(true); + assert!(policy.deletes_missing()); + } + + #[test] + fn conflict_resolution_default_is_host_wins() { + assert_eq!(ConflictResolution::default(), ConflictResolution::HostWins); + } +} diff --git a/src/device/sync/recovery.rs b/src/device/sync/recovery.rs new file mode 100644 index 00000000..58af0eb8 --- /dev/null +++ b/src/device/sync/recovery.rs @@ -0,0 +1,280 @@ +//! Attach/detach detection and safe-recovery primitives. +//! +//! A sync run is meaningful only while the device is attached. The +//! recovery module is the place where "device disappeared" is observed +//! and where a half-finished run can be safely resumed. +//! +//! ## Why attach/detach is its own concern +//! +//! The transfer executor already detects a missing device at every +//! authority revalidation. What it does not do is *decide what to do +//! next*: roll back the partial writes, decide whether the remaining +//! work is salvageable, or hand the caller a token to resume later. That +//! is this module's job. +//! +//! ## The session guard +//! +//! [`SyncSessionGuard`] is the runtime handle the executor checks between +//! stages. It is intentionally narrow: it answers one question — "is the +//! device still attached?" — and exposes a single observable event +//! channel so the executor can record what happened. +//! +//! ## Recovery +//! +//! [`AttachDetachRecovery`] is the caller-side API. It consumes the events +//! the executor emitted during a run and reports a verdict: +//! +//! * `Completed` — every planned stage ran to completion. +//! * `Detached` — the device disappeared at some specific stage; the +//! recovery records which stage ran last and which was the next one +//! the executor would have run. The caller can resume by submitting a +//! new plan that skips the stages the previous run completed. +//! * `Failed` — the run aborted for a reason other than a detach. +//! +//! The recovery never mutates the recorded state. The executor updates +//! the recorded state only after a successful stage; a detach leaves the +//! state consistent with what the device actually has. + +use std::collections::VecDeque; +use std::sync::{Arc, Mutex}; + +use super::executor::SyncStage; + +/// What the executor saw during a single sync run. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum AttachDetachEvent { + /// The device was attached when the run started. + AttachedAtStart, + /// The executor finished one stage successfully. + StageCompleted { stage: SyncStage }, + /// The executor detected that the device had disappeared. The stage + /// named here is the next stage the executor would have attempted; + /// it never ran. + Detached { next_stage: SyncStage }, + /// The executor observed a non-detach failure on the named stage. + Failed { stage: SyncStage, reason: String }, +} + +/// Verdict the recovery produces after reading the event stream. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum RecoveryVerdict { + /// Every stage ran to completion. + Completed, + /// The device detached at `next_stage`. The caller can resume by + /// submitting a new plan that skips the stages the previous run + /// already finished. + Detached { + last_completed: Option, + next_stage: SyncStage, + }, + /// The run failed for a non-detach reason on `stage`. + Failed { stage: SyncStage, reason: String }, +} + +/// The runtime handle the executor checks between stages. +/// +/// The guard is cheap to clone. Cloning shares the same observation +/// state, so the executor's view of the device and the recovery's view +/// are always in agreement. +#[derive(Clone)] +pub struct SyncSessionGuard { + state: Arc>, +} + +struct GuardState { + attached: bool, + events: VecDeque, +} + +impl SyncSessionGuard { + /// Construct a guard in the "attached" state. The caller is + /// responsible for calling [`SyncSessionGuard::mark_detached`] when + /// the device is observed to be gone. + pub fn attached() -> Self { + Self { + state: Arc::new(Mutex::new(GuardState { + attached: true, + events: VecDeque::new(), + })), + } + } + + /// Mark the device as detached. Subsequent calls to + /// [`SyncSessionGuard::is_attached`] return `false`. This call is + /// idempotent: a second call while the device is already detached + /// is a no-op. + pub fn mark_detached(&self) { + let mut state = self.state.lock().expect("guard poisoned"); + if state.attached { + state.attached = false; + } + } + + /// True while the device is still attached. + pub fn is_attached(&self) -> bool { + self.state.lock().expect("guard poisoned").attached + } + + /// Record that the executor completed one stage. The executor calls + /// this after every successful stage; the recovery reads the events + /// to compute its verdict. + pub fn record_stage_completed(&self, stage: SyncStage) { + let mut state = self.state.lock().expect("guard poisoned"); + state + .events + .push_back(AttachDetachEvent::StageCompleted { stage }); + } + + /// Record a detach event. The executor calls this when it sees the + /// device is gone and is about to abort the run. + pub fn record_detach(&self, next_stage: SyncStage) { + let mut state = self.state.lock().expect("guard poisoned"); + state + .events + .push_back(AttachDetachEvent::Detached { next_stage }); + } + + /// Record a non-detach failure on the named stage. + pub fn record_failure(&self, stage: SyncStage, reason: impl Into) { + let mut state = self.state.lock().expect("guard poisoned"); + state.events.push_back(AttachDetachEvent::Failed { + stage, + reason: reason.into(), + }); + } + + /// Borrow the recorded event stream. The stream is owned by the + /// guard and is mutated only by the executor; this method returns a + /// snapshot. + pub fn snapshot_events(&self) -> Vec { + let state = self.state.lock().expect("guard poisoned"); + state.events.iter().cloned().collect() + } +} + +/// The caller-side recovery consumer. +#[derive(Clone, Debug, Default)] +pub struct AttachDetachRecovery { + _private: (), +} + +impl AttachDetachRecovery { + /// Construct a recovery consumer. + pub fn new() -> Self { + Self { _private: () } + } + + /// Read the recorded events and produce a verdict. + /// + /// The verdict is computed by walking the event stream in order. A + /// `Detached` event always wins over an earlier `Completed` event — + /// once the device is gone, every subsequent stage is unattempted. + /// A `Failed` event likewise wins over a `Detached` if it appears + /// later; the executor never emits both, but the recovery tolerates + /// either ordering. + pub fn verdict(guard: &SyncSessionGuard) -> RecoveryVerdict { + let events = guard.snapshot_events(); + let mut last_completed: Option = None; + let mut verdict = RecoveryVerdict::Completed; + for event in events { + match event { + AttachDetachEvent::AttachedAtStart => {} + AttachDetachEvent::StageCompleted { stage } => { + last_completed = Some(stage); + } + AttachDetachEvent::Detached { next_stage } => { + verdict = RecoveryVerdict::Detached { + last_completed, + next_stage, + }; + return verdict; + } + AttachDetachEvent::Failed { stage, reason } => { + return RecoveryVerdict::Failed { stage, reason }; + } + } + } + verdict + } +} + +/// Why the recovery rejected an input. +#[derive(Debug, thiserror::Error)] +pub enum RecoveryError { + /// The guard's event stream was empty; nothing to recover from. + #[error("attach/detach recovery received an empty event stream")] + EmptyEventStream, + /// The guard reported the device as attached while the recovery + /// expected it to be detached. + #[error("recovery called while the device is still attached")] + StillAttached, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn guard_starts_attached() { + let guard = SyncSessionGuard::attached(); + assert!(guard.is_attached()); + } + + #[test] + fn mark_detached_toggles_attached_state() { + let guard = SyncSessionGuard::attached(); + guard.mark_detached(); + assert!(!guard.is_attached()); + guard.mark_detached(); + assert!(!guard.is_attached()); + } + + #[test] + fn verdict_reports_completed_when_nothing_failed() { + let guard = SyncSessionGuard::attached(); + guard.record_stage_completed(SyncStage::OpenSession); + guard.record_stage_completed(SyncStage::BrowseStorage); + let verdict = AttachDetachRecovery::verdict(&guard); + assert!(matches!(verdict, RecoveryVerdict::Completed)); + } + + #[test] + fn verdict_reports_detach_with_last_completed_stage() { + let guard = SyncSessionGuard::attached(); + guard.record_stage_completed(SyncStage::OpenSession); + guard.record_detach(SyncStage::FetchTrack { + track_id: "t".into(), + }); + let verdict = AttachDetachRecovery::verdict(&guard); + match verdict { + RecoveryVerdict::Detached { + last_completed, + next_stage, + } => { + assert_eq!(last_completed, Some(SyncStage::OpenSession)); + assert!(matches!(next_stage, SyncStage::FetchTrack { .. })); + } + other => panic!("unexpected verdict {other:?}"), + } + } + + #[test] + fn verdict_reports_failure() { + let guard = SyncSessionGuard::attached(); + guard.record_stage_completed(SyncStage::OpenSession); + guard.record_failure( + SyncStage::FetchTrack { + track_id: "t".into(), + }, + "io", + ); + let verdict = AttachDetachRecovery::verdict(&guard); + match verdict { + RecoveryVerdict::Failed { stage, reason } => { + assert!(matches!(stage, SyncStage::FetchTrack { .. })); + assert_eq!(reason, "io"); + } + other => panic!("unexpected verdict {other:?}"), + } + } +} diff --git a/src/device/sync/state.rs b/src/device/sync/state.rs new file mode 100644 index 00000000..e6c7c81a --- /dev/null +++ b/src/device/sync/state.rs @@ -0,0 +1,225 @@ +//! Per-track incremental sync state. +//! +//! A re-attached device must not retransmit files it already has. The state +//! records, for every track that has ever been written, the fingerprint the +//! last successful run sent, the device-side path it was sent under, and +//! the wall-clock instant the write completed. +//! +//! The state lives in `IncrementalSyncState`, an append-only collection of +//! [`TrackSyncStatus`] entries indexed by host track id. The planner reads +//! the state to decide whether a track on the host is new, modified, or +//! unchanged since the last sync. The executor updates the state as it +//! commits each file. +//! +//! Invariants: +//! * A track that is recorded as `Synced` always has a non-empty +//! fingerprint and a recorded sync instant. +//! * A track that is recorded as `Modified` has a fingerprint but no +//! sync instant — the planner has decided the device copy is stale. +//! * A track that is recorded as `Pending` has no fingerprint; the +//! executor has not yet committed anything for it. +//! * A track that is recorded as `Missing` was on the device but the host +//! no longer has it. The executor uses this to honour the policy's +//! delete-missing flag. + +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; + +/// What we know about one host track relative to its last sync run. +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub enum TrackSyncStatus { + /// The host track has been written to the device at least once. + /// The fingerprint and path describe the last successful write. + Synced { + fingerprint: String, + device_relative_path: String, + last_synced_at: u64, + }, + /// The host track has been edited since the last sync. The fingerprint + /// is the host's current value; the device copy is stale. + Modified { fingerprint: String }, + /// The host track has not yet been written to the device. + Pending, + /// The host track was once synced but is no longer present in the + /// host playlist. The device copy is a candidate for deletion if the + /// policy allows it. + Missing { + last_known_fingerprint: String, + last_synced_at: u64, + }, +} + +impl TrackSyncStatus { + /// True when the status is `Pending` — the executor should plan a + /// write. + pub fn is_pending(&self) -> bool { + matches!(self, Self::Pending) + } + + /// True when the status is `Missing`. + pub fn is_missing(&self) -> bool { + matches!(self, Self::Missing { .. }) + } + + /// Last known fingerprint, when one is recorded. + pub fn fingerprint(&self) -> Option<&str> { + match self { + Self::Synced { fingerprint, .. } => Some(fingerprint), + Self::Modified { fingerprint } => Some(fingerprint), + Self::Missing { + last_known_fingerprint, + .. + } => Some(last_known_fingerprint), + Self::Pending => None, + } + } +} + +/// A collection of [`TrackSyncStatus`] entries indexed by host track id. +/// +/// The host track id is opaque to the sync module — the caller decides how +/// to construct it. A common choice is a stable per-track fingerprint the +/// library already computes; whatever it is, two calls to +/// [`IncrementalSyncState::record`] with the same id must refer to the +/// same track. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +pub struct IncrementalSyncState { + entries: BTreeMap, +} + +impl IncrementalSyncState { + /// Construct an empty state. + pub fn new() -> Self { + Self::default() + } + + /// Number of tracked tracks. + pub fn len(&self) -> usize { + self.entries.len() + } + + /// True when no tracks are tracked. + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Look up the status of one host track. + pub fn status(&self, track_id: &str) -> Option<&TrackSyncStatus> { + self.entries.get(track_id) + } + + /// Record a successful sync. The status becomes + /// [`TrackSyncStatus::Synced`]. + pub fn record_synced( + &mut self, + track_id: impl Into, + fingerprint: impl Into, + device_relative_path: impl Into, + instant_seconds: u64, + ) { + let fingerprint = fingerprint.into(); + let device_relative_path = device_relative_path.into(); + debug_assert!(!fingerprint.is_empty(), "fingerprint must be non-empty"); + debug_assert!( + !device_relative_path.is_empty(), + "device path must be non-empty" + ); + self.entries.insert( + track_id.into(), + TrackSyncStatus::Synced { + fingerprint, + device_relative_path, + last_synced_at: instant_seconds, + }, + ); + } + + /// Record that the host track has been edited since the last sync. + pub fn record_modified(&mut self, track_id: impl Into, fingerprint: impl Into) { + self.entries.insert( + track_id.into(), + TrackSyncStatus::Modified { + fingerprint: fingerprint.into(), + }, + ); + } + + /// Record that the host track is no longer present. The executor uses + /// this to honour the policy's delete-missing flag. + pub fn record_missing( + &mut self, + track_id: impl Into, + last_known_fingerprint: impl Into, + last_synced_at: u64, + ) { + self.entries.insert( + track_id.into(), + TrackSyncStatus::Missing { + last_known_fingerprint: last_known_fingerprint.into(), + last_synced_at, + }, + ); + } + + /// Drop the entry for a track. Called when the host playlist no longer + /// references the track and the executor has decided not to keep its + /// history. + pub fn forget(&mut self, track_id: &str) { + self.entries.remove(track_id); + } + + /// Iterate every recorded status. + pub fn iter(&self) -> impl Iterator { + self.entries.iter().map(|(k, v)| (k.as_str(), v)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn record_synced_replaces_prior_status() { + let mut state = IncrementalSyncState::new(); + state.record_synced("a", "fp1", "Music/a.flac", 100); + state.record_synced("a", "fp2", "Music/a.flac", 200); + let status = state.status("a").expect("status"); + assert!(matches!(status, TrackSyncStatus::Synced { .. })); + assert_eq!(status.fingerprint(), Some("fp2")); + } + + #[test] + fn record_modified_marks_track_stale() { + let mut state = IncrementalSyncState::new(); + state.record_synced("a", "fp1", "Music/a.flac", 100); + state.record_modified("a", "fp2"); + let status = state.status("a").expect("status"); + assert!(matches!(status, TrackSyncStatus::Modified { .. })); + } + + #[test] + fn record_missing_keeps_last_known_fingerprint() { + let mut state = IncrementalSyncState::new(); + state.record_synced("a", "fp1", "Music/a.flac", 100); + state.record_missing("a", "fp1", 100); + let status = state.status("a").expect("status"); + assert!(status.is_missing()); + assert_eq!(status.fingerprint(), Some("fp1")); + } + + #[test] + fn forget_removes_entry() { + let mut state = IncrementalSyncState::new(); + state.record_synced("a", "fp1", "Music/a.flac", 100); + state.forget("a"); + assert!(state.status("a").is_none()); + } + + #[test] + fn pending_status_has_no_fingerprint() { + let status = TrackSyncStatus::Pending; + assert!(status.fingerprint().is_none()); + assert!(status.is_pending()); + } +}