use std::{
collections::HashMap,
path::{Path, PathBuf},
sync::{Arc, Mutex},
time::{Duration, Instant},
};
use iroh_docs::NamespaceId;
use tokio::{
sync::{broadcast, mpsc},
task::JoinHandle,
};
use crate::{
error::{Result, SyncwebError},
filter::{FilterAction, FilterEngine, FilterEntry},
folder::FolderManager,
fs::{FsWatcher, Importer},
node::{
identity::IdentityManager,
iroh_node::{IrohNode, RelayMode},
},
schedule::ScheduleManager,
stats::BandwidthStats,
storage::Config as AppConfig,
sync::{SubscribeParams, SyncEngine, cancel_session, is_active},
};
use super::{
DaemonHandle, DaemonState, DaemonStatus, FolderEntry, IpcServer, ManagedPool, PidLock, StateFile,
current_timestamp, daemon_socket_path,
state::{BandwidthSnapshot, DaemonStatusReport, FolderStatusReport, ScheduleStatus},
supervisor::{IntentControls, IntentSupervisor},
};
#[derive(Clone, Debug)]
#[non_exhaustive]
pub struct DaemonConfig {
pub data_dir: PathBuf,
pub sync_interval: Duration,
pub observation_ttl: Duration,
pub max_retries: u32,
pub backoff_base: Duration,
pub backoff_max: Duration,
pub rayon_threads: usize,
pub log_level: String,
pub log_file: Option<PathBuf>,
pub watch_debounce: Duration,
}
impl Default for DaemonConfig {
fn default() -> Self {
Self {
data_dir: PathBuf::from("."),
sync_interval: Duration::from_mins(1),
observation_ttl: Duration::from_hours(1),
max_retries: 3,
backoff_base: Duration::from_secs(1),
backoff_max: Duration::from_mins(1),
rayon_threads: std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get),
log_level: "info".to_owned(),
log_file: None,
watch_debounce: Duration::from_millis(500),
}
}
}
impl DaemonConfig {
#[must_use]
pub fn new(data_dir: impl Into<PathBuf>) -> Self {
Self {
data_dir: data_dir.into(),
..Self::default()
}
}
}
pub struct Daemon {
config: DaemonConfig,
state_file: StateFile,
pid_lock: PidLock,
ipc_server: IpcServer,
intent_supervisor: IntentSupervisor,
folder_manager: FolderManager,
sync_engine: SyncEngine,
schedule_manager: tokio::sync::RwLock<Option<ScheduleManager>>,
node: Arc<IrohNode>,
handle: DaemonHandle,
sync_receiver: tokio::sync::Mutex<mpsc::UnboundedReceiver<Option<String>>>,
intent_tasks: Mutex<HashMap<NamespaceId, JoinHandle<()>>>,
intent_controls: IntentControls,
watchers: Mutex<HashMap<String, FsWatcher>>,
pending_watch_events: Mutex<HashMap<String, PendingWatch>>,
filter_engine: tokio::sync::RwLock<Option<FilterEngine>>,
archive_pool: Arc<ManagedPool>,
}
struct PendingWatch {
paths: HashMap<PathBuf, bool>,
ready_at: Instant,
}
impl Daemon {
pub async fn new(config: DaemonConfig) -> Result<Self> {
std::fs::create_dir_all(&config.data_dir)?;
let state_file = StateFile::new(&config.data_dir);
let pid_lock = PidLock::new(&config.data_dir);
if !pid_lock.try_acquire()? {
return Err(SyncwebError::operation(
"daemon already running",
config.data_dir.display(),
));
}
let app_config = match AppConfig::load(config.data_dir.join("config.toml")) {
Ok(value) => value,
Err(error) => {
pid_lock.release()?;
return Err(error);
}
};
let schedule_manager = match ScheduleManager::from_config(&app_config.schedule) {
Ok(value) if app_config.schedule != crate::schedule::ScheduleConfig::default() => Some(value),
Ok(_) => None,
Err(error) => {
pid_lock.release()?;
return Err(error);
}
};
let filter_engine = match load_filter(&config.data_dir) {
Ok(value) => value,
Err(error) => {
pid_lock.release()?;
return Err(error);
}
};
let initial_state = DaemonState::new(
std::process::id(),
String::new(),
current_timestamp(),
&config.data_dir,
DaemonStatus::Starting,
);
if let Err(error) = state_file.save(&initial_state) {
pid_lock.release()?;
return Err(error);
}
let identity = match IdentityManager::new(config.data_dir.join("identity.key")) {
Ok(value) => value,
Err(error) => {
let _cleanup_state = state_file.remove();
let _cleanup_lock = pid_lock.release();
return Err(error);
}
};
let node = match IrohNode::new(identity, config.data_dir.join("data"), RelayMode::Default).await {
Ok(value) => Arc::new(value),
Err(error) => {
let _cleanup_state = state_file.remove();
let _cleanup_lock = pid_lock.release();
return Err(error);
}
};
let folder_manager = FolderManager::new(&node);
let sync_engine = SyncEngine::new(
folder_manager.clone(),
node.blob_store().clone(),
node.docs_engine().clone(),
node.gossip_service().clone(),
);
let archive_pool = match ManagedPool::new("syncweb-archive", config.rayon_threads) {
Ok(value) => Arc::new(value),
Err(error) => {
let _cleanup_node = node.stop().await;
let _cleanup_state = state_file.remove();
let _cleanup_lock = pid_lock.release();
return Err(SyncwebError::operation("failed to create daemon thread pool", error));
}
};
let initial_handle = DaemonHandle::new(initial_state);
{
let mut state = initial_handle.state.write().await;
state.node_id = node.endpoint().id().to_string();
state.status = DaemonStatus::Running;
let running_state = state.clone();
drop(state);
if let Err(error) = state_file.save(&running_state) {
let _cleanup_node = node.stop().await;
let _cleanup_state = state_file.remove();
let _cleanup_lock = pid_lock.release();
return Err(error);
}
}
let (sync_sender, sync_receiver) = mpsc::unbounded_channel();
let handle = DaemonHandle::with_channels_and_reload(
initial_handle.state.clone(),
initial_handle.folder_registry.clone(),
initial_handle.shutdown_sender.clone(),
sync_sender,
initial_handle.reload_requested.clone(),
);
let ipc_server = IpcServer::with_archive_context(
daemon_socket_path(&config.data_dir),
handle.clone(),
node.clone(),
archive_pool.clone(),
);
let intent_supervisor = IntentSupervisor::new(config.max_retries, config.backoff_base, config.backoff_max);
Ok(Self {
config,
state_file,
pid_lock,
ipc_server,
intent_supervisor,
folder_manager,
sync_engine,
schedule_manager: tokio::sync::RwLock::new(schedule_manager),
node,
handle,
sync_receiver: tokio::sync::Mutex::new(sync_receiver),
intent_tasks: Mutex::new(HashMap::new()),
intent_controls: Arc::new(Mutex::new(HashMap::new())),
watchers: Mutex::new(HashMap::new()),
pending_watch_events: Mutex::new(HashMap::new()),
filter_engine: tokio::sync::RwLock::new(filter_engine),
archive_pool,
})
}
pub async fn run(&self) -> Result<()> {
let run_result = self.run_inner().await;
let cleanup_result = self.shutdown_resources().await;
match run_result {
Err(error) => {
cleanup_result?;
Err(error)
}
Ok(()) => cleanup_result,
}
}
pub async fn state(&self) -> DaemonState {
self.handle.state.read().await.clone()
}
#[must_use]
pub fn archive_pool(&self) -> &ManagedPool {
self.archive_pool.as_ref()
}
async fn run_inner(&self) -> Result<()> {
tracing::debug!(
rayon_threads = self.archive_pool.thread_count(),
"daemon runtime initialized"
);
self.load_folders().await?;
self.start_watching().await?;
let server = self.ipc_server.clone();
let mut server_task = tokio::spawn(async move { server.serve().await });
let mut shutdown = self.handle.shutdown_sender.subscribe();
let shutdown_sender = self.handle.shutdown_sender.clone();
let mut signal_task = Box::pin(self.handle_signals(shutdown_sender.clone()));
let interval_duration = self.config.sync_interval.max(Duration::from_millis(1));
let mut interval = tokio::time::interval(interval_duration);
let mut watch_interval = tokio::time::interval(Duration::from_millis(100));
if let Err(error) = self.run_cycle().await {
tracing::error!(%error, "initial daemon cycle failed");
}
let result = self
.run_event_loop(
&mut server_task,
&mut signal_task,
&mut shutdown,
&shutdown_sender,
&mut interval,
&mut watch_interval,
)
.await;
send_shutdown(&shutdown_sender);
if !server_task.is_finished() {
match server_task.await {
Ok(server_result) => server_result?,
Err(error) => return Err(SyncwebError::operation("daemon IPC task failed", error)),
}
}
result
}
async fn run_event_loop(
&self,
server_task: &mut JoinHandle<Result<()>>,
signal_task: &mut std::pin::Pin<std::boxed::Box<impl std::future::Future<Output = Result<()>> + Send>>,
shutdown: &mut broadcast::Receiver<()>,
shutdown_sender: &broadcast::Sender<()>,
interval: &mut tokio::time::Interval,
watch_interval: &mut tokio::time::Interval,
) -> Result<()> {
loop {
tokio::select! {
signal_result = &mut *signal_task => {
signal_result?;
send_shutdown(shutdown_sender);
break Ok(());
}
shutdown_result = shutdown.recv() => {
if matches!(shutdown_result, Ok(()) | Err(broadcast::error::RecvError::Closed)) {
break Ok(());
}
}
server_result = &mut *server_task => {
match server_result {
Ok(result) => break result,
Err(error) => break Err(SyncwebError::operation("daemon IPC task failed", error)),
}
}
trigger = self.receive_sync_trigger() => {
match trigger {
Some(namespace) => self.run_trigger(namespace).await?,
None => break Ok(()),
}
}
_ = watch_interval.tick() => self.handle_watch_events().await?,
_ = interval.tick() => self.run_cycle().await?,
}
}
}
async fn run_cycle(&self) -> Result<()> {
self.reload_if_requested().await?;
self.load_folders().await?;
self.start_watching().await?;
let statuses = self.handle.folder_registry.read().await.statuses();
for folder in statuses {
let namespace = folder
.namespace
.parse::<NamespaceId>()
.map_err(|error| SyncwebError::operation("invalid managed folder namespace", error))?;
let folder_name = (!folder.path.as_os_str().is_empty()).then(|| folder.path.to_string_lossy().into_owned());
let active = {
let schedule_manager = self.schedule_manager.read().await;
schedule_manager
.as_ref()
.is_none_or(|manager| manager.is_active(folder_name.as_deref()))
};
self.start_supervision(namespace).await?;
if active {
self.set_intent_active(namespace, true)?;
} else {
self.set_intent_active(namespace, false)?;
}
}
self.save_status_report().await?;
Ok(())
}
async fn run_trigger(&self, namespace: Option<String>) -> Result<()> {
match namespace {
Some(value) => {
let parsed_namespace = value
.parse::<NamespaceId>()
.map_err(|error| SyncwebError::operation("invalid sync namespace", error))?;
self.start_supervision(parsed_namespace).await?;
self.set_intent_active(parsed_namespace, true)?;
self.save_status_report().await?;
Ok(())
}
None => self.run_cycle().await,
}
}
async fn start_supervision(&self, namespace: NamespaceId) -> Result<()> {
if is_active(namespace) {
return Ok(());
}
{
let mut tasks = self
.intent_tasks
.lock()
.map_err(|error| SyncwebError::operation("daemon intent task mutex is poisoned", error))?;
tasks.retain(|_, task| !task.is_finished());
if tasks.contains_key(&namespace) {
return Ok(());
}
}
let sync = self.sync_engine.clone();
let supervisor = self.intent_supervisor;
let shutdown = self.handle.shutdown_sender.subscribe();
let controls = self.intent_controls.clone();
let filter = self.filter_engine.read().await.clone();
let folder_name = self
.handle
.folder_registry
.read()
.await
.statuses()
.into_iter()
.find(|status| status.namespace == namespace.to_string())
.and_then(|status| {
(!status.path.as_os_str().is_empty()).then(|| status.path.to_string_lossy().into_owned())
});
let bandwidth = self
.schedule_manager
.read()
.await
.as_ref()
.map(|manager| manager.current_limits(folder_name.as_deref()));
let params = bandwidth.map_or_else(SubscribeParams::default, |limits| {
SubscribeParams::default().with_bandwidth_limits(limits)
});
let task = tokio::spawn(async move {
match supervisor
.supervise_with_controls(&sync, namespace, params, shutdown, controls, filter)
.await
{
Ok(result) => {
if let Some(error) = result.last_error {
tracing::warn!(%namespace, retry_count = result.retry_count, %error, "supervised intent stopped");
}
}
Err(error) => tracing::error!(%namespace, %error, "supervised intent failed"),
}
});
self.intent_tasks
.lock()
.map_err(|error| SyncwebError::operation("daemon intent task mutex is poisoned", error))?
.insert(namespace, task);
Ok(())
}
fn set_intent_active(&self, namespace: NamespaceId, active: bool) -> Result<()> {
let control = self
.intent_controls
.lock()
.map_err(|error| SyncwebError::operation("daemon intent control mutex is poisoned", error))?
.get(&namespace)
.cloned();
let Some(intent_control) = control else {
return Ok(());
};
let result = if active {
intent_control.resume()
} else {
intent_control.pause()
};
result.map_err(|error| SyncwebError::operation("failed to update scheduled intent", error))
}
pub async fn is_in_active_window(&self) -> bool {
self.schedule_manager
.read()
.await
.as_ref()
.is_none_or(|manager| manager.is_active(None))
}
pub async fn current_bandwidth_limit(&self) -> Option<u64> {
self.schedule_manager.read().await.as_ref().and_then(|manager| {
let limits = manager.current_limits(None);
limits.max_download.or(limits.max_upload)
})
}
async fn start_watching(&self) -> Result<()> {
let statuses = self.handle.folder_registry.read().await.statuses();
let wanted: HashMap<_, _> = statuses
.iter()
.filter(|status| !status.path.as_os_str().is_empty())
.map(|status| (status.namespace.clone(), status.path.clone()))
.collect();
let mut watchers = self
.watchers
.lock()
.map_err(|error| SyncwebError::operation("daemon watcher mutex is poisoned", error))?;
watchers.retain(|namespace, _| wanted.contains_key(namespace));
for (namespace, path) in wanted {
if watchers.contains_key(&namespace) {
continue;
}
if !path.exists() {
tracing::warn!(%namespace, path = %path.display(), "managed folder path does not exist; watcher deferred");
continue;
}
watchers.insert(namespace, FsWatcher::new(&path)?);
}
drop(watchers);
Ok(())
}
async fn handle_watch_events(&self) -> Result<()> {
let mut observed = Vec::new();
{
let mut watchers = self
.watchers
.lock()
.map_err(|error| SyncwebError::operation("daemon watcher mutex is poisoned", error))?;
for (namespace, watcher) in watchers.iter_mut() {
loop {
match watcher.try_recv() {
Ok(Some(event)) => observed.push((namespace.clone(), event)),
Ok(None) => break,
Err(error) => {
tracing::warn!(%namespace, %error, "filesystem watcher event channel failed");
break;
}
}
}
}
}
if !observed.is_empty() {
let mut pending = self
.pending_watch_events
.lock()
.map_err(|error| SyncwebError::operation("daemon watch queue mutex is poisoned", error))?;
let ready_at = Instant::now()
.checked_add(self.config.watch_debounce)
.unwrap_or_else(Instant::now);
for (namespace, event) in observed {
let removed = matches!(event.event.kind, notify::EventKind::Remove(_));
let entry = pending.entry(namespace).or_insert_with(|| PendingWatch {
paths: HashMap::new(),
ready_at,
});
entry.ready_at = ready_at;
for path in event.paths {
entry.paths.insert(path, removed);
}
}
drop(pending);
}
let ready = {
let mut pending = self
.pending_watch_events
.lock()
.map_err(|error| SyncwebError::operation("daemon watch queue mutex is poisoned", error))?;
let now = Instant::now();
pending
.iter_mut()
.filter_map(|(namespace, batch)| {
(batch.ready_at <= now).then(|| (namespace.clone(), std::mem::take(&mut batch.paths)))
})
.filter(|(_, paths)| !paths.is_empty())
.collect::<Vec<_>>()
};
if ready.is_empty() {
return Ok(());
}
let roots: HashMap<_, _> = self
.handle
.folder_registry
.read()
.await
.statuses()
.into_iter()
.map(|status| (status.namespace, status.path))
.collect();
for (namespace, paths) in ready {
let Some(root) = roots.get(&namespace) else {
continue;
};
let namespace_id = namespace
.parse::<NamespaceId>()
.map_err(|error| SyncwebError::operation("invalid watched folder namespace", error))?;
for (path, removed) in paths {
self.process_watch_event(namespace_id, root, &path, removed).await?;
}
}
self.save_status_report().await?;
Ok(())
}
async fn process_watch_event(&self, namespace: NamespaceId, root: &Path, path: &Path, removed: bool) -> Result<()> {
let relative = path.strip_prefix(root).unwrap_or(path);
if relative.as_os_str().is_empty() {
return Ok(());
}
let size = std::fs::metadata(path).map_or(0, |metadata| metadata.len());
let accepted = self.filter_engine.read().await.as_ref().is_none_or(|filter| {
filter.evaluate_for_folder(&namespace.to_string(), &FilterEntry::new(relative.to_path_buf(), size))
!= FilterAction::Reject
});
if !accepted {
return Ok(());
}
let folder = self.folder_manager.get(namespace).await?;
let result = if removed || !path.exists() {
folder
.delete_entry(relative.as_os_str().as_encoded_bytes())
.await
.map(|()| 1_u64)
} else if path.is_file() {
let importer = Importer::new(
self.node.blob_store().clone(),
self.node.docs_engine().clone(),
folder.doc().clone(),
folder.author(),
)
.with_root(root);
importer
.import_path(path)
.await
.map(|entries| u64::try_from(entries.len()).unwrap_or(u64::MAX))
} else {
Ok(0)
};
match result {
Ok(entries) if entries > 0 => {
self.handle
.folder_registry
.write()
.await
.record_import(namespace, entries, current_timestamp());
self.start_supervision(namespace).await?;
}
Ok(_) => {}
Err(error) => {
self.handle
.folder_registry
.write()
.await
.record_error(namespace, error.to_string());
tracing::warn!(%namespace, path = %path.display(), %error, "filesystem change will be retried");
if !removed && is_recoverable_watch_error(&error) {
let mut pending = self.pending_watch_events.lock().map_err(|poisoned| {
SyncwebError::operation("daemon watch queue mutex is poisoned", poisoned)
})?;
let ready_at = Instant::now()
.checked_add(self.config.watch_debounce)
.unwrap_or_else(Instant::now);
let entry = pending.entry(namespace.to_string()).or_insert_with(|| PendingWatch {
paths: HashMap::new(),
ready_at,
});
entry.ready_at = ready_at;
entry.paths.insert(path.to_path_buf(), false);
drop(pending);
}
}
}
Ok(())
}
async fn save_status_report(&self) -> Result<()> {
let state = self.handle.state.read().await.clone();
let statuses = self.handle.folder_registry.read().await.statuses();
let schedule = self.schedule_manager.read().await.clone();
let schedule_report = schedule.as_ref().map(|manager| {
let minute = current_minute();
let next = manager.next_active_window_start_at(None, minute);
let next_window_start = next.map(|next_minute| {
let offset = if next_minute >= minute {
next_minute.saturating_sub(minute)
} else {
1_440_u16.saturating_sub(minute).saturating_add(next_minute)
};
current_timestamp().saturating_add(u64::from(offset).saturating_mul(60))
});
ScheduleStatus {
in_active_window: manager.is_active(None),
next_window_start,
}
});
let bandwidth_stats = BandwidthStats::load(self.config.data_dir.join("stats.json"))?;
let report = DaemonStatusReport {
pid: state.pid,
node_id: state.node_id,
started_at: state.started_at,
uptime_seconds: current_timestamp().saturating_sub(state.started_at),
folders: statuses
.into_iter()
.map(|folder| FolderStatusReport {
namespace: folder.namespace,
path: folder.path,
session_active: folder.session_active,
last_sync_at: folder.last_sync_at,
entries_synced: folder.entries_synced,
errors: folder.errors,
})
.collect(),
bandwidth: BandwidthSnapshot {
upload_total: bandwidth_stats.total_upload,
download_total: bandwidth_stats.total_download,
upload_rate: 0,
download_rate: 0,
},
schedule: schedule_report,
rayon_threads: self.archive_pool.thread_count(),
};
self.state_file.save_status(&report)
}
async fn load_folders(&self) -> Result<()> {
let folders = self.folder_manager.list().await?;
let mut registry = self.handle.folder_registry.write().await;
for folder in folders {
let namespace = folder.namespace_id();
if !registry
.statuses()
.iter()
.any(|status| status.namespace == namespace.to_string())
{
registry.add(FolderEntry::new(namespace, PathBuf::new()))?;
}
}
drop(registry);
Ok(())
}
async fn reload_if_requested(&self) -> Result<()> {
if !self
.handle
.reload_requested
.swap(false, std::sync::atomic::Ordering::AcqRel)
{
return Ok(());
}
let app_config = AppConfig::load(self.config.data_dir.join("config.toml"))?;
let parsed_schedule = ScheduleManager::from_config(&app_config.schedule)?;
let schedule_manager =
(app_config.schedule != crate::schedule::ScheduleConfig::default()).then_some(parsed_schedule);
let filter = load_filter(&self.config.data_dir)?;
*self.schedule_manager.write().await = schedule_manager;
*self.filter_engine.write().await = filter;
let statuses = self.handle.folder_registry.read().await.statuses();
for status in statuses {
if let Ok(namespace) = status.namespace.parse::<NamespaceId>()
&& is_active(namespace)
&& !cancel_session(namespace)
{
tracing::warn!(%namespace, "reloaded intent did not accept cancellation");
}
}
tracing::info!("daemon configuration reloaded");
Ok(())
}
async fn receive_sync_trigger(&self) -> Option<Option<String>> {
self.sync_receiver.lock().await.recv().await
}
async fn shutdown_resources(&self) -> Result<()> {
self.handle.set_status(DaemonStatus::Stopping).await;
let stopping_state = self.handle.state.read().await.clone();
self.state_file.save(&stopping_state)?;
self.save_status_report().await?;
let namespaces: Vec<_> = self
.handle
.folder_registry
.read()
.await
.statuses()
.into_iter()
.filter_map(|status| status.namespace.parse::<NamespaceId>().ok())
.collect();
for namespace in namespaces {
if is_active(namespace) && !cancel_session(namespace) {
tracing::warn!(%namespace, "intent did not accept shutdown cancellation");
}
}
let tasks = {
let mut task_map = self
.intent_tasks
.lock()
.map_err(|error| SyncwebError::operation("daemon intent task mutex is poisoned", error))?;
std::mem::take(&mut *task_map)
};
for (_, task) in tasks {
task.await
.map_err(|error| SyncwebError::operation("daemon intent task failed", error))?;
}
let node_result = self.node.stop().await;
self.handle.set_status(DaemonStatus::Stopped).await;
let remove_result = self.state_file.remove();
let remove_status_result = self.state_file.remove_status();
let release_result = self.pid_lock.release();
node_result?;
remove_result?;
remove_status_result?;
release_result?;
Ok(())
}
async fn handle_signals(&self, shutdown: broadcast::Sender<()>) -> Result<()> {
#[cfg(unix)]
{
let mut terminate = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
tokio::select! {
result = tokio::signal::ctrl_c() => result?,
_ = terminate.recv() => {}
}
}
#[cfg(not(unix))]
{
tokio::signal::ctrl_c().await?;
}
send_shutdown(&shutdown);
Ok(())
}
}
impl Drop for Daemon {
fn drop(&mut self) {
if self.state_file.exists()
&& let Err(error) = self.state_file.remove()
{
tracing::warn!(path = %self.state_file.path().display(), %error, "failed to remove daemon state");
}
if let Err(error) = self.state_file.remove_status() {
tracing::warn!(path = %self.state_file.status_path().display(), %error, "failed to remove daemon status");
}
}
}
fn send_shutdown(sender: &broadcast::Sender<()>) {
match sender.send(()) {
Ok(_) | Err(broadcast::error::SendError(())) => {}
}
}
fn load_filter(data_dir: &Path) -> Result<Option<FilterEngine>> {
let path = data_dir.join("filters.toml");
if path.exists() {
FilterEngine::load(path).map(Some)
} else {
Ok(None)
}
}
fn current_minute() -> u16 {
let seconds = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |duration| duration.as_secs());
u16::try_from((seconds % 86_400).div_euclid(60)).unwrap_or(0)
}
fn is_recoverable_watch_error(error: &SyncwebError) -> bool {
let message = error.to_string();
message.contains("file changed during import") || message.contains("input path does not exist")
}