From 1dce6380fe561580e1bea7ec85de9e7f01f40bca Mon Sep 17 00:00:00 2001 From: "Ericson \"Fogo\" Soares" Date: Wed, 28 Jun 2023 13:30:19 -0300 Subject: [PATCH] Job System and Watcher improvements (#1030) * Trying to make watcher more resilient between creates and updates * More checks on create vs update * Rust fmt * PCR rev * Windows CI fix * Some indexer improvements * Builder pattern for jobs and reports * fix list view invalid date * fix progress bar * Normalizing job metadata output * fixed job metadata * Removing inner report struct from identifier job --------- Co-authored-by: Jamie Pine <32987599+jamiepine@users.noreply.github.com> Co-authored-by: Utku <74243531+utkubakir@users.noreply.github.com> Co-authored-by: Jamie Pine --- core/src/job/mod.rs | 149 ++++++++------- core/src/job/report.rs | 74 ++++++-- core/src/job/worker.rs | 11 +- .../isolated_file_path_data.rs | 10 +- core/src/location/file_path_helper/mod.rs | 22 ++- core/src/location/indexer/indexer_job.rs | 9 +- core/src/location/indexer/mod.rs | 115 ++++++++---- core/src/location/indexer/shallow.rs | 2 +- core/src/location/indexer/walk.rs | 137 ++++++-------- core/src/location/manager/watcher/macos.rs | 112 ++++++++---- core/src/location/manager/watcher/utils.rs | 173 +++++++++++++++--- core/src/location/mod.rs | 41 +++-- .../file_identifier/file_identifier_job.rs | 57 +++--- core/src/object/file_identifier/mod.rs | 19 +- core/src/object/file_identifier/shallow.rs | 11 +- core/src/object/fs/copy.rs | 3 +- core/src/object/fs/cut.rs | 3 +- core/src/object/fs/delete.rs | 3 +- .../src/object/preview/thumbnail/directory.rs | 10 +- core/src/object/preview/thumbnail/mod.rs | 8 +- core/src/object/preview/thumbnail/shallow.rs | 8 +- .../preview/thumbnail/thumbnailer_job.rs | 11 +- core/src/object/validation/validator_job.rs | 2 +- .../app/$libraryId/Explorer/View/ListView.tsx | 1 + .../Layout/Sidebar/JobManager/Job.tsx | 19 +- .../Layout/Sidebar/JobManager/JobGroup.tsx | 8 +- .../Layout/Sidebar/JobManager/useJobInfo.tsx | 91 +++++---- 27 files changed, 674 insertions(+), 435 deletions(-) diff --git a/core/src/job/mod.rs b/core/src/job/mod.rs index 104d72e85..d0b921f4a 100644 --- a/core/src/job/mod.rs +++ b/core/src/job/mod.rs @@ -11,7 +11,7 @@ use std::{ use serde::{de::DeserializeOwned, Deserialize, Serialize}; use tokio::{select, sync::mpsc}; -use tracing::{debug, info, warn}; +use tracing::{debug, info, trace, warn}; use uuid::Uuid; mod error; @@ -116,6 +116,54 @@ pub trait DynJob: Send + Sync { async fn cancel_children(&mut self, library: &Library) -> Result<(), JobError>; } +pub struct JobBuilder { + id: Uuid, + init: SJob, + report_builder: JobReportBuilder, +} + +impl JobBuilder { + pub fn build(self) -> Box> { + Box::new(Job:: { + id: self.id, + hash: ::hash(&self.init), + report: Some(self.report_builder.build()), + state: Some(JobState { + init: self.init, + data: None, + steps: VecDeque::new(), + step_number: 0, + run_metadata: Default::default(), + }), + next_jobs: VecDeque::new(), + }) + } + + pub fn new(init: SJob) -> Self { + let id = Uuid::new_v4(); + Self { + id, + init, + report_builder: JobReportBuilder::new(id, SJob::NAME.to_string()), + } + } + + pub fn with_action(mut self, action: impl AsRef) -> Self { + self.report_builder = self.report_builder.with_action(action); + self + } + + pub fn with_parent_id(mut self, parent_id: Uuid) -> Self { + self.report_builder = self.report_builder.with_parent_id(parent_id); + self + } + + pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self { + self.report_builder = self.report_builder.with_metadata(metadata); + self + } +} + pub struct Job { id: Uuid, hash: u64, @@ -127,41 +175,7 @@ pub struct Job { impl Job { pub fn new(init: SJob) -> Box { - let id = Uuid::new_v4(); - Box::new(Self { - id, - hash: ::hash(&init), - report: Some(JobReport::new(id, SJob::NAME.to_string())), - state: Some(JobState { - init, - data: None, - steps: VecDeque::new(), - step_number: 0, - run_metadata: Default::default(), - }), - next_jobs: VecDeque::new(), - }) - } - - pub fn new_with_action(init: SJob, action: impl AsRef) -> Box { - let id = Uuid::new_v4(); - Box::new(Self { - id, - hash: ::hash(&init), - report: Some(JobReport::new_with_action( - id, - SJob::NAME.to_string(), - action, - )), - state: Some(JobState { - init, - data: None, - steps: VecDeque::new(), - step_number: 0, - run_metadata: Default::default(), - }), - next_jobs: VecDeque::new(), - }) + JobBuilder::new(init).build() } pub fn queue_next(mut self: Box, init: NextSJob) -> Box @@ -169,17 +183,17 @@ impl Job { NextSJob: StatefulJob + 'static, { let next_job_order = self.next_jobs.len() + 1; - self.next_jobs.push_back(Job::new_dependent( - init, - self.id, - // SAFETY: If we're queueing a next job then we should still have a report - self.report().as_ref().and_then(|parent_report| { - parent_report - .action - .as_ref() - .map(|parent_action| format!("{parent_action}-{next_job_order}")) - }), - )); + + let mut child_job_builder = JobBuilder::new(init).with_parent_id(self.id); + + if let Some(parent_report) = self.report() { + if let Some(parent_action) = &parent_report.action { + child_job_builder = + child_job_builder.with_action(format!("{parent_action}-{next_job_order}")); + } + } + + self.next_jobs.push_back(child_job_builder.build()); self } @@ -205,28 +219,6 @@ impl Job { })) } - fn new_dependent(init: SJob, parent_id: Uuid, parent_action: Option) -> Box { - let id = Uuid::new_v4(); - Box::new(Self { - id, - hash: ::hash(&init), - report: Some(JobReport::new_with_parent( - id, - SJob::NAME.to_string(), - parent_id, - parent_action, - )), - state: Some(JobState { - init, - data: None, - steps: VecDeque::new(), - step_number: 0, - run_metadata: Default::default(), - }), - next_jobs: VecDeque::new(), - }) - } - pub async fn spawn(self, library: &Library) -> Result<(), JobManagerError> { library .node_context @@ -613,7 +605,7 @@ impl DynJob for Job { let inner_step = Arc::clone(&step_arc); let inner_stateful_job = Arc::clone(&stateful_job); - let _step_time = Instant::now(); + let step_time = Instant::now(); let mut job_step_handle = tokio::spawn(async move { inner_stateful_job @@ -658,6 +650,7 @@ impl DynJob for Job { // The job can also be shutdown or canceled while paused WorkerCommand::Shutdown(when, signal_tx) => { job_step_handle.abort(); + let _ = job_step_handle.await; debug!( "Shuting down Job took {:?} \ @@ -698,6 +691,7 @@ impl DynJob for Job { } WorkerCommand::Cancel(when, signal_tx) => { job_step_handle.abort(); + let _ = job_step_handle.await; debug!( "Canceling Job \ took {:?} after running for {:?}", @@ -722,6 +716,7 @@ impl DynJob for Job { WorkerCommand::Shutdown(when, signal_tx) => { job_step_handle.abort(); + let _ = job_step_handle.await; debug!( "Shuting down Job took {:?} \ @@ -758,6 +753,7 @@ impl DynJob for Job { } WorkerCommand::Cancel(when, signal_tx) => { job_step_handle.abort(); + let _ = job_step_handle.await; debug!( "Canceling Job took {:?} \ after running for {:?}", @@ -771,10 +767,10 @@ impl DynJob for Job { // Here we actually run the job, step by step step_result = &mut job_step_handle => { - // debug!( - // "Step finished in {:?} Job ", - // _step_time.elapsed(), - // ); + trace!( + "Step finished in {:?} Job ", + step_time.elapsed(), + ); run_metadata = Arc::try_unwrap(run_metadata_arc) .expect("step already ran, no more refs"); @@ -807,9 +803,10 @@ impl DynJob for Job { if !new_errors.is_empty() { warn!("Job had a step with errors"); - for err in &new_errors { + new_errors.iter().for_each(|err| { warn!("Job error: {:?}", err); - } + }); + errors.extend(new_errors); } } diff --git a/core/src/job/report.rs b/core/src/job/report.rs index 4a924ea81..c6dd7ce4e 100644 --- a/core/src/job/report.rs +++ b/core/src/job/report.rs @@ -174,24 +174,6 @@ impl JobReport { } } - pub fn new_with_action(uuid: Uuid, name: String, action: impl AsRef) -> Self { - let mut report = Self::new(uuid, name); - report.action = Some(action.as_ref().to_string()); - report - } - - pub fn new_with_parent( - uuid: Uuid, - name: String, - parent_id: Uuid, - action: Option, - ) -> Self { - let mut report = Self::new(uuid, name); - report.parent_id = Some(parent_id); - report.action = action; - report - } - pub fn get_meta(&self) -> (String, Option) { // actions are formatted like "added_location" or "added_location-1" let Some(action_name) = self.action @@ -313,3 +295,59 @@ impl TryFrom for JobStatus { Ok(s) } } + +pub struct JobReportBuilder { + pub id: Uuid, + pub name: String, + pub action: Option, + pub metadata: Option, + pub parent_id: Option, +} + +impl JobReportBuilder { + pub fn build(self) -> JobReport { + JobReport { + id: self.id, + is_background: false, // deprecated + name: self.name, + action: self.action, + created_at: None, + started_at: None, + completed_at: None, + status: JobStatus::Queued, + errors_text: vec![], + task_count: 0, + data: None, + metadata: self.metadata, + parent_id: self.parent_id, + completed_task_count: 0, + message: String::new(), + estimated_completion: Utc::now(), + } + } + + pub fn new(id: Uuid, name: String) -> Self { + Self { + id, + name, + action: None, + metadata: None, + parent_id: None, + } + } + + pub fn with_action(mut self, action: impl AsRef) -> Self { + self.action = Some(action.as_ref().to_string()); + self + } + + pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self { + self.metadata = Some(metadata); + self + } + + pub fn with_parent_id(mut self, parent_id: Uuid) -> Self { + self.parent_id = Some(parent_id); + self + } +} diff --git a/core/src/job/worker.rs b/core/src/job/worker.rs index c69e77133..54d3e06f8 100644 --- a/core/src/job/worker.rs +++ b/core/src/job/worker.rs @@ -11,6 +11,7 @@ use std::{ use chrono::{DateTime, Utc}; use serde::Serialize; +use serde_json::json; use specta::Type; use tokio::{ select, @@ -384,7 +385,15 @@ impl Worker { }) if errors.is_empty() => { report.status = JobStatus::Completed; report.data = None; - report.metadata = metadata; + report.metadata = match (report.metadata.take(), metadata) { + (Some(mut current_metadata), Some(new_metadata)) => { + current_metadata["output"] = new_metadata; + Some(current_metadata) + } + (None, Some(new_metadata)) => Some(json!({ "output": new_metadata })), + (Some(current_metadata), None) => Some(current_metadata), + _ => None, + }; report.completed_at = Some(Utc::now()); if let Err(e) = report.update(library).await { error!("failed to update job report: {:#?}", e); diff --git a/core/src/location/file_path_helper/isolated_file_path_data.rs b/core/src/location/file_path_helper/isolated_file_path_data.rs index a85fcf489..0ea38a2e1 100644 --- a/core/src/location/file_path_helper/isolated_file_path_data.rs +++ b/core/src/location/file_path_helper/isolated_file_path_data.rs @@ -24,11 +24,15 @@ static FORBIDDEN_FILE_NAMES: OnceLock = OnceLock::new(); #[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq)] #[non_exhaustive] pub struct IsolatedFilePathData<'a> { + // WARN! These fields MUST NOT be changed outside the location module, that's why they have this visibility + // and are not public. They have some specific logic on them and should not be writen to directly. + // If you wanna access one of them outside from location module, write yourself an accessor method + // to have read only access to them. pub(in crate::location) location_id: location::id::Type, - pub materialized_path: Cow<'a, str>, + pub(in crate::location) materialized_path: Cow<'a, str>, pub(in crate::location) is_dir: bool, - pub name: Cow<'a, str>, - pub extension: Cow<'a, str>, + pub(in crate::location) name: Cow<'a, str>, + pub(in crate::location) extension: Cow<'a, str>, relative_path: Cow<'a, str>, } diff --git a/core/src/location/file_path_helper/mod.rs b/core/src/location/file_path_helper/mod.rs index 33d539919..a59b420ae 100644 --- a/core/src/location/file_path_helper/mod.rs +++ b/core/src/location/file_path_helper/mod.rs @@ -5,6 +5,7 @@ use crate::{ use std::{ fs::Metadata, + hash::{Hash, Hasher}, path::{Path, PathBuf, MAIN_SEPARATOR_STR}, time::SystemTime, }; @@ -23,6 +24,7 @@ pub use isolated_file_path_data::{ }; // File Path selectables! +file_path::select!(file_path_just_pub_id { pub_id }); file_path::select!(file_path_just_pub_id_materialized_path { pub_id materialized_path @@ -36,12 +38,6 @@ file_path::select!(file_path_for_file_identifier { name extension }); -file_path::select!(file_path_for_indexer { - pub_id - materialized_path - name - extension -}); file_path::select!(file_path_for_object_validator { pub_id materialized_path @@ -97,6 +93,20 @@ file_path::select!(file_path_to_full_path { // File Path includes! file_path::include!(file_path_with_object { object }); +impl Hash for file_path_just_pub_id::Data { + fn hash(&self, state: &mut H) { + self.pub_id.hash(state); + } +} + +impl PartialEq for file_path_just_pub_id::Data { + fn eq(&self, other: &Self) -> bool { + self.pub_id == other.pub_id + } +} + +impl Eq for file_path_just_pub_id::Data {} + #[derive(Clone, Copy, Debug, Serialize, Deserialize)] pub struct FilePathMetadata { pub inode: u64, diff --git a/core/src/location/indexer/indexer_job.rs b/core/src/location/indexer/indexer_job.rs index f4cd6f427..30717e47d 100644 --- a/core/src/location/indexer/indexer_job.rs +++ b/core/src/location/indexer/indexer_job.rs @@ -24,6 +24,7 @@ use std::{ use itertools::Itertools; use serde::{Deserialize, Serialize}; +use serde_json::json; use tokio::time::Instant; use tracing::info; @@ -179,7 +180,7 @@ impl StatefulJob for IndexerJobInit { &indexer_rules, update_notifier_fn(ctx), file_paths_db_fetcher_fn!(&db), - to_remove_db_fetcher_fn!(location_id, location_path, &db), + to_remove_db_fetcher_fn!(location_id, &db), iso_file_path_factory(location_id, location_path), 50_000, ) @@ -298,7 +299,7 @@ impl StatefulJob for IndexerJobInit { &data.indexer_rules, update_notifier_fn(ctx), file_paths_db_fetcher_fn!(&db), - to_remove_db_fetcher_fn!(location_id, location_path, &db), + to_remove_db_fetcher_fn!(location_id, &db), iso_file_path_factory(location_id, location_path), ) .await?; @@ -361,7 +362,7 @@ impl StatefulJob for IndexerJobInit { ) -> JobResult { let init = self; info!( - "scan of {} completed in {:?}. {} new files found, \ + "Scan of {} completed in {:?}. {} new files found, \ indexed {} files in db. db write completed in {:?}", maybe_missing(&init.location.path, "location.path")?, run_metadata.scan_read_time, @@ -374,7 +375,7 @@ impl StatefulJob for IndexerJobInit { invalidate_query!(ctx.library, "search.paths"); } - Ok(Some(serde_json::to_value(init)?)) + Ok(Some(json!({"init: ": init, "run_metadata": run_metadata}))) } } diff --git a/core/src/location/indexer/mod.rs b/core/src/location/indexer/mod.rs index 342bead26..588b0ba46 100644 --- a/core/src/location/indexer/mod.rs +++ b/core/src/location/indexer/mod.rs @@ -13,10 +13,10 @@ use sd_prisma::prisma_sync; use serde::{Deserialize, Serialize}; use serde_json::json; use thiserror::Error; -use tracing::info; +use tracing::trace; use super::{ - file_path_helper::{file_path_for_indexer, FilePathError, IsolatedFilePathData}, + file_path_helper::{file_path_just_pub_id, FilePathError, IsolatedFilePathData}, location_with_indexer_rules, }; @@ -171,7 +171,7 @@ async fn execute_indexer_save_step( ) .await?; - info!("Inserted {count} records"); + trace!("Inserted {count} records"); Ok(count) } @@ -186,7 +186,7 @@ fn iso_file_path_factory( } async fn remove_non_existing_file_paths( - to_remove: impl IntoIterator, + to_remove: impl IntoIterator, db: &PrismaClient, ) -> Result { db.file_path() @@ -206,11 +206,24 @@ async fn remove_non_existing_file_paths( macro_rules! file_paths_db_fetcher_fn { ($db:expr) => {{ |found_paths| async { - $db.file_path() - .find_many(found_paths) - .select($crate::location::file_path_helper::file_path_to_isolate::select()) - .exec() + // Each found path is a AND with 4 terms, and SQLite has a expression tree limit of 1000 terms + // so we will use chunks of 200 just to be safe + + // FIXME: Can't pass this chunks variable direct to _batch because of lifetime issues + let chunks = found_paths + .into_iter() + .chunks(200) + .into_iter() + .map(|founds| { + $db.file_path() + .find_many(founds.collect::>()) + .select($crate::location::file_path_helper::file_path_to_isolate::select()) + }) + .collect::>(); + + $db._batch(chunks) .await + .map(|fetched| fetched.into_iter().flatten().collect::>()) .map_err(Into::into) } }}; @@ -222,40 +235,70 @@ macro_rules! file_paths_db_fetcher_fn { // FIXME: (fogodev) I was receiving this error here https://github.com/rust-lang/rust/issues/74497 #[macro_export] macro_rules! to_remove_db_fetcher_fn { - ($location_id:expr, $location_path:expr, $db:expr) => {{ - |iso_file_path| async { + ($location_id:expr, $db:expr) => {{ + |iso_file_path, unique_location_id_materialized_path_name_extension_params| async { let iso_file_path: $crate::location::file_path_helper::IsolatedFilePathData<'static> = iso_file_path; - let mut data = Vec::new(); // one stupid large vec - loop { - let r = $db - .file_path() - .find_many(vec![ - $crate::prisma::file_path::location_id::equals(Some($location_id)), - $crate::prisma::file_path::materialized_path::equals(Some( - iso_file_path - .materialized_path_for_children() - .expect("the received isolated file path must be from a directory"), - )), - ]) - .take(100) - .select($crate::location::file_path_helper::file_path_for_indexer::select()) - .exec() - .await; + // FIXME: Can't pass this chunks variable direct to _batch because of lifetime issues + let chunks = unique_location_id_materialized_path_name_extension_params + .into_iter() + .chunks(200) + .into_iter() + .map(|unique_params| { + $db.file_path() + .find_many(vec![ + $crate::prisma::file_path::location_id::equals(Some($location_id)), + $crate::prisma::file_path::materialized_path::equals(Some( + iso_file_path.materialized_path_for_children().expect( + "the received isolated file path must be from a directory", + ), + )), + ::prisma_client_rust::operator::not(vec![ + ::prisma_client_rust::operator::or(unique_params.collect()), + ]), + ]) + .select($crate::location::file_path_helper::file_path_just_pub_id::select()) + }) + .collect::<::std::vec::Vec<_>>(); - match r { - Ok(mut v) => { - data.append(&mut v); - if v.len() != 100 { - break Ok(data); + $db._batch(chunks) + .await + .map(|to_remove| { + // This is an intersection between all sets + let mut sets = to_remove + .into_iter() + .map(|fetched_vec| { + fetched_vec + .into_iter() + .map(|fetched| { + ::uuid::Uuid::from_slice(&fetched.pub_id) + .expect("file_path.pub_id is invalid!") + }) + .collect::<::std::collections::HashSet<_>>() + }) + .collect::>(); + + let mut intersection = ::std::collections::HashSet::new(); + while let Some(set) = sets.pop() { + for pub_id in set { + // Remove returns true if the element was present in the set + if sets.iter_mut().all(|set| set.remove(&pub_id)) { + intersection.insert(pub_id); + } } } - Err(err) => { - break Err(err.into()); - } - } - } + + intersection + .into_iter() + .map(|pub_id| { + $crate::location::file_path_helper::file_path_just_pub_id::Data { + pub_id: pub_id.as_bytes().to_vec(), + } + }) + .collect() + }) + .map_err(::std::convert::Into::into) } }}; } diff --git a/core/src/location/indexer/shallow.rs b/core/src/location/indexer/shallow.rs index 89fc76c3b..df3e86537 100644 --- a/core/src/location/indexer/shallow.rs +++ b/core/src/location/indexer/shallow.rs @@ -72,7 +72,7 @@ pub async fn shallow( &indexer_rules, |_, _| {}, file_paths_db_fetcher_fn!(&db), - to_remove_db_fetcher_fn!(location_id, location_path, &db), + to_remove_db_fetcher_fn!(location_id, &db), iso_file_path_factory(location_id, &location_path), add_root, ) diff --git a/core/src/location/indexer/walk.rs b/core/src/location/indexer/walk.rs index 21865723c..753cb73a6 100644 --- a/core/src/location/indexer/walk.rs +++ b/core/src/location/indexer/walk.rs @@ -1,6 +1,6 @@ use crate::{ location::file_path_helper::{ - file_path_for_indexer, file_path_to_isolate, FilePathMetadata, IsolatedFilePathData, + file_path_just_pub_id, file_path_to_isolate, FilePathMetadata, IsolatedFilePathData, MetadataExt, }, prisma::file_path, @@ -14,7 +14,7 @@ use crate::location::file_path_helper::get_inode_and_device; use crate::location::file_path_helper::get_inode_and_device_from_path; use std::{ - collections::{HashMap, HashSet, VecDeque}, + collections::{HashSet, VecDeque}, future::Future, hash::{Hash, Hasher}, path::{Path, PathBuf}, @@ -71,7 +71,7 @@ impl Hash for WalkingEntry { pub struct WalkResult where Walked: Iterator, - ToRemove: Iterator, + ToRemove: Iterator, { pub walked: Walked, pub to_walk: VecDeque, @@ -87,19 +87,22 @@ pub(super) async fn walk( indexer_rules: &[IndexerRule], mut update_notifier: impl FnMut(&Path, usize), file_paths_db_fetcher: impl Fn(Vec) -> FilePathDBFetcherFut, - to_remove_db_fetcher: impl Fn(IsolatedFilePathData<'static>) -> ToRemoveDbFetcherFut, + to_remove_db_fetcher: impl Fn( + IsolatedFilePathData<'static>, + Vec, + ) -> ToRemoveDbFetcherFut, iso_file_path_factory: impl Fn(&Path, bool) -> Result, IndexerError>, limit: u64, ) -> Result< WalkResult< impl Iterator, - impl Iterator, + impl Iterator, >, IndexerError, > where FilePathDBFetcherFut: Future, IndexerError>>, - ToRemoveDbFetcherFut: Future, IndexerError>>, + ToRemoveDbFetcherFut: Future, IndexerError>>, { let root = root.as_ref(); @@ -149,18 +152,21 @@ pub(super) async fn keep_walking( indexer_rules: &[IndexerRule], mut update_notifier: impl FnMut(&Path, usize), file_paths_db_fetcher: impl Fn(Vec) -> FilePathDBFetcherFut, - to_remove_db_fetcher: impl Fn(IsolatedFilePathData<'static>) -> ToRemoveDbFetcherFut, + to_remove_db_fetcher: impl Fn( + IsolatedFilePathData<'static>, + Vec, + ) -> ToRemoveDbFetcherFut, iso_file_path_factory: impl Fn(&Path, bool) -> Result, IndexerError>, ) -> Result< WalkResult< impl Iterator, - impl Iterator, + impl Iterator, >, IndexerError, > where FilePathDBFetcherFut: Future, IndexerError>>, - ToRemoveDbFetcherFut: Future, IndexerError>>, + ToRemoveDbFetcherFut: Future, IndexerError>>, { let mut to_keep_walking = VecDeque::with_capacity(TO_WALK_QUEUE_INITIAL_CAPACITY); let mut indexed_paths = HashSet::with_capacity(WALK_SINGLE_DIR_PATHS_BUFFER_INITIAL_CAPACITY); @@ -196,20 +202,23 @@ pub(super) async fn walk_single_dir( indexer_rules: &[IndexerRule], mut update_notifier: impl FnMut(&Path, usize) + '_, file_paths_db_fetcher: impl Fn(Vec) -> FilePathDBFetcherFut, - to_remove_db_fetcher: impl Fn(IsolatedFilePathData<'static>) -> ToRemoveDbFetcherFut, + to_remove_db_fetcher: impl Fn( + IsolatedFilePathData<'static>, + Vec, + ) -> ToRemoveDbFetcherFut, iso_file_path_factory: impl Fn(&Path, bool) -> Result, IndexerError>, add_root: bool, ) -> Result< ( impl Iterator, - Vec, + Vec, Vec, ), IndexerError, > where FilePathDBFetcherFut: Future, IndexerError>>, - ToRemoveDbFetcherFut: Future, IndexerError>>, + ToRemoveDbFetcherFut: Future, IndexerError>>, { let root = root.as_ref(); @@ -281,8 +290,6 @@ where F: Future, IndexerError>>, { if !indexed_paths.is_empty() { - // TODO: Converting paths like this into PCR params will break if we pass in too many paths. - // TODO: I am not hitting errors here on my system so gonna leave this for now. file_paths_db_fetcher( indexed_paths .iter() @@ -327,7 +334,10 @@ async fn inner_walk_single_dir( }: &ToWalkEntry, indexer_rules: &[IndexerRule], update_notifier: &mut impl FnMut(&Path, usize), - to_remove_db_fetcher: &impl Fn(IsolatedFilePathData<'static>) -> ToRemoveDbFetcherFut, + to_remove_db_fetcher: impl Fn( + IsolatedFilePathData<'static>, + Vec, + ) -> ToRemoveDbFetcherFut, iso_file_path_factory: &impl Fn(&Path, bool) -> Result, IndexerError>, WorkingTable { indexed_paths, @@ -335,9 +345,9 @@ async fn inner_walk_single_dir( mut maybe_to_walk, errors, }: WorkingTable<'_>, -) -> Vec +) -> Vec where - ToRemoveDbFetcherFut: Future, IndexerError>>, + ToRemoveDbFetcherFut: Future, IndexerError>>, { let Ok(iso_file_path_to_walk) = iso_file_path_factory(path, true).map_err(|e| errors.push(e)) else { @@ -356,7 +366,6 @@ where paths_buffer.clear(); let mut found_paths_counts = 0; - let mut to_remove = HashMap::new(); // Marking with a loop label here in case of rejection or errors, to continue with next entry 'entries: loop { @@ -496,51 +505,22 @@ where } if accept_by_children_dir.unwrap_or(true) { - // Fetch all files then as we index we remove them and anything left over must have been deleted from FS and can be removed from DB - match to_remove_db_fetcher(iso_file_path_to_walk.clone()).await { - Ok(v) => { - for e in v { - to_remove.insert( - ( - e.materialized_path - .clone() - .expect("materialized_path should be set"), - e.name.clone().expect("name should be set"), - e.extension.clone().expect("extension should be set"), - ), - e, - ); - } - } - Err(e) => { - errors.push(e); - } - }; - let Ok(iso_file_path) = iso_file_path_factory(¤t_path, is_dir) .map_err(|e| errors.push(e)) else { continue 'entries; }; - { - // Remove file path as we don't want it to be removed from the DB - to_remove.remove(&( - iso_file_path.materialized_path.to_string(), - iso_file_path.name.to_string(), - iso_file_path.extension.to_string(), - )); - paths_buffer.push(WalkingEntry { - iso_file_path, - maybe_metadata: Some(FilePathMetadata { - inode, - device, - size_in_bytes: metadata.len(), - created_at: metadata.created_or_now().into(), - modified_at: metadata.modified_or_now().into(), - }), - }); - } + paths_buffer.push(WalkingEntry { + iso_file_path, + maybe_metadata: Some(FilePathMetadata { + inode, + device, + size_in_bytes: metadata.len(), + created_at: metadata.created_or_now().into(), + modified_at: metadata.modified_or_now().into(), + }), + }); // If the ancestors directories wasn't indexed before, now we do for ancestor in current_path @@ -591,21 +571,7 @@ where modified_at: metadata.modified_or_now().into(), }); - { - // Remove file path as we don't want it to be removed from the DB - to_remove.remove(&( - ancestor_iso_walking_entry - .iso_file_path - .materialized_path - .to_string(), - ancestor_iso_walking_entry.iso_file_path.name.to_string(), - ancestor_iso_walking_entry - .iso_file_path - .extension - .to_string(), - )); - paths_buffer.push(ancestor_iso_walking_entry); - } + paths_buffer.push(ancestor_iso_walking_entry); } else { // If indexed_paths contains the current ancestors, then it will contain // also all if its ancestors too, so we can stop here @@ -615,11 +581,28 @@ where } } + // We continue the function even if we fail to fetch `file_path`s to remove, + // the DB will have old `file_path`s but at least this is better than + // don't adding the newly indexed paths + let to_remove = to_remove_db_fetcher( + iso_file_path_to_walk, + paths_buffer + .iter() + .map(|entry| &entry.iso_file_path) + .map(Into::into) + .collect(), + ) + .await + .unwrap_or_else(|e| { + errors.push(e); + vec![] + }); + // Just merging the `found_paths` with `indexed_paths` here in the end to avoid possibly // multiple rehashes during function execution indexed_paths.extend(paths_buffer.drain(..)); - to_remove.into_values().collect() + to_remove } #[cfg(test)] @@ -753,7 +736,7 @@ mod tests { &[], |_, _| {}, |_| async { Ok(vec![]) }, - |_| async { Ok(vec![]) }, + |_, _| async { Ok(vec![]) }, |path, is_dir| { IsolatedFilePathData::new(0, root_path, path, is_dir).map_err(Into::into) }, @@ -817,7 +800,7 @@ mod tests { only_photos_rule, |_, _| {}, |_| async { Ok(vec![]) }, - |_| async { Ok(vec![]) }, + |_, _| async { Ok(vec![]) }, |path, is_dir| { IsolatedFilePathData::new(0, root_path, path, is_dir).map_err(Into::into) }, @@ -890,7 +873,7 @@ mod tests { git_repos, |_, _| {}, |_| async { Ok(vec![]) }, - |_| async { Ok(vec![]) }, + |_, _| async { Ok(vec![]) }, |path, is_dir| { IsolatedFilePathData::new(0, root_path, path, is_dir).map_err(Into::into) }, @@ -981,7 +964,7 @@ mod tests { git_repos_no_deps_no_build_dirs, |_, _| {}, |_| async { Ok(vec![]) }, - |_| async { Ok(vec![]) }, + |_, _| async { Ok(vec![]) }, |path, is_dir| { IsolatedFilePathData::new(0, root_path, path, is_dir).map_err(Into::into) }, diff --git a/core/src/location/manager/watcher/macos.rs b/core/src/location/manager/watcher/macos.rs index 81d3c3b22..d19ccdd47 100644 --- a/core/src/location/manager/watcher/macos.rs +++ b/core/src/location/manager/watcher/macos.rs @@ -22,10 +22,7 @@ use crate::{ util::error::FileIOError, }; -use std::{ - collections::{BTreeMap, HashMap}, - path::PathBuf, -}; +use std::{collections::HashMap, path::PathBuf}; use async_trait::async_trait; use notify::{ @@ -47,7 +44,8 @@ use super::{ pub(super) struct MacOsEventHandler<'lib> { location_id: location::id::Type, library: &'lib Library, - recently_created_files: BTreeMap, + recently_created_files: HashMap, + recently_created_files_buffer: Vec<(PathBuf, Instant)>, last_check_created_files: Instant, latest_created_dir: Option, last_check_rename: Instant, @@ -65,7 +63,8 @@ impl<'lib> EventHandler<'lib> for MacOsEventHandler<'lib> { Self { location_id, library, - recently_created_files: BTreeMap::new(), + recently_created_files: HashMap::new(), + recently_created_files_buffer: Vec::new(), last_check_created_files: Instant::now(), latest_created_dir: None, last_check_rename: Instant::now(), @@ -107,16 +106,6 @@ impl<'lib> EventHandler<'lib> for MacOsEventHandler<'lib> { self.latest_created_dir = Some(paths.remove(0)); } EventKind::Create(CreateKind::File) => { - let path = &paths[0]; - create_file( - self.location_id, - path, - &fs::metadata(path) - .await - .map_err(|e| FileIOError::from((path, e)))?, - self.library, - ) - .await?; self.recently_created_files .insert(paths.remove(0), Instant::now()); } @@ -143,66 +132,109 @@ impl<'lib> EventHandler<'lib> for MacOsEventHandler<'lib> { } async fn tick(&mut self) { - // Cleaning out recently created files that are older than 1 second - if self.last_check_created_files.elapsed() > ONE_SECOND { + // Cleaning out recently created files that are older than 200 milliseconds + if self.last_check_created_files.elapsed() > HUNDRED_MILLIS * 2 { + if let Err(e) = self.handle_recently_created_eviction().await { + error!("Error while handling recently created files eviction: {e:#?}"); + } self.last_check_created_files = Instant::now(); - self.recently_created_files - .retain(|_, created_at| created_at.elapsed() < ONE_SECOND); } if self.last_check_rename.elapsed() > HUNDRED_MILLIS { // Cleaning out recently renamed files that are older than 100 milliseconds - self.handle_create_eviction().await; - self.handle_remove_eviction().await; + if let Err(e) = self.handle_rename_create_eviction().await { + error!("Failed to create file_path on MacOS : {e:#?}"); + } + + if let Err(e) = self.handle_rename_remove_eviction().await { + error!("Failed to remove file_path: {e:#?}"); + } + + self.last_check_rename = Instant::now(); } } } impl MacOsEventHandler<'_> { - async fn handle_create_eviction(&mut self) { + async fn handle_recently_created_eviction(&mut self) -> Result<(), LocationManagerError> { + self.recently_created_files_buffer.clear(); + let mut should_invalidate = false; + + for (path, created_at) in self.recently_created_files.drain() { + if created_at.elapsed() < ONE_SECOND { + self.recently_created_files_buffer.push((path, created_at)); + } else { + create_file( + self.location_id, + &path, + &fs::metadata(&path) + .await + .map_err(|e| FileIOError::from((&path, e)))?, + self.library, + ) + .await?; + should_invalidate = true; + } + } + + if should_invalidate { + invalidate_query!(self.library, "search.paths"); + } + + self.recently_created_files + .extend(self.recently_created_files_buffer.drain(..)); + + Ok(()) + } + + async fn handle_rename_create_eviction(&mut self) -> Result<(), LocationManagerError> { // Just to make sure that our buffer is clean self.paths_map_buffer.clear(); + let mut should_invalidate = false; for (inode_and_device, (instant, path)) in self.new_paths_map.drain() { if instant.elapsed() > HUNDRED_MILLIS { - if let Err(e) = create_dir_or_file(self.location_id, &path, self.library).await { - error!("Failed to create file_path on MacOS : {e}"); - } else { - trace!("Created file_path due timeout: {}", path.display()); - invalidate_query!(self.library, "search.paths"); - } + create_dir_or_file(self.location_id, &path, self.library).await?; + trace!("Created file_path due timeout: {}", path.display()); + should_invalidate = true; } else { self.paths_map_buffer .push((inode_and_device, (instant, path))); } } - for (key, value) in self.paths_map_buffer.drain(..) { - self.new_paths_map.insert(key, value); + if should_invalidate { + invalidate_query!(self.library, "search.paths"); } + + self.new_paths_map.extend(self.paths_map_buffer.drain(..)); + + Ok(()) } - async fn handle_remove_eviction(&mut self) { + async fn handle_rename_remove_eviction(&mut self) -> Result<(), LocationManagerError> { // Just to make sure that our buffer is clean self.paths_map_buffer.clear(); + let mut should_invalidate = false; for (inode_and_device, (instant, path)) in self.old_paths_map.drain() { if instant.elapsed() > HUNDRED_MILLIS { - if let Err(e) = remove(self.location_id, &path, self.library).await { - error!("Failed to remove file_path: {e}"); - } else { - trace!("Removed file_path due timeout: {}", path.display()); - invalidate_query!(self.library, "search.paths"); - } + remove(self.location_id, &path, self.library).await?; + trace!("Removed file_path due timeout: {}", path.display()); + should_invalidate = true; } else { self.paths_map_buffer .push((inode_and_device, (instant, path))); } } - for (key, value) in self.paths_map_buffer.drain(..) { - self.old_paths_map.insert(key, value); + if should_invalidate { + invalidate_query!(self.library, "search.paths"); } + + self.old_paths_map.extend(self.paths_map_buffer.drain(..)); + + Ok(()) } async fn handle_single_rename_event( diff --git a/core/src/location/manager/watcher/utils.rs b/core/src/location/manager/watcher/utils.rs index 10fae3b2c..952b1d4af 100644 --- a/core/src/location/manager/watcher/utils.rs +++ b/core/src/location/manager/watcher/utils.rs @@ -44,7 +44,7 @@ use notify::{Event, EventKind}; use prisma_client_rust::{raw, PrismaValue}; use serde_json::json; use tokio::{fs, io::ErrorKind}; -use tracing::{debug, error, info, trace, warn}; +use tracing::{debug, error, trace, warn}; use uuid::Uuid; use super::INodeAndDevice; @@ -94,7 +94,7 @@ pub(super) async fn create_dir( { // FIXME: This is a workaround for Windows, because we can't get the inode and device from the metadata let _ = metadata; // To avoid unused variable warning - get_inode_and_device_from_path(&path).await? + get_inode_and_device_from_path(path).await? } }; @@ -113,7 +113,7 @@ pub(super) async fn create_dir( .materialized_path_for_children() .expect("We're in the create dir function lol"); - info!("Creating path: {}", iso_file_path); + debug!("Creating path: {}", iso_file_path); create_file_path( library, @@ -142,9 +142,26 @@ pub(super) async fn create_file( path: impl AsRef, metadata: &Metadata, library: &Library, +) -> Result<(), LocationManagerError> { + inner_create_file( + location_id, + extract_location_path(location_id, library).await?, + path, + metadata, + library, + ) + .await +} + +async fn inner_create_file( + location_id: location::id::Type, + location_path: impl AsRef, + path: impl AsRef, + metadata: &Metadata, + library: &Library, ) -> Result<(), LocationManagerError> { let path = path.as_ref(); - let location_path = extract_location_path(location_id, library).await?; + let location_path = location_path.as_ref(); trace!( "Location: creating file: {}", @@ -154,7 +171,7 @@ pub(super) async fn create_file( let db = &library.db; - let iso_file_path = IsolatedFilePathData::new(location_id, &location_path, path, false)?; + let iso_file_path = IsolatedFilePathData::new(location_id, location_path, path, false)?; let extension = iso_file_path.extension.to_string(); let (inode, device) = { @@ -171,6 +188,52 @@ pub(super) async fn create_file( } }; + // First we check if already exist a file with these same inode and device numbers + // if it does, we just update it + if let Some(file_path) = db + .file_path() + .find_unique(file_path::location_id_inode_device( + location_id, + inode.to_le_bytes().to_vec(), + device.to_le_bytes().to_vec(), + )) + .include(file_path_with_object::include()) + .exec() + .await? + { + trace!( + "File already exists with that inode and device: {}", + iso_file_path + ); + return inner_update_file(location_path, &file_path, path, library, None).await; + + // If we can't find an existing file with the same inode and device, we check if there is a file with the same path + } else if let Some(file_path) = db + .file_path() + .find_unique(file_path::location_id_materialized_path_name_extension( + location_id, + iso_file_path.materialized_path.to_string(), + iso_file_path.name.to_string(), + iso_file_path.extension.to_string(), + )) + .include(file_path_with_object::include()) + .exec() + .await? + { + trace!( + "File already exists with that iso_file_path: {}", + iso_file_path + ); + return inner_update_file( + location_path, + &file_path, + path, + library, + Some((inode, device)), + ) + .await; + } + let parent_iso_file_path = iso_file_path.parent(); if !parent_iso_file_path.is_root() && !check_file_path_exists::(&parent_iso_file_path, &library.db).await? @@ -186,7 +249,7 @@ pub(super) async fn create_file( fs_metadata, } = FileMetadata::new(&location_path, &iso_file_path).await?; - info!("Creating path: {}", iso_file_path); + debug!("Creating path: {}", iso_file_path); let created_file = create_file_path( library, @@ -292,32 +355,44 @@ pub(super) async fn update_file( .exec() .await? { - let ret = inner_update_file(location_id, file_path, full_path, library).await; - invalidate_query!(library, "search.paths"); - ret + inner_update_file(location_path, file_path, full_path, library, None).await } else { - // FIXME(fogodev): Have to handle files excluded by indexer rules - Err(LocationManagerError::UpdateNonExistingFile( - full_path.to_path_buf(), - )) + inner_create_file( + location_id, + location_path, + full_path, + &fs::metadata(full_path) + .await + .map_err(|e| FileIOError::from((full_path, e)))?, + library, + ) + .await } + .map(|_| invalidate_query!(library, "search.paths")) } async fn inner_update_file( - location_id: location::id::Type, + location_path: impl AsRef, file_path: &file_path_with_object::Data, full_path: impl AsRef, library @ Library { db, sync, .. }: &Library, + maybe_new_inode_and_device: Option, ) -> Result<(), LocationManagerError> { let full_path = full_path.as_ref(); - let location = db - .location() - .find_unique(location::id::equals(location_id)) - .exec() - .await? - .ok_or_else(|| LocationManagerError::MissingLocation(location_id))?; + let location_path = location_path.as_ref(); - let location_path = maybe_missing(location.path.map(PathBuf::from), "location.path")?; + let (current_inode, current_device) = ( + u64::from_le_bytes( + maybe_missing(file_path.inode.as_ref(), "file_path.inode")?[0..8] + .try_into() + .map_err(|_| LocationManagerError::InvalidInode)?, + ), + u64::from_le_bytes( + maybe_missing(file_path.device.as_ref(), "file_path.device")?[0..8] + .try_into() + .map_err(|_| LocationManagerError::InvalidDevice)?, + ), + ); trace!( "Location: updating file: {}", @@ -333,6 +408,29 @@ async fn inner_update_file( kind, } = FileMetadata::new(&location_path, &iso_file_path).await?; + let (inode, device) = if let Some((inode, device)) = maybe_new_inode_and_device { + (inode, device) + } else { + #[cfg(target_family = "unix")] + { + get_inode_and_device(&fs_metadata)? + } + + #[cfg(target_family = "windows")] + { + // FIXME: This is a workaround for Windows, because we can't get the inode and device from the metadata + get_inode_and_device_from_path(full_path).await? + } + }; + + let (maybe_new_inode, maybe_new_device) = + match (current_inode == inode, current_device == device) { + (true, true) => (None, None), + (true, false) => (None, Some(device)), + (false, true) => (Some(inode), None), + (false, false) => (Some(inode), Some(device)), + }; + if let Some(old_cas_id) = &file_path.cas_id { if old_cas_id != &cas_id { let (sync_params, db_params): (Vec<_>, Vec<_>) = { @@ -341,21 +439,23 @@ async fn inner_update_file( [ ( (cas_id::NAME, json!(old_cas_id)), - cas_id::set(Some(old_cas_id.clone())), + Some(cas_id::set(Some(old_cas_id.clone()))), ), ( ( size_in_bytes_bytes::NAME, json!(fs_metadata.len().to_be_bytes().to_vec()), ), - size_in_bytes_bytes::set(Some(fs_metadata.len().to_be_bytes().to_vec())), + Some(size_in_bytes_bytes::set(Some( + fs_metadata.len().to_be_bytes().to_vec(), + ))), ), { let date = DateTime::::from(fs_metadata.modified_or_now()).into(); ( (date_modified::NAME, json!(date)), - date_modified::set(Some(date)), + Some(date_modified::set(Some(date))), ) }, { @@ -373,11 +473,34 @@ async fn inner_update_file( ( (integrity_checksum::NAME, json!(checksum)), - integrity_checksum::set(checksum), + Some(integrity_checksum::set(checksum)), ) }, + { + if let Some(new_inode) = maybe_new_inode { + ( + (inode::NAME, json!(new_inode)), + Some(inode::set(Some(new_inode.to_le_bytes().to_vec()))), + ) + } else { + ((inode::NAME, serde_json::Value::Null), None) + } + }, + { + if let Some(new_device) = maybe_new_device { + ( + (device::NAME, json!(new_device)), + Some(device::set(Some(new_device.to_le_bytes().to_vec()))), + ) + } else { + ((device::NAME, serde_json::Value::Null), None) + } + }, ] .into_iter() + .filter_map(|(sync_param, maybe_db_param)| { + maybe_db_param.map(|db_param| (sync_param, db_param)) + }) .unzip() }; diff --git a/core/src/location/mod.rs b/core/src/location/mod.rs index de249f104..07c391e64 100644 --- a/core/src/location/mod.rs +++ b/core/src/location/mod.rs @@ -1,6 +1,6 @@ use crate::{ invalidate_query, - job::{Job, JobError, JobManagerError}, + job::{JobBuilder, JobError, JobManagerError}, library::Library, location::file_path_helper::filter_existing_file_path_params, object::{ @@ -28,7 +28,7 @@ use serde::Deserialize; use serde_json::json; use specta::Type; use tokio::{fs, io}; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use uuid::Uuid; mod error; @@ -381,13 +381,13 @@ pub async fn scan_location( let location_base_data = location::Data::from(&location); - Job::new_with_action( - IndexerJobInit { - location, - sub_path: None, - }, - "scan_location", - ) + JobBuilder::new(IndexerJobInit { + location, + sub_path: None, + }) + .with_action("scan_location") + .with_metadata(json!({"location": location_base_data.clone()})) + .build() .queue_next(FileIdentifierJobInit { location: location_base_data.clone(), sub_path: None, @@ -414,13 +414,16 @@ pub async fn scan_location_sub_path( let location_base_data = location::Data::from(&location); - Job::new_with_action( - IndexerJobInit { - location, - sub_path: Some(sub_path.clone()), - }, - "scan_location_sub_path", - ) + JobBuilder::new(IndexerJobInit { + location, + sub_path: Some(sub_path.clone()), + }) + .with_action("scan_location_sub_path") + .with_metadata(json!({ + "location": location_base_data.clone(), + "sub_path": sub_path.clone(), + })) + .build() .queue_next(FileIdentifierJobInit { location: location_base_data.clone(), sub_path: Some(sub_path.clone()), @@ -778,7 +781,11 @@ async fn check_nested_location( let comps = location_path.components().collect::>(); let is_a_child_location = potential_children.into_iter().any(|v| { - let comps2 = PathBuf::from(v.path.unwrap()); + let Some(location_path) = v.path else { + warn!("Missing location path on location at check nested location", v.id); + return false; + }; + let comps2 = PathBuf::from(location_path); let comps2 = comps2.components().collect::>(); if comps.len() > comps2.len() { diff --git a/core/src/object/file_identifier/file_identifier_job.rs b/core/src/object/file_identifier/file_identifier_job.rs index d615ec820..ac1779e3f 100644 --- a/core/src/object/file_identifier/file_identifier_job.rs +++ b/core/src/object/file_identifier/file_identifier_job.rs @@ -18,7 +18,8 @@ use std::{ }; use serde::{Deserialize, Serialize}; -use tracing::info; +use serde_json::json; +use tracing::{debug, info, trace}; use super::{process_identifier_file_paths, FileIdentifierJobError, CHUNK_SIZE}; @@ -50,28 +51,23 @@ pub struct FileIdentifierJobData { #[derive(Serialize, Deserialize, Default, Debug)] pub struct FileIdentifierJobRunMetadata { - report: FileIdentifierReport, cursor: file_path::id::Type, -} - -impl JobRunMetadata for FileIdentifierJobRunMetadata { - fn update(&mut self, new_data: Self) { - self.report.total_orphan_paths += new_data.report.total_orphan_paths; - self.report.total_objects_created += new_data.report.total_objects_created; - self.report.total_objects_linked += new_data.report.total_objects_linked; - self.report.total_objects_ignored += new_data.report.total_objects_ignored; - self.cursor = new_data.cursor; - } -} - -#[derive(Serialize, Deserialize, Debug, Default)] -pub struct FileIdentifierReport { total_orphan_paths: usize, total_objects_created: usize, total_objects_linked: usize, total_objects_ignored: usize, } +impl JobRunMetadata for FileIdentifierJobRunMetadata { + fn update(&mut self, new_data: Self) { + self.total_orphan_paths += new_data.total_orphan_paths; + self.total_objects_created += new_data.total_objects_created; + self.total_objects_linked += new_data.total_objects_linked; + self.total_objects_ignored += new_data.total_objects_ignored; + self.cursor = new_data.cursor; + } +} + #[async_trait::async_trait] impl StatefulJob for FileIdentifierJobInit { type Data = FileIdentifierJobData; @@ -88,7 +84,7 @@ impl StatefulJob for FileIdentifierJobInit { let init = self; let Library { db, .. } = &ctx.library; - info!("Identifying orphan File Paths..."); + debug!("Identifying orphan File Paths..."); let location_id = init.location.id; @@ -138,10 +134,10 @@ impl StatefulJob for FileIdentifierJobInit { }); } - info!("Found {} orphan file paths", orphan_count); + debug!("Found {} orphan file paths", orphan_count); let task_count = (orphan_count as f64 / CHUNK_SIZE as f64).ceil() as usize; - info!( + debug!( "Found {} orphan Paths. Will execute {} tasks...", orphan_count, task_count ); @@ -160,11 +156,9 @@ impl StatefulJob for FileIdentifierJobInit { Ok(( FileIdentifierJobRunMetadata { - report: FileIdentifierReport { - total_orphan_paths: orphan_count, - ..Default::default() - }, + total_orphan_paths: orphan_count, cursor: first_path.id, + ..Default::default() }, vec![(); task_count], ) @@ -209,18 +203,18 @@ impl StatefulJob for FileIdentifierJobInit { step_number, run_metadata.cursor, &ctx.library, - run_metadata.report.total_orphan_paths, + run_metadata.total_orphan_paths, ) .await?; - new_metadata.report.total_objects_created = total_objects_created; - new_metadata.report.total_objects_linked = total_objects_linked; + new_metadata.total_objects_created = total_objects_created; + new_metadata.total_objects_linked = total_objects_linked; new_metadata.cursor = new_cursor; ctx.progress_msg(format!( "Processed {} of {} orphan Paths", step_number * CHUNK_SIZE, - run_metadata.report.total_orphan_paths + run_metadata.total_orphan_paths )); Ok(new_metadata.into()) @@ -233,9 +227,9 @@ impl StatefulJob for FileIdentifierJobInit { run_metadata: &Self::RunMetadata, ) -> JobResult { let init = self; - info!("Finalizing identifier job: {:?}", &run_metadata.report); + info!("Finalizing identifier job: {:?}", &run_metadata); - Ok(Some(serde_json::to_value(init)?)) + Ok(Some(json!({"init: ": init, "run_metadata": run_metadata}))) } } @@ -286,9 +280,10 @@ async fn get_orphan_file_paths( file_path_id: file_path::id::Type, maybe_sub_materialized_path: &Option>, ) -> Result, prisma_client_rust::QueryError> { - info!( + trace!( "Querying {} orphan Paths at cursor: {:?}", - CHUNK_SIZE, file_path_id + CHUNK_SIZE, + file_path_id ); db.file_path() .find_many(orphan_path_filters( diff --git a/core/src/object/file_identifier/mod.rs b/core/src/object/file_identifier/mod.rs index 0ca88c09d..2ea078b2f 100644 --- a/core/src/object/file_identifier/mod.rs +++ b/core/src/object/file_identifier/mod.rs @@ -26,7 +26,7 @@ use futures::future::join_all; use serde_json::json; use thiserror::Error; use tokio::fs; -use tracing::{error, info}; +use tracing::{error, trace}; use uuid::Uuid; pub mod file_identifier_job; @@ -83,8 +83,7 @@ impl FileMetadata { .await .map_err(|e| FileIOError::from((&path, e)))?; - #[cfg(debug_assertions)] - tracing::debug!("Analyzed file: {path:?} {cas_id:?} {kind:?}"); + trace!("Analyzed file: {path:?} {cas_id:?} {kind:?}"); Ok(FileMetadata { cas_id, @@ -204,7 +203,7 @@ async fn identifier_job_step( ) .await?; - info!( + trace!( "Found {} existing Objects in Library, linking file paths...", existing_objects.len() ); @@ -221,7 +220,7 @@ async fn identifier_job_step( .map(|(_, (meta, _))| &meta.cas_id) .collect::>(); - info!( + trace!( "Creating {} new Objects in Library... {:#?}", file_paths_requiring_new_object.len(), new_objects_cas_ids @@ -283,10 +282,10 @@ async fn identifier_job_step( 0 }); - info!("Created {} new Objects in Library", total_created_files); + trace!("Created {} new Objects in Library", total_created_files); if total_created_files > 0 { - info!("Updating file paths with created objects"); + trace!("Updating file paths with created objects"); sync.write_ops(db, { let data: (Vec<_>, Vec<_>) = file_path_update_args.into_iter().unzip(); @@ -295,7 +294,7 @@ async fn identifier_job_step( }) .await?; - info!("Updated file paths with created objects"); + trace!("Updated file paths with created objects"); } total_created_files as usize @@ -313,7 +312,7 @@ fn file_path_object_connect_ops<'db>( db: &'db PrismaClient, ) -> (CRDTOperation, file_path::UpdateQuery<'db>) { #[cfg(debug_assertions)] - tracing::debug!("Connecting to "); + trace!("Connecting to "); let vec_id = object_id.as_bytes().to_vec(); @@ -342,7 +341,7 @@ async fn process_identifier_file_paths( library: &Library, orphan_count: usize, ) -> Result<(usize, usize, file_path::id::Type), JobError> { - info!( + trace!( "Processing {:?} orphan Paths. ({} completed of {})", file_paths.len(), step_number, diff --git a/core/src/object/file_identifier/shallow.rs b/core/src/object/file_identifier/shallow.rs index 8fef7b3e7..38aef6a06 100644 --- a/core/src/object/file_identifier/shallow.rs +++ b/core/src/object/file_identifier/shallow.rs @@ -13,7 +13,7 @@ use crate::{ use std::path::{Path, PathBuf}; use serde::{Deserialize, Serialize}; -use tracing::info; +use tracing::{debug, trace}; use super::{process_identifier_file_paths, FileIdentifierJobError, CHUNK_SIZE}; @@ -30,7 +30,7 @@ pub async fn shallow( ) -> Result<(), JobError> { let Library { db, .. } = &library; - info!("Identifying orphan File Paths..."); + debug!("Identifying orphan File Paths..."); let location_id = location.id; let location_path = maybe_missing(&location.path, "location.path").map(Path::new)?; @@ -68,7 +68,7 @@ pub async fn shallow( } let task_count = (orphan_count as f64 / CHUNK_SIZE as f64).ceil() as usize; - info!( + debug!( "Found {} orphan Paths. Will execute {} tasks...", orphan_count, task_count ); @@ -153,9 +153,10 @@ async fn get_orphan_file_paths( file_path_id_cursor: file_path::id::Type, sub_iso_file_path: &IsolatedFilePathData<'_>, ) -> Result, prisma_client_rust::QueryError> { - info!( + trace!( "Querying {} orphan Paths at cursor: {:?}", - CHUNK_SIZE, file_path_id_cursor + CHUNK_SIZE, + file_path_id_cursor ); db.file_path() .find_many(orphan_path_filters( diff --git a/core/src/object/fs/copy.rs b/core/src/object/fs/copy.rs index 27b0aeda0..4662e614c 100644 --- a/core/src/object/fs/copy.rs +++ b/core/src/object/fs/copy.rs @@ -16,6 +16,7 @@ use crate::{ use std::{hash::Hash, path::PathBuf}; use serde::{Deserialize, Serialize}; +use serde_json::json; use specta::Type; use tokio::{fs, io}; use tracing::{trace, warn}; @@ -207,6 +208,6 @@ impl StatefulJob for FileCopierJobInit { invalidate_query!(ctx.library, "search.paths"); - Ok(Some(serde_json::to_value(init)?)) + Ok(Some(json!({ "init": init }))) } } diff --git a/core/src/object/fs/cut.rs b/core/src/object/fs/cut.rs index 4d2c11687..259f5051a 100644 --- a/core/src/object/fs/cut.rs +++ b/core/src/object/fs/cut.rs @@ -14,6 +14,7 @@ use crate::{ use std::{hash::Hash, path::PathBuf}; use serde::{Deserialize, Serialize}; +use serde_json::json; use specta::Type; use tokio::{fs, io}; use tracing::{trace, warn}; @@ -130,6 +131,6 @@ impl StatefulJob for FileCutterJobInit { let init = self; invalidate_query!(ctx.library, "search.paths"); - Ok(Some(serde_json::to_value(init)?)) + Ok(Some(json!({ "init": init }))) } } diff --git a/core/src/object/fs/delete.rs b/core/src/object/fs/delete.rs index ff2fb5143..00fb581cb 100644 --- a/core/src/object/fs/delete.rs +++ b/core/src/object/fs/delete.rs @@ -11,6 +11,7 @@ use crate::{ use std::hash::Hash; use serde::{Deserialize, Serialize}; +use serde_json::json; use specta::Type; use tokio::{fs, io}; use tracing::warn; @@ -97,6 +98,6 @@ impl StatefulJob for FileDeleterJobInit { let init = self; invalidate_query!(ctx.library, "search.paths"); - Ok(Some(serde_json::to_value(init)?)) + Ok(Some(json!({ "init": init }))) } } diff --git a/core/src/object/preview/thumbnail/directory.rs b/core/src/object/preview/thumbnail/directory.rs index ca0a33680..1d5484831 100644 --- a/core/src/object/preview/thumbnail/directory.rs +++ b/core/src/object/preview/thumbnail/directory.rs @@ -2,7 +2,7 @@ use std::path::PathBuf; use tokio::fs as async_fs; use int_enum::IntEnum; -use tracing::{error, info}; +use tracing::{debug, error, trace}; use crate::util::{error::FileIOError, version_manager::VersionManager}; @@ -17,14 +17,14 @@ pub enum ThumbnailVersion { } pub async fn init_thumbnail_dir(data_dir: PathBuf) -> Result { - info!("Initializing thumbnail directory"); + debug!("Initializing thumbnail directory"); let thumbnail_dir = data_dir.join(THUMBNAIL_CACHE_DIR_NAME); let version_file = thumbnail_dir.join("version.txt"); let version_manager = VersionManager::::new(version_file.to_str().expect("Invalid path")); - info!("Thumbnail directory: {:?}", thumbnail_dir); + debug!("Thumbnail directory: {:?}", thumbnail_dir); // create all necessary directories if they don't exist async_fs::create_dir_all(&thumbnail_dir) @@ -34,7 +34,7 @@ pub async fn init_thumbnail_dir(data_dir: PathBuf) -> Result version, Err(_) => { - info!("Thumbnail version file does not exist, starting fresh"); + debug!("Thumbnail version file does not exist, starting fresh"); // Version file does not exist, start fresh version_manager.set_version(ThumbnailVersion::V1)?; ThumbnailVersion::V1 @@ -94,7 +94,7 @@ async fn move_webp_files(dir: &PathBuf) -> Result<(), ThumbnailerError> { } } } - info!( + trace!( "Moved {} webp files to their respective shard folders.", count ); diff --git a/core/src/object/preview/thumbnail/mod.rs b/core/src/object/preview/thumbnail/mod.rs index 25a0eb9b7..10e676cc5 100644 --- a/core/src/object/preview/thumbnail/mod.rs +++ b/core/src/object/preview/thumbnail/mod.rs @@ -23,7 +23,7 @@ use once_cell::sync::Lazy; use serde::{Deserialize, Serialize}; use thiserror::Error; use tokio::{fs, io, task::block_in_place}; -use tracing::{error, info, trace, warn}; +use tracing::{error, trace, warn}; use webp::Encoder; mod directory; @@ -226,14 +226,14 @@ pub async fn inner_process_step( match fs::metadata(&output_path).await { Ok(_) => { - info!( + trace!( "Thumb already exists, skipping generation for {}", output_path.display() ); return Ok(false); } Err(e) if e.kind() == io::ErrorKind::NotFound => { - info!("Writing {:?} to {:?}", path, output_path); + trace!("Writing {:?} to {:?}", path, output_path); match kind { ThumbnailerJobStepKind::Image => { @@ -249,7 +249,7 @@ pub async fn inner_process_step( } } - info!("Emitting new thumbnail event"); + trace!("Emitting new thumbnail event"); library.emit(CoreEvent::NewThumbnail { thumb_key: get_thumb_key(cas_id), }); diff --git a/core/src/object/preview/thumbnail/shallow.rs b/core/src/object/preview/thumbnail/shallow.rs index bcec742a8..044f46113 100644 --- a/core/src/object/preview/thumbnail/shallow.rs +++ b/core/src/object/preview/thumbnail/shallow.rs @@ -17,7 +17,7 @@ use sd_file_ext::extensions::Extension; use std::path::{Path, PathBuf}; use thumbnail::init_thumbnail_dir; use tokio::fs; -use tracing::info; +use tracing::{debug, trace}; #[cfg(feature = "ffmpeg")] use super::FILTERED_VIDEO_EXTENSIONS; @@ -66,7 +66,7 @@ pub async fn shallow_thumbnailer( ) }; - info!( + debug!( "Searching for images in location {location_id} at path {}", path.display() ); @@ -86,7 +86,7 @@ pub async fn shallow_thumbnailer( ) .await?; - info!("Found {:?} image files", image_files.len()); + trace!("Found {:?} image files", image_files.len()); #[cfg(feature = "ffmpeg")] let video_files = { @@ -100,7 +100,7 @@ pub async fn shallow_thumbnailer( ) .await?; - info!("Found {:?} video files", video_files.len()); + trace!("Found {:?} video files", video_files.len()); video_files }; diff --git a/core/src/object/preview/thumbnail/thumbnailer_job.rs b/core/src/object/preview/thumbnail/thumbnailer_job.rs index 370b33ed6..876222e0b 100644 --- a/core/src/object/preview/thumbnail/thumbnailer_job.rs +++ b/core/src/object/preview/thumbnail/thumbnailer_job.rs @@ -23,7 +23,8 @@ use sd_file_ext::extensions::Extension; use serde::{Deserialize, Serialize}; -use tracing::info; +use serde_json::json; +use tracing::{debug, info, trace}; use super::{ inner_process_step, ThumbnailerError, ThumbnailerJobStep, ThumbnailerJobStepKind, @@ -120,7 +121,7 @@ impl StatefulJob for ThumbnailerJobInit { ), }; - info!("Searching for images in location {location_id} at directory {iso_file_path}"); + debug!("Searching for images in location {location_id} at directory {iso_file_path}"); // query database for all image files in this location that need thumbnails let image_files = get_files_by_extensions( @@ -130,7 +131,7 @@ impl StatefulJob for ThumbnailerJobInit { ThumbnailerJobStepKind::Image, ) .await?; - info!("Found {:?} image files", image_files.len()); + trace!("Found {:?} image files", image_files.len()); #[cfg(feature = "ffmpeg")] let all_files = { @@ -142,7 +143,7 @@ impl StatefulJob for ThumbnailerJobInit { ThumbnailerJobStepKind::Video, ) .await?; - info!("Found {:?} video files", video_files.len()); + trace!("Found {:?} video files", video_files.len()); image_files .into_iter() @@ -228,7 +229,7 @@ impl StatefulJob for ThumbnailerJobInit { invalidate_query!(ctx.library, "search.paths"); } - Ok(Some(serde_json::to_value(run_metadata)?)) + Ok(Some(json!({"init: ": init, "run_metadata": run_metadata}))) } } diff --git a/core/src/object/validation/validator_job.rs b/core/src/object/validation/validator_job.rs index 3f606d2a9..d0ced3ec8 100644 --- a/core/src/object/validation/validator_job.rs +++ b/core/src/object/validation/validator_job.rs @@ -191,6 +191,6 @@ impl StatefulJob for ObjectValidatorJobInit { data.task_count ); - Ok(Some(serde_json::to_value(init)?)) + Ok(Some(json!({ "init": init }))) } } diff --git a/interface/app/$libraryId/Explorer/View/ListView.tsx b/interface/app/$libraryId/Explorer/View/ListView.tsx index fa58bfcc1..ffccd13e1 100644 --- a/interface/app/$libraryId/Explorer/View/ListView.tsx +++ b/interface/app/$libraryId/Explorer/View/ListView.tsx @@ -213,6 +213,7 @@ export default () => { id: 'dateAccessed', header: 'Date Accessed', accessorFn: (file) => + getItemObject(file)?.date_accessed && dayjs(getItemObject(file)?.date_accessed).format('MMM Do YYYY') }, { diff --git a/interface/app/$libraryId/Layout/Sidebar/JobManager/Job.tsx b/interface/app/$libraryId/Layout/Sidebar/JobManager/Job.tsx index 5f0277fbe..1910bb9be 100644 --- a/interface/app/$libraryId/Layout/Sidebar/JobManager/Job.tsx +++ b/interface/app/$libraryId/Layout/Sidebar/JobManager/Job.tsx @@ -88,16 +88,15 @@ function Job({ job, className, isChild }: JobProps) { // textItems={[[{ text: job.status }, { text: job.id, }]]} isChild={isChild} > - {isRunning || - (isPaused && ( -
- -
- ))} + {(isRunning || isPaused) && ( +
+ +
+ )} ); } diff --git a/interface/app/$libraryId/Layout/Sidebar/JobManager/JobGroup.tsx b/interface/app/$libraryId/Layout/Sidebar/JobManager/JobGroup.tsx index 30c68f4fb..b6b83c5d7 100644 --- a/interface/app/$libraryId/Layout/Sidebar/JobManager/JobGroup.tsx +++ b/interface/app/$libraryId/Layout/Sidebar/JobManager/JobGroup.tsx @@ -103,12 +103,10 @@ function JobGroup({ data: { jobs, ...data }, clearJob }: JobGroupProps) { )} - {/* TODO: FIX THIS, why is this not working? */} - {!isJobsRunning && (