From 490acaa3e73830840ed6d39f1c30e5bcd808bfd0 Mon Sep 17 00:00:00 2001 From: fastace Date: Thu, 6 Aug 2026 16:31:33 +0800 Subject: [PATCH 1/2] feat(backend): add opendal ProgressLayer with conditional chunked writer --- .gitignore | 2 + Cargo.lock | 7 + crates/backend/Cargo.toml | 1 + crates/backend/src/lib.rs | 4 + crates/backend/src/opendal.rs | 164 +++++++++++++++------ crates/backend/src/progress_layer.rs | 115 +++++++++++++++ crates/backend/tests/progress_layer.rs | 55 +++++++ crates/backend/tests/progress_layer_cos.rs | 112 ++++++++++++++ 8 files changed, 411 insertions(+), 49 deletions(-) create mode 100644 crates/backend/src/progress_layer.rs create mode 100644 crates/backend/tests/progress_layer.rs create mode 100644 crates/backend/tests/progress_layer_cos.rs diff --git a/.gitignore b/.gitignore index 13ddefec3..e44cede27 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,8 @@ mutants.out cargo-test* coverage/*lcov .testscompletions-* +.env +.idea/ # Generated by Cargo # Use in library crates diff --git a/Cargo.lock b/Cargo.lock index 60ad3f0b2..ad9a58a74 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1180,6 +1180,12 @@ dependencies = [ "const-random", ] +[[package]] +name = "dotenvy" +version = "0.15.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" + [[package]] name = "downcast" version = "0.11.0" @@ -4240,6 +4246,7 @@ dependencies = [ "conflate", "derive_setters", "displaydoc", + "dotenvy", "hex", "jiff", "log", diff --git a/crates/backend/Cargo.toml b/crates/backend/Cargo.toml index f290a199f..41fc75203 100644 --- a/crates/backend/Cargo.toml +++ b/crates/backend/Cargo.toml @@ -101,6 +101,7 @@ opendal = { version = "0.57.0", features = ["blocking", "services-b2", "services anyhow = { workspace = true } rstest = { workspace = true } toml = "1.0.3" +dotenvy = "0.15.7" [lints] workspace = true diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 7ce3da126..7fc1d0438 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -65,6 +65,10 @@ pub mod util; #[cfg(feature = "opendal")] pub mod opendal; +/// Progress counting layer for the `OpenDAL` backend. +#[cfg(feature = "opendal")] +mod progress_layer; + /// `Rclone` backend for Rustic. #[cfg(feature = "rclone")] pub mod rclone; diff --git a/crates/backend/src/opendal.rs b/crates/backend/src/opendal.rs index bb8b14f1d..535125a0c 100644 --- a/crates/backend/src/opendal.rs +++ b/crates/backend/src/opendal.rs @@ -14,7 +14,7 @@ use opendal::{ Entry, Metadata, blocking::{Operator, StdReader}, layers::{ConcurrentLimitLayer, LoggingLayer, RetryLayer, ThrottleLayer}, - options::{ListOptions, ReadOptions}, + options::{ListOptions, ReadOptions, WriteOptions}, }; use rayon::prelude::{IntoParallelIterator, ParallelIterator}; use tokio::runtime::Runtime; @@ -26,15 +26,27 @@ use rustic_core::{ repofile::{Node, NodeType}, }; +use crate::progress_layer::{ProgressLayer, WrittenCounter}; + mod constants { /// Default number of retries pub(super) const DEFAULT_RETRY: usize = 5; + + /// Chunk size used for streaming (multipart) writes. + /// + /// 8 MiB is above the 5 MiB minimum part size required by S3-style + /// backends (including Tencent COS). Feeding data in chunks of this + /// size bypasses opendal's single-shot `write_once` optimization and + /// forces a real multipart upload, so progress is reported per part + /// instead of in one jump at the end. + pub(super) const CHUNK_SIZE: usize = 8 * 1024 * 1024; } /// `OpenDALBackend` contains a wrapper around an blocking operator of the `OpenDAL` library. #[derive(Clone, Debug)] pub struct OpenDALBackend { operator: Operator, + counter: Option, } fn runtime() -> &'static Runtime { @@ -68,7 +80,7 @@ impl FromStr for Throttle { "Parsing ByteSize from throttle string `{string}` failed", err, ) - .attach_context("string", s) + .attach_context("string", s) }) }) .map(|b| -> RusticResult { @@ -79,7 +91,7 @@ impl FromStr for Throttle { "Converting ByteSize `{bytesize}` to u32 failed", err, ) - .attach_context("bytesize", bytesize.to_string()) + .attach_context("bytesize", bytesize.to_string()) }) }); @@ -113,6 +125,31 @@ impl OpenDALBackend { /// /// A new `OpenDAL` backend. pub fn new(path: impl AsRef, options: BTreeMap) -> RusticResult { + Self::new_with_progress(path, options, None) + } + + /// Create a new openDAL backend, optionally attaching a byte-counting layer. + /// + /// # Arguments + /// + /// * `path` - The path to the `OpenDAL` backend. + /// * `options` - Additional options for the `OpenDAL` backend. + /// * `counter` - Optional shared counter. When `Some`, a `ProgressLayer` is + /// attached *outside* the `RetryLayer` so that each logical write is counted + /// exactly once (retried re-transmissions are NOT double-counted). + /// + /// # Errors + /// + /// * If the path is not a valid `OpenDAL` path. + /// + /// # Returns + /// + /// A new `OpenDAL` backend. + pub fn new_with_progress( + path: impl AsRef, + options: BTreeMap, + counter: Option, + ) -> RusticResult { let max_retries = match options.get("retry").map(String::as_str) { Some("false" | "off") => 0, None | Some("default") => constants::DEFAULT_RETRY, @@ -122,7 +159,7 @@ impl OpenDALBackend { "Parsing retry value `{value}` failed, the value must be a valid integer.", err, ) - .attach_context("value", value.to_string()) + .attach_context("value", value.to_string()) })?, }; let connections = options @@ -134,7 +171,7 @@ impl OpenDALBackend { "Parsing connections value `{value}` failed, the value must be a valid integer.", err, ) - .attach_context("value", c) + .attach_context("value", c) }) }) .transpose()?; @@ -157,11 +194,18 @@ impl OpenDALBackend { "Creating Operator from path `{path}` failed. Please check the given schema and options.", err, ) - .attach_context("path", path.as_ref().to_string()) - .attach_context("schema", scheme.to_string()) + .attach_context("path", path.as_ref().to_string()) + .attach_context("schema", scheme.to_string()) })? .layer(RetryLayer::new().with_max_times(max_retries).with_jitter()); + // Attach the byte-counting layer *outside* the RetryLayer so that each + // logical write is counted exactly once (retried re-transmissions are + // NOT double-counted). + if let Some(counter) = &counter { + operator = operator.layer(ProgressLayer::new(counter.clone())); + } + if let Some(Throttle { bandwidth, burst }) = throttle { operator = operator.layer(ThrottleLayer::new(bandwidth, burst)); } @@ -177,10 +221,10 @@ impl OpenDALBackend { "Creating blocking Operator from path `{path}` failed.", err, ) - .attach_context("path", path.as_ref().to_string()) + .attach_context("path", path.as_ref().to_string()) })?; - Ok(Self { operator }) + Ok(Self { operator, counter }) } /// Return a path for the given file type and id. @@ -204,7 +248,7 @@ impl OpenDALBackend { .join(&hex_id[..]), _ => UnixPathBuf::from(tpe.dirname()).join(&hex_id[..]), } - .to_string() + .to_string() } /// Turn this `OpenDALBackend into a ReadSource` @@ -273,7 +317,7 @@ impl ReadBackend for OpenDALBackend { "Path `config` does not exist.", err, ) - .ask_report() + .ask_report() })? { vec![Id::default()] } else { @@ -319,8 +363,8 @@ impl ReadBackend for OpenDALBackend { fn length(entry: &Metadata, file_name: &str, tpe: FileType) -> Option { let length = entry.content_length(); length.try_into().inspect_err(|err| { - error!("Failed to convert file length {length} of {file_name} to u32 while listing {tpe}: {err}"); - }).ok() + error!("Failed to convert file length {length} of {file_name} to u32 while listing {tpe}: {err}"); + }).ok() } trace!("listing tpe: {tpe:?}"); @@ -334,7 +378,7 @@ impl ReadBackend for OpenDALBackend { "Getting Metadata of type `{type}` failed in the backend. Please check if `{type}` exists.", err, ) - .attach_context("type", tpe.to_string()) + .attach_context("type", tpe.to_string()) ), }; } @@ -382,9 +426,9 @@ impl ReadBackend for OpenDALBackend { "Reading file `{path}` failed in the backend. Please check if the given path is correct.", err, ) - .attach_context("path", path) - .attach_context("type", tpe.to_string()) - .attach_context("id", id.to_string()) + .attach_context("path", path) + .attach_context("type", tpe.to_string()) + .attach_context("id", id.to_string()) )? .to_bytes()) } @@ -414,11 +458,11 @@ impl ReadBackend for OpenDALBackend { "Partially reading file `{path}` failed in the backend. Please check if the given path is correct.", err, ) - .attach_context("path", path) - .attach_context("type", tpe.to_string()) - .attach_context("id", id.to_string()) - .attach_context("offset", offset.to_string()) - .attach_context("length", length.to_string()) + .attach_context("path", path) + .attach_context("type", tpe.to_string()) + .attach_context("id", id.to_string()) + .attach_context("offset", offset.to_string()) + .attach_context("length", length.to_string()) )? .to_bytes()) } @@ -456,9 +500,9 @@ impl WriteBackend for OpenDALBackend { "Creating directory `{path}` failed in the backend `{location}`. Please check if the given path is correct.", err, ) - .attach_context("path", path) - .attach_context("location", self.location()) - .attach_context("type", tpe.to_string()) + .attach_context("path", path) + .attach_context("location", self.location()) + .attach_context("type", tpe.to_string()) )?; } // creating 256 dirs can be slow on remote backends, hence we parallelize it. @@ -466,10 +510,10 @@ impl WriteBackend for OpenDALBackend { .into_par_iter() .try_for_each(|i| { let path = UnixPathBuf::from("data") - .join(hex::encode([i])) - .to_string_lossy() - .to_string() - + "/"; + .join(hex::encode([i])) + .to_string_lossy() + .to_string() + + "/"; self.operator.create_dir(&path).map_err(|err| RusticError::with_source( @@ -477,8 +521,8 @@ impl WriteBackend for OpenDALBackend { "Creating directory `{path}` failed in the backend `{location}`. Please check if the given path is correct.", err, ) - .attach_context("path", path) - .attach_context("location", self.location()) + .attach_context("path", path) + .attach_context("location", self.location()) ) })?; @@ -502,16 +546,38 @@ impl WriteBackend for OpenDALBackend { ) -> RusticResult<()> { trace!("writing tpe: {:?}, id: {}", &tpe, &id); let filename = self.path(tpe, id); - _ = self.operator.write(&filename, buf).map_err(|err| { + + let map_write_err = |err| { RusticError::with_source( ErrorKind::Backend, "Writing file `{path}` failed in the backend. Please check if the given path is correct.", err, ) - .attach_context("path", filename) - .attach_context("type", tpe.to_string()) - .attach_context("id", id.to_string()) - })?; + .attach_context("path", filename.clone()) + .attach_context("type", tpe.to_string()) + .attach_context("id", id.to_string()) + }; + + if self.counter.is_some() { + // Progress path: chunked writer so ProgressLayer fires per part. + let write_options = WriteOptions { + chunk: Some(constants::CHUNK_SIZE), + ..Default::default() + }; + let mut writer = self + .operator + .writer_options(&filename, write_options) + .map_err(map_write_err)?; + for chunk in buf.chunks(constants::CHUNK_SIZE) { + writer + .write(Bytes::copy_from_slice(chunk)) + .map_err(map_write_err)?; + } + _ = writer.close().map_err(map_write_err)?; + } else { + // Default path: single one-shot write, unchanged behavior. + _ = self.operator.write(&filename, buf).map_err(map_write_err)?; + } Ok(()) } @@ -532,9 +598,9 @@ impl WriteBackend for OpenDALBackend { "Deleting file `{path}` failed in the backend. Please check if the given path is correct.", err, ) - .attach_context("path", filename) - .attach_context("type", tpe.to_string()) - .attach_context("id", id.to_string()) + .attach_context("path", filename) + .attach_context("type", tpe.to_string()) + .attach_context("id", id.to_string()) })?; Ok(()) } @@ -640,14 +706,14 @@ impl ReadSourceOpen for OpenFile { let reader = || self.0.operator.reader(&path)?.into_std_read(..); let reader = reader() - .map_err(|err| { - RusticError::with_source( - ErrorKind::InputOutput, - "Failed to open file at `{path}`. Please make sure the file exists and is accessible.", - err, - ) - .attach_context("path", path) - })?; + .map_err(|err| { + RusticError::with_source( + ErrorKind::InputOutput, + "Failed to open file at `{path}`. Please make sure the file exists and is accessible.", + err, + ) + .attach_context("path", path) + })?; Ok(reader) } } @@ -687,7 +753,7 @@ impl Iterator for OpenDALLister { open, } })) - .transpose() + .transpose() } } @@ -713,4 +779,4 @@ impl ReadSource for OpenDALReadSource { fn entries(&self) -> Self::Iter { OpenDALLister(self.entries.clone().into_iter(), self.be.clone()) } -} +} \ No newline at end of file diff --git a/crates/backend/src/progress_layer.rs b/crates/backend/src/progress_layer.rs new file mode 100644 index 000000000..02a329be6 --- /dev/null +++ b/crates/backend/src/progress_layer.rs @@ -0,0 +1,115 @@ +// crates/backend/src/progress_layer.rs +//! A custom OpenDAL Layer that accumulates the number of bytes written in real time as data is flushed down to the underlying service. +//! +//! This Layer operates on the async operator (assembled before the blocking wrapper). It intercepts the +//! writer returned by the underlying accessor and accumulates the count after each `write(Buffer)` is passed through to inner. +//! Granularity = each chunk of bytes the underlying service writer receives per call (for S3-like services this is usually each multipart part). + +use std::fmt::Debug; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use opendal::raw::oio; +use opendal::raw::{ + Access, Layer, LayeredAccess, OpList, OpRead, OpWrite, RpDelete, RpList, RpRead, + RpWrite, +}; +use opendal::{Buffer, Metadata, Result}; + +/// Shared handle for the written-bytes counter. Callers (e.g. the JNI layer) keep an `Arc` clone for polling. +pub type WrittenCounter = Arc; + +/// The counting Layer itself, holding the shared counter. +#[derive(Clone, Debug)] +pub struct ProgressLayer { + counter: WrittenCounter, +} + +impl ProgressLayer { + /// Create a Layer with the given counter. + #[must_use] + pub fn new(counter: WrittenCounter) -> Self { + Self { counter } + } +} + +impl Layer for ProgressLayer { + type LayeredAccess = ProgressAccessor; + + fn layer(&self, inner: A) -> Self::LayeredAccess { + ProgressAccessor { + inner, + counter: self.counter.clone(), + } + } +} + +/// Accessor that wraps the underlying accessor. Everything except `write` is passed through to inner. +#[derive(Debug)] +pub struct ProgressAccessor { + inner: A, + counter: WrittenCounter, +} + +impl LayeredAccess for ProgressAccessor { + type Inner = A; + type Reader = A::Reader; + type Writer = ProgressWriter; + type Lister = A::Lister; + type Deleter = A::Deleter; + type Copier = A::Copier; + + fn inner(&self) -> &Self::Inner { + &self.inner + } + + async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> { + // The read path is not counted; pass through directly. + self.inner.read(path, args).await + } + + async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> { + // After obtaining the underlying writer, wrap it with ProgressWriter to inject the counter. + let (rp, writer) = self.inner.write(path, args).await?; + Ok((rp, ProgressWriter::new(writer, self.counter.clone()))) + } + + async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> { + self.inner.delete().await + } + + async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> { + self.inner.list(path, args).await + } +} + +/// Wraps the underlying writer and accumulates the count after each successful `write`. +pub struct ProgressWriter { + inner: W, + counter: WrittenCounter, +} + +impl ProgressWriter { + fn new(inner: W, counter: WrittenCounter) -> Self { + Self { inner, counter } + } +} + +impl oio::Write for ProgressWriter { + async fn write(&mut self, bs: Buffer) -> Result<()> { + // Take the length first (bs will be moved into inner.write afterwards). + let len = bs.len() as u64; + self.inner.write(bs).await?; + // Only count on successful write to avoid inflation from failures/retries. + let _ = self.counter.fetch_add(len, Ordering::Relaxed); + Ok(()) + } + + async fn close(&mut self) -> Result { + self.inner.close().await + } + + async fn abort(&mut self) -> Result<()> { + self.inner.abort().await + } +} \ No newline at end of file diff --git a/crates/backend/tests/progress_layer.rs b/crates/backend/tests/progress_layer.rs new file mode 100644 index 000000000..21bddbef2 --- /dev/null +++ b/crates/backend/tests/progress_layer.rs @@ -0,0 +1,55 @@ +// crates/backend/tests/progress_layer.rs +#![cfg(feature = "opendal")] +#![allow(missing_docs)] + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use anyhow::Result; +use rustic_backend::OpenDALBackend; +use rustic_core::{FileType, Id, ReadBackend, WriteBackend}; + +/// Counting path connectivity: after writing through a backend with a counter, the counter should be >= the number of bytes written. +#[test] +fn progress_layer_counts_written_bytes() -> Result<()> { + // 1) Shared counter + let counter: Arc = Arc::new(AtomicU64::new(0)); + + // 2) Build a backend with the counting layer using the memory service + let be = OpenDALBackend::new_with_progress( + "memory", + BTreeMap::new(), + Some(counter.clone()), + )?; + + // 3) Write a chunk of data + let data = vec![0u8; 4096]; + be.write_bytes(FileType::Pack, &Id::random(), false, data.clone().into())?; + + // 4) Assert the counting path was triggered (memory does not use multipart, may complete in one shot) + let written = counter.load(Ordering::Relaxed); + assert!( + written >= data.len() as u64, + "counter should be >= written bytes; got {written}, expected >= {}", + data.len() + ); + + Ok(()) +} + +/// Default path regression: with counter=None the write behaves normally and does not panic. +#[test] +fn default_path_without_counter_still_writes() -> Result<()> { + let be = OpenDALBackend::new("memory", BTreeMap::new())?; + + let data = vec![1u8; 4096]; + let id = Id::random(); + be.write_bytes(FileType::Pack, &id, false, data.clone().into())?; + + // Read back and verify content matches (default read/write path works) + let read_back = be.read_full(FileType::Pack, &id)?; + assert_eq!(read_back.as_ref(), data.as_slice()); + + Ok(()) +} \ No newline at end of file diff --git a/crates/backend/tests/progress_layer_cos.rs b/crates/backend/tests/progress_layer_cos.rs new file mode 100644 index 000000000..661ddace2 --- /dev/null +++ b/crates/backend/tests/progress_layer_cos.rs @@ -0,0 +1,112 @@ +// crates/backend/tests/progress_layer_cos.rs +#![cfg(feature = "opendal")] +#![allow(missing_docs)] + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::thread; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anyhow::Result; +use rustic_backend::OpenDALBackend; +use rustic_core::{FileType, Id, WriteBackend}; + +/// Per-part granularity verification against a real COS backend. +/// +/// This test is `#[ignore]` by default; it only runs when `--ignored` is explicitly passed, +/// and it requires the following environment variables: +/// COS_SECRET_ID —— Tencent Cloud SecretId +/// COS_SECRET_KEY —— Tencent Cloud SecretKey +/// COS_BUCKET —— bucket name (COS usually needs the appid suffix, e.g. my-bucket-1250000000) +/// COS_ENDPOINT —— region/endpoint (per the opendal 0.57 services-cos docs you use) +/// COS_ROOT —— optional, repository root path +/// +/// If any required variable is missing, it skips immediately (returns Ok(())), ensuring +/// environments without credentials (including the official CI) will not fail. +/// +/// How to run: +/// cargo test -p rustic_backend --features opendal --test progress_layer_cos -- --ignored --nocapture +/// +/// Interpretation: if the counter increases in multiple steps (e.g. 8MiB -> 16MiB -> 24MiB) then ... +/// per-part granularity is in effect; if it jumps to 16MiB instantly and then write_bytes returns +/// only much later, that opendal version / COS service is buffering the whole payload inside the +/// writer, and you need to tune the writer's chunk/buffer size to align with the part size +/// (the adjustment point is in the opendal writer configuration, not the counting layer itself). +#[test] +#[ignore] +fn cos_progress_layer_per_part_granularity() -> Result<()> { + let _ = dotenvy::dotenv(); + // 1) Read credentials from environment variables; skip if any required item is missing. + let (secret_id, secret_key, bucket, endpoint) = match ( + std::env::var("COS_SECRET_ID"), + std::env::var("COS_SECRET_KEY"), + std::env::var("COS_BUCKET"), + std::env::var("COS_ENDPOINT"), + ) { + (Ok(id), Ok(key), Ok(bucket), Ok(endpoint)) => (id, key, bucket, endpoint), + _ => { + eprintln!("COS credentials not set, skipping cos_progress_layer_per_part_granularity"); + return Ok(()); + } + }; + + // 2) Build options. The key names for opendal 0.57 services-cos follow the actual docs. + let mut options: BTreeMap = BTreeMap::new(); + let _ = options.insert("secret_id".to_string(), secret_id); + let _ = options.insert("secret_key".to_string(), secret_key); + let _ = options.insert("bucket".to_string(), bucket); + let _ = options.insert("endpoint".to_string(), endpoint); + + // Optional root: use if let to avoid producing a discarded Option<()>. + if let Ok(root) = std::env::var("COS_ROOT") { + let _ = options.insert("root".to_string(), root); + } + + // 3) Shared counter + backend with the counting layer. + let counter: Arc = Arc::new(AtomicU64::new(0)); + let be = OpenDALBackend::new_with_progress("cos", options, Some(counter.clone()))?; + + // 4) Build pack data spanning >2 multipart parts (128 MiB; S3/COS min part is ~8 MiB). + let data = vec![0u8; 128 * 1024 * 1024]; + let id = Id::random(); + + // 5) Background thread periodically prints the counter to observe whether it increases per part flush. + let stop = Arc::new(AtomicBool::new(false)); + let monitor = { + let counter = counter.clone(); + let stop = stop.clone(); + thread::spawn(move || { + while !stop.load(Ordering::Relaxed) { + let ts = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis()) + .unwrap_or(0); + let written = counter.load(Ordering::Relaxed); + eprintln!("PROGRESS-COS t={ts} counter={written}"); + thread::sleep(Duration::from_millis(200)); + } + }) + }; + + // 6) Perform the upload (blocks until the whole payload is written). + let write_result = be.write_bytes(FileType::Pack, &id, false, data.clone().into()); + + // 7) Stop the monitor thread and wait for it to exit. + stop.store(true, Ordering::Relaxed); + let _ = monitor.join(); + + // 8) Regardless of whether the write succeeded, attempt cleanup to avoid leaving junk objects in the real bucket. + let _ = be.remove(FileType::Pack, &id, false); + + // 9) The write must succeed, and the count must cover at least the logical byte size. + write_result?; + let written = counter.load(Ordering::Relaxed); + assert!( + written >= data.len() as u64, + "counter should be >= written bytes; got {written}, expected >= {}", + data.len() + ); + + Ok(()) +} \ No newline at end of file From c81a9c1b78c50e1870f8e5fbf9bf0d852fa47df6 Mon Sep 17 00:00:00 2001 From: fastace Date: Mon, 10 Aug 2026 11:42:41 +0800 Subject: [PATCH 2/2] feat(backup): add cooperative cancellation via Arc Add a cooperative cancellation mechanism to the backup pipeline so a running backup can be aborted from outside (e.g. an Android/JNI UI thread or a Ctrl-C handler) without relying on any platform-specific signal. Motivation: Previously the backup flow (Repository::backup -> commands::backup::backup -> archive -> Archiver::archive) had no way to be interrupted; it could only return early on error via '?'. Environments like Android have no SIGINT, so cancellation must be driven by an explicit, thread-safe flag. Approach: Introduce an Arc cancellation token threaded through the public backup API down into the archiver's main loop. The token is checked at file/tree-entry granularity; when set, archiving returns ErrorKind::Cancelled via '?' before finalize/save_file, guaranteeing no snapshot is persisted. AtomicBool is used instead of tokio-util's CancellationToken to avoid adding a dependency and because it is Send + Sync and fits the existing pariter parallel structure and rayon scope threads. Changes: - error.rs: add a new ErrorKind::Cancelled variant (enum is #[non_exhaustive], so this is not a breaking change for external matches). - archiver.rs: add a 'cancel: &Arc' parameter to Archiver::archive; poll cancel.load(Ordering::Relaxed) inside the try_for_each closure (before tree_archiver.add) and in the background size-scan thread; return RusticError::new(ErrorKind::Cancelled, ...) when cancelled. - commands/backup.rs: thread 'cancel' through archive() and backup() and pass it into archiver.archive(...). - repository.rs: add the 'cancel: &Arc' parameter to the public Repository::backup and Repository::archive methods and forward it; document the new argument (setting it to true aborts at the next checkpoint and returns ErrorKind::Cancelled without writing a snapshot). - Update all call sites (integration tests + doctests in lib.rs and both README.md files) to pass a non-cancelling token. - Add integration test test_backup_cancelled_writes_no_snapshot asserting that a pre-cancelled token makes backup return a cancellation error and persists no snapshot. Notes: - Cancellation is cooperative and granular to file/tree entries; an in-flight chunked write of a single large file is not interrupted mid-write. - Ordering::Relaxed is sufficient as the flag is a one-way cancel signal. - The JNI/Android trigger side is not part of this crate; rustic_core only exposes the &Arc entry point. Verified locally on Windows: cargo build passes; the cancellation test passes. Not yet verified on a Unix host: tests gated by #[cfg(not(windows))] (e.g. restore::test_restore_preserves_hardlinks, backup::test_backup_excludes_xattr_entries) and external-command tests (echo-based) were not compiled/run on Windows and must be validated on Linux/WSL/macOS. --- README.md | 4 +- crates/core/README.md | 4 +- crates/core/src/archiver.rs | 36 +++-- crates/core/src/commands/backup.rs | 9 +- crates/core/src/error.rs | 2 + crates/core/src/lib.rs | 3 +- crates/core/src/repository.rs | 16 +- crates/core/tests/integration/append_only.rs | 25 ++-- crates/core/tests/integration/backup.rs | 146 ++++++++++++------- crates/core/tests/integration/chunker.rs | 21 +-- crates/core/tests/integration/copy.rs | 17 ++- crates/core/tests/integration/dump.rs | 21 +-- crates/core/tests/integration/find.rs | 23 +-- crates/core/tests/integration/hotcold.rs | 37 ++--- crates/core/tests/integration/ls.rs | 5 +- crates/core/tests/integration/prune.rs | 9 +- crates/core/tests/integration/restore.rs | 5 +- crates/core/tests/integration/rewrite.rs | 35 ++--- crates/core/tests/integration/snapshots.rs | 8 +- crates/core/tests/integration/vfs.rs | 31 ++-- 20 files changed, 272 insertions(+), 185 deletions(-) diff --git a/README.md b/README.md index 41cc67e1d..2ef3ab040 100644 --- a/README.md +++ b/README.md @@ -120,8 +120,8 @@ fn main() -> Result<(), Box> { .to_snapshot()?; // Create snapshot - let snap = repo.backup(&backup_opts, &source, snap)?; - + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let snap = repo.backup(&backup_opts, &source, snap, &cancel)?; println!("successfully created snapshot:\n{snap:#?}"); Ok(()) } diff --git a/crates/core/README.md b/crates/core/README.md index 720b71863..4df19d9ca 100644 --- a/crates/core/README.md +++ b/crates/core/README.md @@ -123,10 +123,12 @@ fn main() -> Result<(), Box> { .to_snapshot()?; // Create snapshot - let snap = repo.backup(&backup_opts, &source, snap)?; + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let snap = repo.backup(&backup_opts, &source, snap, &cancel)?; println!("successfully created snapshot:\n{snap:#?}"); Ok(()) +} ``` ### Example: Restoring a snapshot diff --git a/crates/core/src/archiver.rs b/crates/core/src/archiver.rs index 3fb6d5979..9870861cb 100644 --- a/crates/core/src/archiver.rs +++ b/crates/core/src/archiver.rs @@ -4,6 +4,8 @@ pub(crate) mod tree; pub(crate) mod tree_archiver; use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::thread::scope; use jiff::Zoned; @@ -18,7 +20,7 @@ use crate::{ }, backend::{ReadSource, ReadSourceEntry, decrypt::DecryptFullBackend}, blob::BlobType, - error::RusticResult, + error::{ErrorKind, RusticError, RusticResult}, index::{ ReadGlobalIndex, indexer::{Indexer, SharedIndexer}, @@ -132,6 +134,7 @@ impl<'a, BE: DecryptFullBackend, I: ReadGlobalIndex> Archiver<'a, BE, I> { skip_identical_parent: bool, no_scan: bool, p: &Progress, + cancel: &Arc, ) -> RusticResult where R: ReadSource + 'static, @@ -141,7 +144,7 @@ impl<'a, BE: DecryptFullBackend, I: ReadGlobalIndex> Archiver<'a, BE, I> { scope(|s| -> RusticResult<_> { // determine backup size in parallel to running backup let src_size_handle = s.spawn(|| { - if !no_scan && !p.is_hidden() { + if !no_scan && !p.is_hidden() && !cancel.load(Ordering::Relaxed) { match src.size() { Ok(Some(size)) => p.set_length(size), Ok(None) => {} @@ -191,17 +194,24 @@ impl<'a, BE: DecryptFullBackend, I: ReadGlobalIndex> Archiver<'a, BE, I> { } }, ) - // archive files in parallel - .parallel_map_scoped(s, |item| self.file_archiver.process(item, p)) - .readahead_scoped(s) - .filter_map(|item| match item { - Ok(item) => Some(item), - Err(err) => { - warn!("ignoring error: {}", err.display_log()); - None - } - }) - .try_for_each(|item| self.tree_archiver.add(item))?; + .parallel_map_scoped(s, |item| self.file_archiver.process(item, p)) + .readahead_scoped(s) + .filter_map(|item| match item { + Ok(item) => Some(item), + Err(err) => { + warn!("ignoring error: {}", err.display_log()); + None + } + }) + .try_for_each(|item| { + if cancel.load(Ordering::Relaxed) { + return Err(RusticError::new( + ErrorKind::Cancelled, + "The backup was cancelled by the user.", + )); + } + self.tree_archiver.add(item) + })?; src_size_handle .join() diff --git a/crates/core/src/commands/backup.rs b/crates/core/src/commands/backup.rs index 0237fa87e..70fe41f82 100644 --- a/crates/core/src/commands/backup.rs +++ b/crates/core/src/commands/backup.rs @@ -228,6 +228,7 @@ pub(crate) fn archive( src: &R, mut snap: SnapshotFile, backup_paths: &[PathBuf], + cancel: &std::sync::Arc, ) -> RusticResult where S: IndexedIds, @@ -294,6 +295,7 @@ where opts.parent_opts.skip_if_unchanged, opts.no_scan, &p, + cancel, ) } @@ -326,6 +328,7 @@ pub(crate) fn backup( opts: &BackupOptions, source: &PathList, snap: SnapshotFile, + cancel: &std::sync::Arc, ) -> RusticResult { let backup_stdin = PathList::from_string("-")?; @@ -334,12 +337,12 @@ pub(crate) fn backup( let backup_paths = vec![path.clone()]; if let Some(command) = &opts.stdin_command { let src = ChildStdoutSource::new(command, path)?; - let res = archive(repo, opts, &src, snap, &backup_paths)?; + let res = archive(repo, opts, &src, snap, &backup_paths, cancel)?; src.finish()?; res } else { let src = StdinSource::new(path); - archive(repo, opts, &src, snap, &backup_paths)? + archive(repo, opts, &src, snap, &backup_paths, cancel)? } } else { let backup_path = source.paths(); @@ -349,7 +352,7 @@ pub(crate) fn backup( &opts.ignore_filter_opts, &backup_path, )?; - archive(repo, opts, &src, snap, &backup_path)? + archive(repo, opts, &src, snap, &backup_path, cancel)? }; Ok(snap) diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index 0e72a7837..cb66afde0 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -142,6 +142,8 @@ pub enum ErrorKind { Verification, /// the virtual filesystem Vfs, + /// a cancelled operation + Cancelled, } #[derive(thiserror::Error, Debug)] diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index d6c02edc1..ad6eca5b8 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -72,7 +72,8 @@ implement [`serde::Serialize`] and [`serde::Deserialize`]. let source = PathList::from_string("src").unwrap().sanitize().unwrap(); // run the backup and return the snapshot pointing to the backup'ed data. - let snap = repo.backup(&backup_opts, &source, snap).unwrap(); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let snap = repo.backup(&backup_opts, &source, snap, &cancel).unwrap(); // assert_eq!(&snap.paths, ["src"]); // Get all snapshots from the repository diff --git a/crates/core/src/repository.rs b/crates/core/src/repository.rs index 66a09ffe9..d48ed39a9 100644 --- a/crates/core/src/repository.rs +++ b/crates/core/src/repository.rs @@ -10,6 +10,7 @@ use std::{ io::Write, path::{Path, PathBuf}, sync::Arc, + sync::atomic::AtomicBool, }; use bytes::Bytes; @@ -1671,21 +1672,24 @@ impl Repository { /// * `opts` - The options to use /// * `source` - The source to backup /// * `snap` - The snapshot to modify and save + /// * `cancel` - Cooperative cancel flag; set to `true` externally to abort + /// the backup at the next checkpoint (returns [`ErrorKind::Cancelled`]). /// /// # Errors /// // TODO: Document errors /// /// # Returns - /// + /// /// The saved snapshot. pub fn backup( &self, opts: &BackupOptions, source: &PathList, snap: SnapshotFile, + cancel: &Arc, ) -> RusticResult { - commands::backup::backup(self, opts, source, snap) + commands::backup::backup(self, opts, source, snap, cancel) } /// Run a backup of `source` using a `ReadSource`. @@ -1697,13 +1701,16 @@ impl Repository { /// * `opts` - The options to use /// * `src` - The source to backup /// * `snap` - The snapshot to modify and save + /// * `backup_paths` - The paths to backup + /// * `cancel` - Cooperative cancel flag; set to `true` externally to abort + /// the backup at the next checkpoint (returns [`ErrorKind::Cancelled`]). /// /// # Errors /// // TODO: Document errors /// /// # Returns - /// + /// /// The saved snapshot. pub fn archive( &self, @@ -1711,6 +1718,7 @@ impl Repository { src: &R, snap: SnapshotFile, backup_paths: &[PathBuf], + cancel: &Arc, ) -> RusticResult where S: IndexedIds, @@ -1718,7 +1726,7 @@ impl Repository { ::Open: Send, ::Iter: Send, { - commands::backup::archive(self, opts, src, snap, backup_paths) + commands::backup::archive(self, opts, src, snap, backup_paths, cancel) } } diff --git a/crates/core/tests/integration/append_only.rs b/crates/core/tests/integration/append_only.rs index f87cd7736..774974f33 100644 --- a/crates/core/tests/integration/append_only.rs +++ b/crates/core/tests/integration/append_only.rs @@ -12,39 +12,40 @@ fn test_append_only( tar_gz_testdata: Result, set_up_repo: Result, ) -> Result<()> { - // uncomment for logging output - // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; + // uncomment for logging output + // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; - // Fixtures + // Fixtures let (source, mut repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); - // set repo to append-only mode + // set repo to append-only mode let config_opts = ConfigOptions::default().set_append_only(true); assert!(repo.apply_config(&config_opts)?); let paths = &source.path_list(); - // backup should still work + // backup should still work let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - let snap = repo.backup(&opts, paths, SnapshotFile::default())?; + let snap = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; - // deleting snapshots should fail + // deleting snapshots should fail assert!(repo.delete_snapshots(&[snap.id]).is_err()); - // pruning should fail + // pruning should fail let prune_options = PruneOptions::default(); let prune_plan = repo.prune_plan(&prune_options)?; assert!(repo.prune(&prune_options, prune_plan).is_err()); - // modifying config should fail + // modifying config should fail let config_opts = ConfigOptions::default().set_extra_verify(false); assert!(repo.apply_config(&config_opts).is_err()); - // disable append-only-mode + // disable append-only-mode let config_opts = ConfigOptions::default().set_append_only(false); assert!(repo.apply_config(&config_opts)?); - // operations should now work + // operations should now work repo.delete_snapshots(&[snap.id])?; let prune_plan = repo.prune_plan(&prune_options)?; repo.prune(&prune_options, prune_plan)?; @@ -52,4 +53,4 @@ fn test_append_only( _ = repo.apply_config(&config_opts)?; Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/backup.rs b/crates/core/tests/integration/backup.rs index c7d93d498..2f049d6b3 100644 --- a/crates/core/tests/integration/backup.rs +++ b/crates/core/tests/integration/backup.rs @@ -26,31 +26,32 @@ fn test_backup_with_tar_gz_passes( insta_snapshotfile_redaction: Settings, insta_node_redaction: Settings, ) -> Result<()> { - // uncomment for logging output - // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; + // uncomment for logging output + // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; - // Fixtures + // Fixtures let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - // first backup - let first_snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + // first backup + let first_snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; - // We can also bind to scope ( https://docs.rs/insta/latest/insta/struct.Settings.html#method.bind_to_scope ) - // But I think that can get messy with a lot of tests, also checking which settings are currently applied - // will be probably harder + // We can also bind to scope ( https://docs.rs/insta/latest/insta/struct.Settings.html#method.bind_to_scope ) + // But I think that can get messy with a lot of tests, also checking which settings are currently applied + // will be probably harder insta_snapshotfile_redaction.bind(|| { assert_with_win("backup-tar-summary-first", &first_snapshot); }); assert!(first_snapshot.get_parents().is_empty()); - // tree of first backup - // re-read index + // tree of first backup + // re-read index let repo = repo.to_indexed_ids()?; let tree = repo.node_from_path(first_snapshot.tree, Path::new("test/0/tests"))?; let tree: rustic_core::repofile::Tree = repo.get_tree(&tree.subtree.expect("Sub tree"))?; @@ -59,16 +60,16 @@ fn test_backup_with_tar_gz_passes( assert_with_win("backup-tar-tree", tree); }); - // get all snapshots and check them + // get all snapshots and check them let all_snapshots = repo.get_all_snapshots()?; assert_eq!(vec![first_snapshot.clone()], all_snapshots); - // save list of pack files + // save list of pack files let packs1: Vec = repo.list()?.collect(); - // re-read index + // re-read index let repo = repo.to_indexed_ids()?; - // second backup - let second_snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + // second backup + let second_snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; insta_snapshotfile_redaction.bind(|| { assert_with_win("backup-tar-summary-second", &second_snapshot); @@ -77,19 +78,19 @@ fn test_backup_with_tar_gz_passes( assert_eq!(second_snapshot.get_parents(), &[first_snapshot.id]); assert_eq!(first_snapshot.tree, second_snapshot.tree); - // pack files should be unchanged + // pack files should be unchanged let packs2: Vec<_> = repo.list()?.collect(); assert_eq!(packs1, packs2); - // re-read index + // re-read index let repo = repo.to_indexed_ids()?; - // third backup with tags and explicitly given parent + // third backup with tags and explicitly given parent let snap = SnapshotOptions::default() .tags([StringList::from_str("a,b")?]) .to_snapshot()?; let opts = opts.parent_opts(ParentOptions::default().parents(vec![second_snapshot.id.to_string()])); - let third_snapshot = repo.backup(&opts, paths, snap)?; + let third_snapshot = repo.backup(&opts, paths, snap, &cancel)?; insta_snapshotfile_redaction.bind(|| { assert_with_win("backup-tar-summary-third", &third_snapshot); @@ -97,7 +98,7 @@ fn test_backup_with_tar_gz_passes( assert_eq!(third_snapshot.get_parents(), &[second_snapshot.id]); assert_eq!(third_snapshot.tree, second_snapshot.tree); - // get all snapshots and check them + // get all snapshots and check them let mut all_snapshots = repo.get_all_snapshots()?; all_snapshots.sort_unstable(); assert_eq!( @@ -105,18 +106,18 @@ fn test_backup_with_tar_gz_passes( all_snapshots ); - // pack files should be unchanged + // pack files should be unchanged let packs2: Vec<_> = repo.list()?.collect(); assert_eq!(packs1, packs2); let packs3: Vec<_> = repo.list()?.collect(); assert_eq!(packs1, packs3); - // Check if snapshots can be retrieved + // Check if snapshots can be retrieved let mut ids: Vec<_> = all_snapshots.iter().map(|sn| sn.id.to_string()).collect(); let snaps = repo.get_snapshots(&ids)?; assert_eq!(snaps, all_snapshots); - // reverse order and add duplicate snapshot - test if update_snapshots works as expected + // reverse order and add duplicate snapshot - test if update_snapshots works as expected all_snapshots.reverse(); all_snapshots.push(first_snapshot.clone()); ids.reverse(); @@ -124,11 +125,11 @@ fn test_backup_with_tar_gz_passes( let snaps = repo.update_snapshots(snaps, &ids)?; assert_eq!(snaps, all_snapshots); - // get snapshot group + // get snapshot group let group_by = SnapshotGroupCriterion::new().tags(true); let mut snapshots = repo.get_all_snapshots()?; - // sort to get unique result + // sort to get unique result snapshots.sort(); insta_snapshotfile_redaction.bind(|| { @@ -138,7 +139,7 @@ fn test_backup_with_tar_gz_passes( ); }); - // filter snapshots by tag + // filter snapshots by tag let filter = |snap: &SnapshotFile| snap.tags.contains("a"); let snaps = repo.get_matching_snapshots(filter)?; insta_snapshotfile_redaction.bind(|| { @@ -155,37 +156,38 @@ fn test_backup_dry_run_with_tar_gz_passes( insta_snapshotfile_redaction: Settings, insta_node_redaction: Settings, ) -> Result<()> { - // Fixtures + // Fixtures let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default() .as_path(PathBuf::from_str("test")?) .dry_run(true); - // dry-run backup - let snap_dry_run = repo.backup(&opts, paths, SnapshotFile::default())?; + // dry-run backup + let snap_dry_run = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; insta_snapshotfile_redaction.bind(|| { assert_with_win("dryrun-tar-summary-first", &snap_dry_run); }); - // check that repo is still empty + // check that repo is still empty let snaps = repo.get_all_snapshots()?; assert_eq!(snaps.len(), 0); assert_eq!(repo.list::()?.count(), 0); assert_eq!(repo.list::()?.count(), 0); - // first real backup + // first real backup let opts = opts.dry_run(false); - let first_snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + let first_snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; assert_eq!(snap_dry_run.tree, first_snapshot.tree); let packs: Vec<_> = repo.list::()?.collect(); - // tree of first backup - // re-read index + // tree of first backup + // re-read index let repo = repo.to_indexed_ids()?; let tree = repo.node_from_path(first_snapshot.tree, Path::new("test/0/tests"))?; let tree = repo.get_tree(&tree.subtree.expect("Sub tree"))?; @@ -194,27 +196,27 @@ fn test_backup_dry_run_with_tar_gz_passes( assert_with_win("dryrun-tar-tree", tree); }); - // re-read index + // re-read index let repo = repo.to_indexed_ids()?; - // second dry-run backup + // second dry-run backup let opts = opts.dry_run(true); - let snap_dry_run = repo.backup(&opts, paths, SnapshotFile::default())?; + let snap_dry_run = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; insta_snapshotfile_redaction.bind(|| { assert_with_win("dryrun-tar-summary-second", &snap_dry_run); }); - // check that no data has been added + // check that no data has been added let snaps = repo.get_all_snapshots()?; assert_eq!(snaps, vec![first_snapshot]); let packs_dry_run: Vec = repo.list()?.collect(); assert_eq!(packs_dry_run, packs); - // re-read index + // re-read index let repo = repo.to_indexed_ids()?; - // second real backup + // second real backup let opts = opts.dry_run(false); - let second_snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + let second_snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; assert_eq!(snap_dry_run.tree, second_snapshot.tree); Ok(()) } @@ -224,24 +226,25 @@ fn test_backup_stdin_command( set_up_repo: Result, insta_snapshotfile_redaction: Settings, ) -> Result<()> { - // Fixtures + // Fixtures let repo = set_up_repo?.to_indexed_ids()?; let paths = PathList::from_string("-")?; + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let cmd: CommandInput = "echo test".parse()?; let opts = BackupOptions::default() .stdin_filename("test") .stdin_command(cmd); - // backup data from cmd - let snapshot = repo.backup(&opts, &paths, SnapshotFile::default())?; + // backup data from cmd + let snapshot = repo.backup(&opts, &paths, SnapshotFile::default(), &cancel)?; insta_snapshotfile_redaction.bind(|| { assert_with_win("stdin-command-summary", &snapshot); }); - // re-read index + // re-read index let repo = repo.to_indexed()?; - // check content + // check content let node = repo.node_from_snapshot_path("latest:test", |_| true)?; let mut content = Vec::new(); repo.dump(&node, &mut content)?; @@ -282,6 +285,7 @@ fn test_backup_excludes_xattr_entries(set_up_repo: Result) -> Result<( let repo = set_up_repo?.to_indexed_ids()?; let paths = PathList::from_iter(Some(base.to_path_buf())); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let filter_opts = LocalSourceFilterOptions::default() .exclude_if_xattr(vec!["user.rustic_test_exclude".to_string()]); @@ -289,7 +293,7 @@ fn test_backup_excludes_xattr_entries(set_up_repo: Result) -> Result<( .as_path(PathBuf::from("test")) .ignore_filter_opts(filter_opts); - let snapshot = repo.backup(&opts, &paths, SnapshotFile::default())?; + let snapshot = repo.backup(&opts, &paths, SnapshotFile::default(), &cancel)?; let repo = repo.to_indexed_ids()?; let mut root_node = Node::new_node(OsStr::new(""), NodeType::Dir, Metadata::default()); @@ -307,3 +311,47 @@ fn test_backup_excludes_xattr_entries(set_up_repo: Result) -> Result<( Ok(()) } + +/// The cancel token is set to `true` before the backup starts, so the backup +/// should return `ErrorKind::Cancelled` at the very first checkpoint and must +/// not write out any snapshot. +#[rstest] +fn test_backup_cancelled_writes_no_snapshot( + tar_gz_testdata: Result, + set_up_repo: Result, +) -> Result<()> { + // Fixtures + let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let paths = &source.path_list(); + + let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); + + // Set the token to cancelled before the backup so the first checkpoint is + // hit immediately, making the test deterministic (no thread race). + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true)); + + // The backup should return an error instead of a snapshot. + let err = repo + .backup(&opts, paths, SnapshotFile::default(), &cancel) + .unwrap_err(); + + // Assert this is a cancellation error. + // The guidance text is "The backup was cancelled by the user.", and the + // Display output also contains the kind text "a cancelled operation"; + // both contain "cancel". + assert!( + err.to_string().to_lowercase().contains("cancel"), + "expected a cancellation error, got: {err}" + ); + + // A cancelled backup must not persist any snapshot. + let repo = repo.to_indexed_ids()?; + let snapshots = repo.get_all_snapshots()?; + assert!( + snapshots.is_empty(), + "cancelled backup must not persist a snapshot, found {}", + snapshots.len() + ); + + Ok(()) +} \ No newline at end of file diff --git a/crates/core/tests/integration/chunker.rs b/crates/core/tests/integration/chunker.rs index 45fa9a17c..8b7ef48d3 100644 --- a/crates/core/tests/integration/chunker.rs +++ b/crates/core/tests/integration/chunker.rs @@ -18,31 +18,32 @@ fn test_chunker_params( set_up_repo: Result, insta_snapshotfile_redaction: Settings, ) -> Result<()> { - // uncomment for logging output - // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; + // uncomment for logging output + // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; - // Fixtures + // Fixtures let (source, mut repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); - // set fixed size chunker with a given chunk size + // set fixed size chunker with a given chunk size let config_opts = ConfigOptions::default() .set_chunker(Chunker::FixedSize) .set_chunk_size(ByteSize(8000)); assert!(repo.apply_config(&config_opts)?); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - let snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + let snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; - // We can also bind to scope ( https://docs.rs/insta/latest/insta/struct.Settings.html#method.bind_to_scope ) - // But I think that can get messy with a lot of tests, also checking which settings are currently applied - // will be probably harder + // We can also bind to scope ( https://docs.rs/insta/latest/insta/struct.Settings.html#method.bind_to_scope ) + // But I think that can get messy with a lot of tests, also checking which settings are currently applied + // will be probably harder insta_snapshotfile_redaction.bind(|| { assert_ron_snapshot!("chunker-fixedsize", &snapshot); }); Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/copy.rs b/crates/core/tests/integration/copy.rs index a4a52f0ec..efc3f505f 100644 --- a/crates/core/tests/integration/copy.rs +++ b/crates/core/tests/integration/copy.rs @@ -10,21 +10,22 @@ use super::{RepoOpen, TestSource, set_up_repo, tar_gz_testdata}; #[rstest] fn test_copy(tar_gz_testdata: Result, set_up_repo: Result) -> Result<()> { - // uncomment for logging output - // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; + // uncomment for logging output + // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; - // Fixtures + // Fixtures let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - // first backup - let snap = repo.backup(&opts, paths, SnapshotFile::default())?; + // first backup + let snap = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; - // re-read index + // re-read index let repo = repo.to_indexed()?; let target = super::set_up_repo()?; @@ -43,4 +44,4 @@ fn test_copy(tar_gz_testdata: Result, set_up_repo: Result) target.check(check_opts)?.is_ok()?; Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/dump.rs b/crates/core/tests/integration/dump.rs index 42a9a4e51..7e1a759b5 100644 --- a/crates/core/tests/integration/dump.rs +++ b/crates/core/tests/integration/dump.rs @@ -13,18 +13,18 @@ use rustic_core::{ use super::{RepoOpen, set_up_repo}; -/// Build a deterministic byte payload of the requested length. +/// Build a deterministic byte payload of the requested length. fn payload(len: usize) -> Vec { (0..len) .map(|i| u8::try_from(i % 251).expect("251 always fits in u8")) .collect() } -/// Backup a single file with the given content into `repo`, configuring the -/// fixed-size chunker so the file reliably splits into multiple blobs. -/// -/// Returns the repository in the [`IndexedFullStatus`] state along with the -/// snapshot path that points at the backed-up file. +/// Backup a single file with the given content into `repo`, configuring the +/// fixed-size chunker so the file reliably splits into multiple blobs. +/// +/// Returns the repository in the [`IndexedFullStatus`] state along with the +/// snapshot path that points at the backed-up file. fn backup_single_file( repo: RepoOpen, name: &str, @@ -42,7 +42,8 @@ fn backup_single_file( let paths = PathList::from_iter([file_path]); let opts = BackupOptions::default().as_path(PathBuf::from_str(name)?); - let _snapshot = repo.backup(&opts, &paths, SnapshotFile::default())?; + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let _snapshot = repo.backup(&opts, &paths, SnapshotFile::default(), &cancel)?; Ok((repo.to_indexed()?, format!("latest:{name}"))) } @@ -53,8 +54,8 @@ fn test_dump_multi_blob_matches_source(set_up_repo: Result) -> Result< let (repo, snapshot_path) = backup_single_file(set_up_repo?, "file.bin", &data)?; let node = repo.node_from_snapshot_path(&snapshot_path, |_| true)?; - // Sanity: the configured chunker must have produced more than one blob, - // otherwise the parallel path is never taken. + // Sanity: the configured chunker must have produced more than one blob, + // otherwise the parallel path is never taken. let blob_count = node.content.as_ref().map_or(0, Vec::len); assert!( blob_count > 1, @@ -77,4 +78,4 @@ fn test_dump_default_options_match_source(set_up_repo: Result) -> Resu repo.dump(&node, &mut out)?; assert_eq!(out, data); Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/find.rs b/crates/core/tests/integration/find.rs index 787abca1b..426f80fe9 100644 --- a/crates/core/tests/integration/find.rs +++ b/crates/core/tests/integration/find.rs @@ -16,32 +16,33 @@ use super::{RepoOpen, TestSource, assert_with_win, set_up_repo, tar_gz_testdata} #[rstest] fn test_find(tar_gz_testdata: Result, set_up_repo: Result) -> Result<()> { - // Fixtures + // Fixtures let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - // backup test-data - let snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + // backup test-data + let snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; - // re-read index + // re-read index let repo = repo.to_indexed_ids()?; - // test non-existing path + // test non-existing path let not_found = repo.find_nodes_from_path(vec![snapshot.tree], Path::new("not_existing"))?; assert_with_win("find-nodes-not-found", not_found); - // test non-existing match + // test non-existing match let glob = Glob::new("not_existing")?.compile_matcher(); let not_found = repo.find_matching_nodes(vec![snapshot.tree], &|path, _| glob.is_match(path))?; assert_with_win("find-matching-nodes-not-found", not_found); - // test existing path + // test existing path let FindNode { matches, .. } = repo.find_nodes_from_path(vec![snapshot.tree], Path::new("test/0/tests/testfile"))?; assert_with_win("find-nodes-existing", matches); - // test existing match + // test existing match let glob = Glob::new("testfile")?.compile_matcher(); let match_func = |path: &Path, _: &Node| { glob.is_match(path) || path.file_name().is_some_and(|f| glob.is_match(f)) @@ -49,7 +50,7 @@ fn test_find(tar_gz_testdata: Result, set_up_repo: Result) let FindMatches { paths, matches, .. } = repo.find_matching_nodes(vec![snapshot.tree], &match_func)?; assert_with_win("find-matching-existing", (paths, matches)); - // test existing match + // test existing match let glob = Glob::new("testfile*")?.compile_matcher(); let match_func = |path: &Path, _: &Node| { glob.is_match(path) || path.file_name().is_some_and(|f| glob.is_match(f)) @@ -58,4 +59,4 @@ fn test_find(tar_gz_testdata: Result, set_up_repo: Result) repo.find_matching_nodes(vec![snapshot.tree], &match_func)?; assert_with_win("find-matching-wildcard-existing", (paths, matches)); Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/hotcold.rs b/crates/core/tests/integration/hotcold.rs index c08164ccb..50e02ac2c 100644 --- a/crates/core/tests/integration/hotcold.rs +++ b/crates/core/tests/integration/hotcold.rs @@ -14,7 +14,7 @@ use super::{TestSource, tar_gz_testdata}; #[rstest] fn hot_cold(tar_gz_testdata: Result) -> Result<()> { - // Fixtures + // Fixtures let source = tar_gz_testdata?; let be_hot = InMemoryBackend::new(); @@ -31,25 +31,26 @@ fn hot_cold(tar_gz_testdata: Result) -> Result<()> { .to_indexed_ids()?; let paths = &source.path_list(); + let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false)); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - // backup - let snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + // backup + let snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; - // get all snapshots and check them + // get all snapshots and check them let all_snapshots = repo.get_all_snapshots()?; assert_eq!(vec![snapshot], all_snapshots); repo.check(CheckOptions::default())?.is_ok()?; - // check with read_data should fail - as accessing packs from the cold storage is not implemented + // check with read_data should fail - as accessing packs from the cold storage is not implemented assert!( repo.check(CheckOptions::default().read_data(true))? .is_ok() .is_err() ); - // remove keys, config and index files from hot repository + // remove keys, config and index files from hot repository for tpe in [FileType::Key, FileType::Config, FileType::Index] { for id in be.repo_hot().unwrap().list(tpe)? { be.repo_hot().unwrap().remove(tpe, &id, false)?; @@ -57,49 +58,49 @@ fn hot_cold(tar_gz_testdata: Result) -> Result<()> { } assert!(repo.check(CheckOptions::default())?.is_ok().is_err()); - // repo cannot be opened normally + // repo cannot be opened normally let repo = Repository::new(&options, &be)?; assert!(repo.open(&creds).is_err()); - // but with open_with_password_only_cold + // but with open_with_password_only_cold let repo = Repository::new(&options, &be)?; let repo = repo.open_only_cold(&Credentials::password("test"))?; - // repair repository + // repair repository repo.init_hot()?; repo.repair_hotcold_except_packs(false)?; - // now we should be able to open the repository again. + // now we should be able to open the repository again. let repo = Repository::new(&options, &be)?.open(&creds)?; - // remove pack files from hot repository + // remove pack files from hot repository for id in be.repo_hot().unwrap().list(FileType::Pack)? { be.repo_hot().unwrap().remove(FileType::Pack, &id, true)?; } assert!(repo.check(CheckOptions::default())?.is_ok().is_err()); - // repair + // repair repo.repair_hotcold_packs(false)?; repo.check(CheckOptions::default())?.is_ok()?; - // remove index files from cold repository + // remove index files from cold repository for id in be_cold.list(FileType::Index)? { be_cold.remove(FileType::Index, &id, true)?; } assert!(repo.check(CheckOptions::default())?.is_ok().is_err()); - // repair + // repair repo.repair_hotcold_except_packs(false)?; repo.check(CheckOptions::default())?.is_ok()?; - // remove tree pack files from cold repository + // remove tree pack files from cold repository for id in be.repo_hot().unwrap().list(FileType::Pack)? { be_cold.remove(FileType::Pack, &id, true)?; } assert!(repo.check(CheckOptions::default())?.is_ok().is_err()); - // repair + // repair repo.repair_hotcold_packs(false)?; repo.check(CheckOptions::default())?.is_ok()?; Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/ls.rs b/crates/core/tests/integration/ls.rs index 00452258b..7aaff4023 100644 --- a/crates/core/tests/integration/ls.rs +++ b/crates/core/tests/integration/ls.rs @@ -22,12 +22,13 @@ fn test_ls( ) -> Result<()> { // Fixtures let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); // backup test-data - let snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + let snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; // test non-existing entries let mut node = Node::new_node( @@ -49,4 +50,4 @@ fn test_ls( }); Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/prune.rs b/crates/core/tests/integration/prune.rs index 70c27f605..e16e22688 100644 --- a/crates/core/tests/integration/prune.rs +++ b/crates/core/tests/integration/prune.rs @@ -33,24 +33,25 @@ fn test_prune( // Fixtures let (source, mut repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); _ = repo.apply_config(&opts)?; + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let opts = BackupOptions::default(); // first backup let paths = PathList::from_iter(Some(source.0.path().join("0/0/9"))); - let snapshot1 = repo.backup(&opts, &paths, SnapshotFile::default())?; + let snapshot1 = repo.backup(&opts, &paths, SnapshotFile::default(), &cancel)?; // re-read index let repo = repo.to_indexed_ids()?; // second backup let paths = PathList::from_iter(Some(source.0.path().join("0/0/9/2"))); - let _ = repo.backup(&opts, &paths, SnapshotFile::default())?; + let _ = repo.backup(&opts, &paths, SnapshotFile::default(), &cancel)?; // re-read index let repo = repo.to_indexed_ids()?; // third backup let paths = PathList::from_iter(Some(source.0.path().join("0/0/9/3"))); - let _ = repo.backup(&opts, &paths, SnapshotFile::default())?; + let _ = repo.backup(&opts, &paths, SnapshotFile::default(), &cancel)?; // drop index let repo = repo.drop_index(); @@ -79,4 +80,4 @@ fn test_prune( } Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/restore.rs b/crates/core/tests/integration/restore.rs index dc3b49de6..74771c4e4 100644 --- a/crates/core/tests/integration/restore.rs +++ b/crates/core/tests/integration/restore.rs @@ -21,9 +21,10 @@ fn test_restore_preserves_hardlinks( set_up_repo: Result, ) -> Result<()> { let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - let _snapshot = repo.backup(&opts, &source.path_list(), SnapshotFile::default())?; + let _snapshot = repo.backup(&opts, &source.path_list(), SnapshotFile::default(), &cancel)?; let repo = repo.to_indexed()?; let node = repo.node_from_snapshot_path("latest", |_| true)?; @@ -57,4 +58,4 @@ fn test_restore_preserves_hardlinks( assert_eq!(fs::read_link(&symlink)?, PathBuf::from("testfile")); Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/rewrite.rs b/crates/core/tests/integration/rewrite.rs index 3f23b1179..6513970e3 100644 --- a/crates/core/tests/integration/rewrite.rs +++ b/crates/core/tests/integration/rewrite.rs @@ -24,19 +24,20 @@ fn test_rewrite( insta_snapshotfile_redaction: Settings, insta_node_redaction: Settings, ) -> Result<()> { - // uncomment for logging output - // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; + // uncomment for logging output + // SimpleLogger::init(log::LevelFilter::Debug, Config::default())?; - // Fixtures + // Fixtures let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let backup_opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - // first backup - let snapshot = repo.backup(&backup_opts, paths, SnapshotFile::default())?; + // first backup + let snapshot = repo.backup(&backup_opts, paths, SnapshotFile::default(), &cancel)?; let modification = SnapshotModification::default() .set_label("label".to_string()) @@ -62,7 +63,7 @@ fn test_rewrite( .excludes(Excludes::default().globs(vec!["!/test/0/0/9/6*".to_string()])) .node_modification(NodeModification::default()); - // with dry_run + // with dry_run rewrite_opts.dry_run = true; let rewrite_snaps_dryrun = repo.rewrite_snapshots_and_trees(snaps.clone(), &rewrite_opts, &rewrite_tree_params)?; @@ -78,13 +79,13 @@ fn test_rewrite( assert_eq!(rewrite_snaps_dryrun, rewrite_snaps); assert_eq!(rewrite_snaps.len(), 1); - // re-read index + // re-read index let repo = repo.to_indexed_ids()?; - // re-read index + // re-read index let repo = repo.to_indexed_ids()?; - // test entries + // test entries let mut node = Node::new_node( OsStr::new(""), rustic_core::repofile::NodeType::Dir, @@ -100,17 +101,17 @@ fn test_rewrite( assert_with_win("rewrite-nodes", &entries); }); - // backup with excludes - let glob = "!".to_string() + source.path().to_str().unwrap() + "/0/0/9/6*"; // other exclude as we use as-path + // backup with excludes + let glob = "!".to_string() + source.path().to_str().unwrap() + "/0/0/9/6*"; // other exclude as we use as-path - // #[cfg(windows)] - let glob = glob.replace('\\', "/"); // correct windows paths for glob + // #[cfg(windows)] + let glob = glob.replace('\\', "/"); // correct windows paths for glob let excludes = Excludes::default().globs(vec![glob]); let backup_opts = backup_opts.excludes(excludes); - let snapshot = repo.backup(&backup_opts, paths, SnapshotFile::default())?; - // trees should be identical + let snapshot = repo.backup(&backup_opts, paths, SnapshotFile::default(), &cancel)?; + // trees should be identical assert_eq!(snapshot.tree, rewrite_snaps[0].tree); Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/snapshots.rs b/crates/core/tests/integration/snapshots.rs index d06654d42..c6b50203b 100644 --- a/crates/core/tests/integration/snapshots.rs +++ b/crates/core/tests/integration/snapshots.rs @@ -18,6 +18,7 @@ use rustic_core::{BackupOptions, Grouped, IndexedIdsStatus, Repository, Snapshot fn repo_and_snapshots() -> (Repository, Vec) { let repo = set_up_repo().unwrap().to_indexed_ids().unwrap(); let source = tar_gz_testdata().unwrap(); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let snapshot_timestamp = [ Timestamp::from_second(1_752_483_600) @@ -32,7 +33,7 @@ fn repo_and_snapshots() -> (Repository, Vec) { ]; let mut snapshot_files = Vec::new(); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let backup_options = BackupOptions::default().as_path(PathBuf::from_str("test").unwrap()); for snap_ts in snapshot_timestamp { let snapshot_file = repo @@ -43,6 +44,7 @@ fn repo_and_snapshots() -> (Repository, Vec) { time: snap_ts, ..Default::default() }, + &cancel, ) .unwrap(); snapshot_files.push(snapshot_file); @@ -88,7 +90,7 @@ fn test_get_snapshot_latest_id( let (repo, snapshots) = repo_and_snapshots; let res = repo.get_snapshots_from_strs(&[String::from("latest")], |_| true)?; - // latest => most recent + // latest => most recent assert_eq!(res, vec![snapshots[2].clone()]); Ok(()) } @@ -150,4 +152,4 @@ fn test_get_snapshots_from_strs_latest( assert_eq!(snap_latest[0], snapshots[2]); assert_eq!(snap_latest[1], snapshots[1]); Ok(()) -} +} \ No newline at end of file diff --git a/crates/core/tests/integration/vfs.rs b/crates/core/tests/integration/vfs.rs index a2fd819af..ea2c4610d 100644 --- a/crates/core/tests/integration/vfs.rs +++ b/crates/core/tests/integration/vfs.rs @@ -18,43 +18,44 @@ fn test_vfs( set_up_repo: Result, insta_node_redaction: Settings, ) -> Result<()> { - // Fixtures + // Fixtures let (source, repo) = (tar_gz_testdata?, set_up_repo?.to_indexed_ids()?); + let cancel = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); let paths = &source.path_list(); - // we use as_path to not depend on the actual tempdir + // we use as_path to not depend on the actual tempdir let opts = BackupOptions::default().as_path(PathBuf::from_str("test")?); - // backup test-data - let snapshot = repo.backup(&opts, paths, SnapshotFile::default())?; + // backup test-data + let snapshot = repo.backup(&opts, paths, SnapshotFile::default(), &cancel)?; - // re-read index + // re-read index let repo = repo.to_indexed()?; - // create Vfs + // create Vfs let node = repo.node_from_snapshot_and_path(&snapshot, "")?; let vfs = Vfs::from_dir_node(&node); - // test reading a directory using vfs + // test reading a directory using vfs let path: PathBuf = ["test", "0", "tests"].iter().collect(); let entries = vfs.dir_entries_from_path(&repo, &path)?; insta_node_redaction.bind(|| { assert_with_win("vfs", &entries); }); - // test reading a file from the repository + // test reading a file from the repository let path: PathBuf = ["test", "0", "tests", "testfile"].iter().collect(); let node = vfs.node_from_path(&repo, &path)?; let file = repo.open_file(&node)?; - let data = repo.read_file_at(&file, 0, 21)?; // read full content + let data = repo.read_file_at(&file, 0, 21)?; // read full content assert_eq!(Bytes::from("This is a test file.\n"), &data); - assert_eq!(data, repo.read_file_at(&file, 0, 4096)?); // read beyond file end - assert_eq!(Bytes::new(), repo.read_file_at(&file, 25, 1)?); // offset beyond file end - assert_eq!(Bytes::from("test"), repo.read_file_at(&file, 10, 4)?); // read partial content + assert_eq!(data, repo.read_file_at(&file, 0, 4096)?); // read beyond file end + assert_eq!(Bytes::new(), repo.read_file_at(&file, 25, 1)?); // offset beyond file end + assert_eq!(Bytes::from("test"), repo.read_file_at(&file, 10, 4)?); // read partial content - // test reading an empty file from the repository + // test reading an empty file from the repository let path: PathBuf = ["test", "0", "tests", "empty-file"].iter().collect(); let node = vfs.node_from_path(&repo, &path)?; let file = repo.open_file(&node)?; - assert_eq!(Bytes::new(), repo.read_file_at(&file, 0, 0)?); // empty files + assert_eq!(Bytes::new(), repo.read_file_at(&file, 0, 0)?); // empty files Ok(()) -} +} \ No newline at end of file