diff --git a/src/config.rs b/src/config.rs index 4144a58..bcb3f31 100644 --- a/src/config.rs +++ b/src/config.rs @@ -49,7 +49,10 @@ impl AppaConfig { } let config = Self { version: CONFIG_VERSION, - folders: folders.iter().map(ConfiguredFolder::from_folder).collect(), + folders: folders + .iter() + .map(ConfiguredFolder::from_storage_config) + .collect(), }; fs::write(path, toml::to_string_pretty(&config)?)?; Ok(()) @@ -86,7 +89,7 @@ impl AppaConfig { } impl ConfiguredFolder { - fn from_folder(folder: &FolderConfig) -> Self { + fn from_storage_config(folder: &FolderConfig) -> Self { Self { path: folder.path.clone(), name: Some(folder.name.clone()), diff --git a/src/filesystem.rs b/src/filesystem.rs index 919d669..1dcd79a 100644 --- a/src/filesystem.rs +++ b/src/filesystem.rs @@ -6,6 +6,7 @@ use std::{ sync::mpsc::{self, Receiver}, }; +use anyhow::Context; use ignore::gitignore::{Gitignore, GitignoreBuilder}; use notify::{EventKind, RecursiveMode, Watcher}; @@ -52,13 +53,21 @@ pub fn collect_entries( files: Vec::new(), directories: Vec::new(), }; - collect_entries_into(directory, ignore_rules, &mut entries)?; + collect_entries_into(directory, ignore_rules, &mut entries) + .with_context(|| format!("could not scan {}", directory.display()))?; Ok(entries) } pub fn relative_path(folder: &Path, file_path: &Path) -> anyhow::Result { let relative_path = file_path - .strip_prefix(folder)? + .strip_prefix(folder) + .with_context(|| { + format!( + "{} is not inside synced folder {}", + file_path.display(), + folder.display() + ) + })? .to_str() .ok_or_else(|| anyhow::anyhow!("path contains non-Unicode characters: {file_path:?}"))?; Ok(relative_path.replace('\\', "/")) @@ -103,7 +112,9 @@ pub fn watch_folder(path: &Path) -> anyhow::Result { let _ = sender.send(()); } })?; - watcher.watch(path, RecursiveMode::Recursive)?; + watcher + .watch(path, RecursiveMode::Recursive) + .with_context(|| format!("could not watch {}", path.display()))?; Ok(FolderChangeWatcher { _watcher: watcher, events: receiver, @@ -125,9 +136,15 @@ pub struct FolderLock { } pub fn lock_app_process(lock_directory: &Path) -> anyhow::Result { - fs::create_dir_all(lock_directory)?; + fs::create_dir_all(lock_directory).with_context(|| { + format!( + "could not create Appa lock directory {}", + lock_directory.display() + ) + })?; let lock_path = lock_directory.join("appa.lock"); - let file = File::create(lock_path)?; + let file = File::create(&lock_path) + .with_context(|| format!("could not open Appa process lock {}", lock_path.display()))?; file.try_lock() .map_err(|error| anyhow::anyhow!("another Appa process is already running: {error}"))?; Ok(FolderLock { _file: file }) @@ -138,9 +155,14 @@ fn collect_entries_into( ignore_rules: &IgnoreRules, entries: &mut CollectedEntries, ) -> anyhow::Result<()> { - for entry in fs::read_dir(directory)? { - let path = entry?.path(); - let metadata = fs::symlink_metadata(&path)?; + for entry in fs::read_dir(directory) + .with_context(|| format!("could not read directory {}", directory.display()))? + { + let path = entry + .with_context(|| format!("could not read an entry in {}", directory.display()))? + .path(); + let metadata = fs::symlink_metadata(&path) + .with_context(|| format!("could not inspect {}", path.display()))?; if metadata.file_type().is_symlink() { tracing::warn!(path = %path.display(), "Skipping symlink in synced folder"); continue; diff --git a/src/iroh.rs b/src/iroh.rs index 055fa36..8a99788 100644 --- a/src/iroh.rs +++ b/src/iroh.rs @@ -2,22 +2,25 @@ use std::{ collections::{BTreeMap, BTreeSet}, + future::IntoFuture, path::Path, sync::{Arc, Mutex}, }; +use ::iroh as iroh_network; +use anyhow::Context; use futures_util::{StreamExt, stream}; -use iroh::{ - Endpoint, EndpointAddr, EndpointId, SecretKey, - endpoint::{Connection, presets}, - protocol::Router, -}; use iroh_blobs::{ BlobsProtocol, Hash, provider::events::{EventMask, EventSender, ProviderMessage, RequestMode}, store::fs::FsStore, }; use iroh_mdns_address_lookup::{DiscoveryEvent, MdnsAddressLookup}; +use iroh_network::{ + Endpoint, EndpointAddr, EndpointId, SecretKey, + endpoint::{Connection, presets}, + protocol::Router, +}; use tokio::sync::RwLock; use crate::{ @@ -98,7 +101,12 @@ impl NodeHost { lan_discovery_enabled: bool, ) -> anyhow::Result { let blob_directory = data_directory.join("blobs"); - std::fs::create_dir_all(&blob_directory)?; + std::fs::create_dir_all(&blob_directory).with_context(|| { + format!( + "could not create Appa blob store at {}", + blob_directory.display() + ) + })?; let endpoint = Endpoint::builder(presets::N0) .secret_key(identity) .bind() @@ -174,13 +182,11 @@ impl NodeHost { } pub async fn import_file(&self, file_path: &Path) -> anyhow::Result { - Ok(tokio::time::timeout( - BLOB_OPERATION_TIMEOUT, - self.store.blobs().add_path(file_path), + Ok( + wait_for_blob_operation("import", self.store.blobs().add_path(file_path)) + .await? + .hash, ) - .await - .map_err(|_| anyhow::anyhow!("blob import timed out"))?? - .hash) } pub async fn download_blob( @@ -188,24 +194,19 @@ impl NodeHost { blob_hash: Hash, provider: EndpointAddr, ) -> anyhow::Result<()> { - tokio::time::timeout( - BLOB_OPERATION_TIMEOUT, + wait_for_blob_operation( + "download", self.store .downloader(&self.endpoint) .download(blob_hash, Some(provider.id)), ) - .await - .map_err(|_| anyhow::anyhow!("blob download timed out"))??; + .await?; Ok(()) } pub async fn export_blob(&self, blob_hash: Hash, destination: &Path) -> anyhow::Result<()> { - tokio::time::timeout( - BLOB_OPERATION_TIMEOUT, - self.store.blobs().export(blob_hash, destination), - ) - .await - .map_err(|_| anyhow::anyhow!("blob export timed out"))??; + wait_for_blob_operation("export", self.store.blobs().export(blob_hash, destination)) + .await?; Ok(()) } @@ -441,6 +442,17 @@ impl NodeHost { } } +async fn wait_for_blob_operation(operation: &str, future: F) -> anyhow::Result +where + F: IntoFuture>, + E: Into, +{ + tokio::time::timeout(BLOB_OPERATION_TIMEOUT, future) + .await + .map_err(|_| anyhow::anyhow!("blob {operation} timed out"))? + .map_err(Into::into) +} + fn verify_connected_peer( connection: &Connection, peer_address: &EndpointAddr, diff --git a/src/service.rs b/src/service.rs index cde3237..ee7ea00 100644 --- a/src/service.rs +++ b/src/service.rs @@ -17,20 +17,20 @@ const RESTART_DELAY_SECONDS: u8 = 5; pub fn install() -> anyhow::Result { require_linux()?; - let unit_path = unit_path()?; + let unit_file_path = unit_path()?; let executable = env::current_exe()?; let app_paths = AppPaths::discover()?; let unit_contents = unit_contents(&executable, &app_paths.data_directory)?; - let unit_directory = unit_path + let unit_directory = unit_file_path .parent() .ok_or_else(|| anyhow::anyhow!("systemd unit path has no parent directory"))?; fs::create_dir_all(unit_directory)?; - fs::write(&unit_path, unit_contents)?; + fs::write(&unit_file_path, unit_contents)?; run_systemctl(&["daemon-reload"])?; run_systemctl(&["enable", "--now", UNIT_NAME])?; - Ok(unit_path) + Ok(unit_file_path) } pub fn status() -> anyhow::Result<()> { @@ -112,9 +112,12 @@ fn run_systemctl(arguments: &[&str]) -> anyhow::Result<()> { fn unit_contents(executable: &Path, data_directory: &Path) -> anyhow::Result { let executable = systemd_argument(executable)?; let data_directory = systemd_argument(data_directory)?; - Ok(format!( - "[Unit]\nDescription=Appa folder synchronization\nAfter=network-online.target\nWants=network-online.target\n\n[Service]\nType=simple\nEnvironment=APPA_HOME={data_directory}\nExecStart={executable} run\nRestart=on-failure\nRestartSec={RESTART_DELAY_SECONDS}\n\n[Install]\nWantedBy=default.target\n" - )) + let unit_header = "[Unit]\nDescription=Appa folder synchronization\nAfter=network-online.target\nWants=network-online.target"; + let service = format!( + "[Service]\nType=simple\nEnvironment=APPA_HOME={data_directory}\nExecStart={executable} run\nRestart=on-failure\nRestartSec={RESTART_DELAY_SECONDS}" + ); + let installation = "[Install]\nWantedBy=default.target"; + Ok(format!("{unit_header}\n\n{service}\n\n{installation}\n")) } fn systemd_argument(path: &Path) -> anyhow::Result {