coven_sync/
file_device_only.rs

1//! Preparing cloud files to be made device-only on this device (§17.3):
2//! download each to the path the app names under the folder the person
3//! picked, reporting how far the downloads have come, for the write that
4//! switches them all to this device. A prepare that fails leaves nothing
5//! behind.
6
7use crate::files::Files;
8use crate::FileReadError;
9use coven_database::{
10    CopyReservation, DbError, DeviceOnlyCopy, FileHasher, FilePlace, FileRef, FolderCopy,
11    PreparedDeviceOnly,
12};
13use coven_foundation::{
14    files::{CopyFolder, CopyPathError, FileError, FlushBatch, Flushed, RelativePath},
15    id_source::FileId,
16    progress::{Progress, ProgressReport},
17};
18use std::path::{Path, PathBuf};
19use std::sync::Arc;
20use tokio::sync::watch;
21
22/// A step of preparing files to be made device-only (§17.3, E8), which a
23/// host draws a bar for: one step for each file downloaded, in order. Done
24/// and total count the file's bytes; done plus `before`, out of `all`, is how
25/// far the whole call has come.
26#[derive(Clone, Copy, Debug, PartialEq, Eq)]
27pub enum DeviceOnlyStep {
28    /// Downloading one file to its copy.
29    Downloading {
30        /// The file.
31        file: FileId,
32        /// The bytes of the files downloaded before it.
33        before: u64,
34        /// The bytes of every file the call downloads.
35        all: u64,
36    },
37}
38
39/// Why no file of a call was made device-only (E8). Whatever the call made
40/// is removed before it returns, or, if removing it failed too, at the next
41/// open.
42#[derive(Debug, thiserror::Error)]
43pub enum DeviceOnlyError {
44    /// The call names one file twice.
45    #[error("file {file} is named twice")]
46    RepeatedFile {
47        /// The file.
48        file: FileId,
49    },
50    /// A copy can't go where the app named it.
51    #[error(transparent)]
52    Path(CopyPathError),
53    /// This device may not make a file device-only, or the write switching
54    /// them failed: no record of the file here (`FileAbsent`), as for a
55    /// file of a circle this member was never in; one not in the cloud
56    /// (`NotInCloud`); an app file (`NamespaceKind`); a file of a circle
57    /// this member has left (`NotInCircle`); or another member's row of an
58    /// `owner_only` table referencing it (`NotOwner`).
59    #[error(transparent)]
60    Database(DbError),
61    /// Downloading a file failed.
62    #[error("downloading file {file} failed: {error}")]
63    Download {
64        /// The file.
65        file: FileId,
66        /// The read's failure.
67        #[source]
68        error: FileReadError,
69    },
70    /// Writing a folder or a copy failed.
71    #[error(transparent)]
72    Disk(FileError),
73    /// The app cancelled the call before its files switched.
74    #[error("cancelled")]
75    Cancelled,
76    /// Removing what the call made failed too. What stays is recorded, and
77    /// the next open removes it.
78    #[error("{error}; removing the copies failed: {cleanup}")]
79    Cleanup {
80        /// The call's failure.
81        error: Box<DeviceOnlyError>,
82        /// The removal's.
83        #[source]
84        cleanup: DbError,
85    },
86    /// The call's work panicked: a bug, which the caller reports (E15).
87    #[error(transparent)]
88    Bug(#[from] coven_foundation::task::Bug),
89}
90
91impl From<CopyPathError> for DeviceOnlyError {
92    fn from(error: CopyPathError) -> Self {
93        match error {
94            CopyPathError::Bug(bug) => Self::Bug(bug),
95            error => Self::Path(error),
96        }
97    }
98}
99
100impl From<DbError> for DeviceOnlyError {
101    fn from(error: DbError) -> Self {
102        match error {
103            DbError::Bug(bug) => Self::Bug(bug),
104            error => Self::Database(error),
105        }
106    }
107}
108
109impl From<FileError> for DeviceOnlyError {
110    fn from(error: FileError) -> Self {
111        match error {
112            FileError::Bug(bug) => Self::Bug(bug),
113            error => Self::Disk(error),
114        }
115    }
116}
117
118impl coven_foundation::task::CarriesBug for DeviceOnlyError {
119    fn bug(&self) -> Option<&coven_foundation::task::Bug> {
120        match self {
121            Self::Bug(bug) => Some(bug),
122            Self::Path(error) => error.bug(),
123            Self::Database(error) => error.bug(),
124            Self::Download { error, .. } => error.bug(),
125            Self::Disk(error) => error.bug(),
126            Self::Cleanup { error, cleanup } => error.bug().or_else(|| cleanup.bug()),
127            Self::RepeatedFile { .. } | Self::Cancelled => None,
128        }
129    }
130}
131
132impl Files {
133    /// Download each cloud file to its path under `folder`, for one write
134    /// to switch them all to this device only, each copy becoming the
135    /// original this device records (§17.3). Each download checks the
136    /// file's content hash, and each copy is written under a temporary name
137    /// in its final folder and renamed once synced. Each name is written as
138    /// the one rule for every platform makes it, which keeps it inside the
139    /// folder. A path through a symbolic link below the folder, where
140    /// something is already, or naming what another path names once
141    /// written, case and Unicode form aside, is refused. `progress` hears
142    /// each file's download and how many of its bytes, and of the call's,
143    /// are down. Until it returns, `cancel` stops the call; a failure, a
144    /// panic in a download among them, or a cancel removes every copy and
145    /// folder the call made.
146    pub async fn prepare_device_only(
147        &self,
148        folder: &Path,
149        files: Vec<(FileId, RelativePath)>,
150        cancel: &watch::Receiver<bool>,
151        progress: watch::Sender<Option<Progress<DeviceOnlyStep>>>,
152    ) -> Result<PreparedDeviceOnly, DeviceOnlyError> {
153        self.inner.check_open()?;
154        let mut ids = Vec::new();
155        let mut paths = Vec::new();
156        for (file, path) in files {
157            if ids.contains(&file) {
158                return Err(DeviceOnlyError::RepeatedFile { file });
159            }
160            ids.push(file);
161            paths.push(path);
162        }
163        // Refused before any byte is downloaded; the write checks again.
164        let refs = self.inner.database.device_only_files(ids).await?;
165        let folder = Arc::new(
166            CopyFolder::open(folder.to_owned(), self.inner.database.bugs().clone()).await?,
167        );
168        let plan = folder.plan(paths).await?;
169        let copies: Vec<FolderCopy> = plan
170            .files
171            .iter()
172            .map(|path| FolderCopy {
173                path: path.clone(),
174                temporary: path
175                    .with_file_name(format!(".coven-{}.download", self.inner.ids.new_id())),
176                folders: plan
177                    .folders
178                    .iter()
179                    .filter(|folder| path.starts_with(folder))
180                    .cloned()
181                    .collect(),
182            })
183            .collect();
184        // Held until the call ends however it ends: dropping the call's
185        // future drops it too, and the next write removes what it made.
186        let reservation = self.inner.database.record_copies(copies.clone()).await?;
187        let mut cancel = cancel.clone();
188        let mut report = ProgressReport::new(progress, self.inner.clock.clone());
189        // A panic in a download fails the call as any failure does, and
190        // what the call made goes before it returns (E15).
191        let downloaded = coven_foundation::task::caught(
192            "download files to make device-only",
193            download(
194                self,
195                &folder,
196                &reservation,
197                &plan.folders,
198                &refs,
199                &copies,
200                &mut cancel,
201                &mut report,
202            ),
203        )
204        .await;
205        let error = match downloaded {
206            Ok(Ok((switches, flushed))) => {
207                return Ok(self.inner.database.prepared_device_only(
208                    reservation,
209                    switches,
210                    flushed,
211                )?)
212            }
213            Ok(Err(error)) => error,
214            Err(bug) => DeviceOnlyError::Bug(bug),
215        };
216        let made = copies.into_iter().map(|copy| copy.path).collect();
217        let abandoned = self.inner.database.abandon_copies(made).await;
218        drop(reservation);
219        match abandoned {
220            Ok(()) => Err(error),
221            Err(cleanup) => Err(DeviceOnlyError::Cleanup { error: Box::new(error), cleanup }),
222        }
223    }
224}
225
226/// Make the plan's folders, then download each file to its copy and
227/// place it, returning the switches the write makes and the flush of
228/// the copies' batch: their folders and the drive's cache, once for the
229/// call, so every copy is on the drive before the write records it
230/// (Appendix C.3). A cancel is heard while waiting for bytes and between
231/// steps, never inside one, so every name a step made is in its record
232/// when the call removes them. A step that makes a name or writes under one,
233/// a folder, a copy's file, an append to it, or its rename to its path, runs
234/// to its end with its record once started, holding the call's copies,
235/// though the call is dropped meanwhile: what it made is then in its record
236/// for the next write or open to remove, and no removal meets a file a step
237/// still holds open, which Windows would refuse. Between steps the call
238/// holds no file open but the copy's writer, which drops with the call.
239#[expect(clippy::too_many_arguments, reason = "the call's parts, each its own")]
240async fn download(
241    handle: &Files,
242    folder: &Arc<CopyFolder>,
243    reservation: &CopyReservation,
244    folders: &[PathBuf],
245    files: &[FileRef],
246    copies: &[FolderCopy],
247    cancel: &mut watch::Receiver<bool>,
248    report: &mut ProgressReport<DeviceOnlyStep>,
249) -> Result<(Vec<DeviceOnlyCopy>, Flushed), DeviceOnlyError> {
250    let bugs = handle.inner.database.bugs().clone();
251    for made in folders {
252        check(cancel)?;
253        let (owner, folder, made) = (handle.inner.clone(), folder.clone(), made.clone());
254        step(reservation, "make a folder for copies", &bugs, async move {
255            // A folder someone made since planning is theirs, and stays.
256            if folder.create_folder(made.clone()).await? {
257                owner.database.made_folder(&made).await?;
258            }
259            Ok(())
260        })
261        .await?;
262    }
263    let mut switches = Vec::new();
264    let mut batch = FlushBatch::new(handle.inner.database.bugs().clone());
265    let all = files.iter().map(FileRef::size).sum();
266    let mut before = 0;
267    for (file, copy) in files.iter().zip(copies) {
268        check(cancel)?;
269        report.begin(DeviceOnlyStep::Downloading { file: file.id(), before, all }, file.size());
270        let mut done = 0;
271        let download = |error| DeviceOnlyError::Download { file: file.id(), error };
272        let FilePlace::Cloud { location, .. } = file.place() else {
273            return Err(DbError::NotInCloud { file: file.id() }.into());
274        };
275        let location = location.clone();
276        let stream = handle.open_file_stream(file).await.map_err(download)?;
277        let mut writer = {
278            let (folder, temporary) = (folder.clone(), copy.temporary.clone());
279            step(reservation, "create a copy", &bugs, async move {
280                Ok(folder.create_copy(temporary).await?)
281            })
282            .await?
283        };
284        // A whole read in order checks the content hash before its last
285        // bytes are handed over (§17.7). The bytes written are hashed as
286        // they go, for a later move of the copy to check against (§17.3).
287        let mut range = stream.read_range(0, file.size()).map_err(download)?;
288        let mut hashes = FileHasher::new();
289        loop {
290            let bytes = tokio::select! {
291                biased;
292                () = cancelled(cancel) => return Err(DeviceOnlyError::Cancelled),
293                bytes = range.next() => bytes.map_err(download)?,
294            };
295            let Some(bytes) = bytes else { break };
296            hashes.update(&bytes);
297            done += bytes.len() as u64;
298            writer = {
299                #[cfg(test)]
300                let owner = handle.inner.clone();
301                step(reservation, "append to a copy", &bugs, async move {
302                    writer.append(&bytes).await?;
303                    #[cfg(test)]
304                    owner.pass_device_only_gate(DeviceOnlyPoint::Appended).await;
305                    Ok(writer)
306                })
307                .await?
308            };
309            report.done(done);
310        }
311        before += file.size();
312        batch = {
313            let (owner, folder, path) = (handle.inner.clone(), folder.clone(), copy.path.clone());
314            step(reservation, "place a copy", &bugs, async move {
315                writer.publish(&path, &mut batch).await?;
316                #[cfg(test)]
317                owner.pass_device_only_gate(DeviceOnlyPoint::Renamed).await;
318                let observed = folder.observe(&path).await.map_err(DbError::from)?;
319                owner.database.place_copy(&observed).await?;
320                Ok(batch)
321            })
322            .await?
323        };
324        switches.push(DeviceOnlyCopy {
325            file: file.id(),
326            location,
327            path: copy.path.clone(),
328            hashes: hashes.finish(),
329        });
330    }
331    check(cancel)?;
332    Ok((switches, batch.flush().await?))
333}
334
335/// Where a test holds a device-only call (`FilesInner::device_only_gate`).
336#[cfg(test)]
337#[derive(Clone, Copy, Debug, PartialEq, Eq)]
338pub(crate) enum DeviceOnlyPoint {
339    /// A copy's bytes were just appended.
340    Appended,
341    /// A copy was renamed to its path and is not yet recorded placed.
342    Renamed,
343}
344
345/// A test's hold on a device-only call: where, the signal that it got
346/// there, and the resume it waits for.
347#[cfg(test)]
348pub(crate) type DeviceOnlyGate =
349    (DeviceOnlyPoint, tokio::sync::oneshot::Sender<()>, tokio::sync::oneshot::Receiver<()>);
350
351impl crate::files::FilesInner {
352    /// Wait at the gate a test set for `point`, if any, once there.
353    #[cfg(test)]
354    async fn pass_device_only_gate(&self, point: DeviceOnlyPoint) {
355        let gate = {
356            let mut gate = coven_foundation::locks::lock_unpoisoned(&self.device_only_gate);
357            if gate.as_ref().is_some_and(|(at, _, _)| *at == point) {
358                gate.take()
359            } else {
360                None
361            }
362        };
363        if let Some((_, reached, resume)) = gate {
364            reached.send(()).ok();
365            resume.await.ok();
366        }
367    }
368}
369
370/// Run `work`, a step of the call named `operation`, to its end with its
371/// record once started, holding the call's copies (`run_step`).
372async fn step<T: Send + 'static>(
373    reservation: &CopyReservation,
374    operation: &'static str,
375    bugs: &coven_foundation::task::BugSinkRef,
376    work: impl std::future::Future<Output = Result<T, DeviceOnlyError>> + Send + 'static,
377) -> Result<T, DeviceOnlyError> {
378    reservation.run_step(operation, bugs.clone(), work).await.map_err(FileError::from)?
379}
380
381fn check(cancel: &watch::Receiver<bool>) -> Result<(), DeviceOnlyError> {
382    if *cancel.borrow() {
383        Err(DeviceOnlyError::Cancelled)
384    } else {
385        Ok(())
386    }
387}
388
389/// Resolves once `cancel` says to stop; never, once its sender is gone.
390async fn cancelled(cancel: &mut watch::Receiver<bool>) {
391    loop {
392        if *cancel.borrow_and_update() {
393            return;
394        }
395        if cancel.changed().await.is_err() {
396            std::future::pending::<()>().await;
397        }
398    }
399}
coven_sync/file_device_only.rs

coven_sync/
file_device_only.rs
