Automatically reindex when relevant config changes are made

src/app/config.rs

@@ -8,7 +8,7 @@ use log::{error, info};
 use notify::{RecommendedWatcher, RecursiveMode, Watcher};
 use notify_debouncer_full::{Debouncer, FileIdMap};
 use regex::Regex;
-use tokio::sync::{mpsc::unbounded_channel, RwLock};
+use tokio::sync::{futures::Notified, mpsc::unbounded_channel, Notify, RwLock};
 
 use crate::app::Error;
 
@@ -75,6 +75,7 @@ pub struct Manager {
 	auth_secret: auth::Secret,
 	#[allow(dead_code)]
 	file_watcher: Arc<Debouncer<RecommendedWatcher, FileIdMap>>,
+	change_notify: Arc<Notify>,
 }
 
 impl Manager {
@@ -96,6 +97,7 @@ impl Manager {
 			config: Arc::new(RwLock::new(Config::default())),
 			auth_secret,
 			file_watcher: Arc::new(debouncer),
+			change_notify: Arc::default(),
 		};
 
 		tokio::task::spawn({
@@ -121,6 +123,10 @@ impl Manager {
 		Ok(manager)
 	}
 
+	pub fn on_config_change(&self) -> Notified {
+		self.change_notify.notified()
+	}
+
 	async fn reload_config(&self) -> Result<(), Error> {
 		let config = Self::read_config(&self.config_file_path).await?;
 		self.apply_config(config).await
@@ -147,6 +153,7 @@ impl Manager {
 	pub async fn apply_config(&self, new_config: storage::Config) -> Result<(), Error> {
 		let mut config = self.config.write().await;
 		*config = new_config.try_into()?;
+		self.change_notify.notify_waiters();
 		Ok(())
 	}
 
@@ -166,6 +173,7 @@ impl Manager {
 			let mut config = self.config.write().await;
 			op(&mut config)?;
 		}
+		self.change_notify.notify_waiters();
 		self.save_config().await?;
 		Ok(())
 	}

src/app/index.rs

@@ -33,7 +33,7 @@ impl Manager {
 			.await
 			.map_err(|e| Error::Io(directory.to_owned(), e))?;
 
-		let mut index_manager = Self {
+		let index_manager = Self {
 			index_file_path: directory.join("collection.index"),
 			index: Arc::default(),
 		};
@@ -59,7 +59,7 @@ impl Manager {
 		.unwrap()
 	}
 
-	pub async fn replace_index(&mut self, new_index: Index) {
+	pub async fn replace_index(&self, new_index: Index) {
 		spawn_blocking({
 			let index_manager = self.clone();
 			move || {
@@ -71,7 +71,7 @@ impl Manager {
 		.unwrap()
 	}
 
-	pub async fn persist_index(&mut self, index: &Index) -> Result<(), Error> {
+	pub async fn persist_index(&self, index: &Index) -> Result<(), Error> {
 		let serialized = match bitcode::serialize(index) {
 			Ok(s) => s,
 			Err(_) => return Err(Error::IndexSerializationError),
@@ -82,7 +82,7 @@ impl Manager {
 		Ok(())
 	}
 
-	async fn try_restore_index(&mut self) -> Result<bool, Error> {
+	async fn try_restore_index(&self) -> Result<bool, Error> {
 		match tokio::fs::try_exists(&self.index_file_path).await {
 			Ok(true) => (),
 			Ok(false) => return Ok(false),
@@ -385,7 +385,7 @@ mod test {
 
 	#[tokio::test]
 	async fn can_persist_index() {
-		let mut ctx = test::ContextBuilder::new(test_name!()).build().await;
+		let ctx = test::ContextBuilder::new(test_name!()).build().await;
 		assert_eq!(ctx.index_manager.try_restore_index().await.unwrap(), false);
 		let index = index::Builder::new().build();
 		ctx.index_manager.persist_index(&index).await.unwrap();

src/app/playlist.rs

@@ -238,13 +238,13 @@ mod test {
 
 	#[tokio::test]
 	async fn save_playlist_is_idempotent() {
-		let mut ctx = test::ContextBuilder::new(test_name!())
+		let ctx = test::ContextBuilder::new(test_name!())
 			.user(TEST_USER, TEST_PASSWORD, false)
 			.mount(TEST_MOUNT_NAME, "test-data/small-collection")
 			.build()
 			.await;
 
-		ctx.scanner.update().await.unwrap();
+		ctx.scanner.run_scan().await.unwrap();
 
 		let songs = list_all_songs(&ctx).await;
 
@@ -293,13 +293,13 @@ mod test {
 
 	#[tokio::test]
 	async fn read_playlist_golden_path() {
-		let mut ctx = test::ContextBuilder::new(test_name!())
+		let ctx = test::ContextBuilder::new(test_name!())
 			.user(TEST_USER, TEST_PASSWORD, false)
 			.mount(TEST_MOUNT_NAME, "test-data/small-collection")
 			.build()
 			.await;
 
-		ctx.scanner.update().await.unwrap();
+		ctx.scanner.run_scan().await.unwrap();
 
 		let songs = list_all_songs(&ctx).await;
 

src/app/scanner.rs

@@ -10,6 +10,7 @@ use std::time::SystemTime;
 use std::{cmp::min, time::Duration};
 use tokio::sync::mpsc::unbounded_channel;
 use tokio::sync::{Notify, RwLock};
+use tokio::task::JoinSet;
 use tokio::time::Instant;
 
 use crate::app::{config, formats, index, Error};
@@ -48,6 +49,20 @@ pub enum State {
 	UpToDate,
 }
 
+#[derive(Clone)]
+struct Parameters {
+	artwork_regex: Option<Regex>,
+	mount_dirs: Vec<config::MountDir>,
+}
+
+impl PartialEq for Parameters {
+	fn eq(&self, other: &Self) -> bool {
+		self.artwork_regex.as_ref().map(|r| r.as_str())
+			== other.artwork_regex.as_ref().map(|r| r.as_str())
+			&& self.mount_dirs == other.mount_dirs
+	}
+}
+
 #[derive(Clone, Default)]
 pub struct Status {
 	pub state: State,
@@ -62,6 +77,7 @@ pub struct Scanner {
 	config_manager: config::Manager,
 	pending_scan: Arc<Notify>,
 	status: Arc<RwLock<Status>>,
+	parameters: Arc<RwLock<Option<Parameters>>>,
 }
 
 impl Scanner {
@@ -71,19 +87,53 @@ impl Scanner {
 	) -> Result<Self, Error> {
 		let scanner = Self {
 			index_manager,
-			config_manager,
+			config_manager: config_manager.clone(),
 			pending_scan: Arc::new(Notify::new()),
 			status: Arc::new(RwLock::new(Status::default())),
+			parameters: Arc::default(),
 		};
 
+		let abort_scan = Arc::new(Notify::new());
+
 		tokio::spawn({
-			let mut scanner = scanner.clone();
+			let config_manager = config_manager.clone();
+			let scanner = scanner.clone();
+			let abort_scan = abort_scan.clone();
+			async move {
+				loop {
+					config_manager.on_config_change().await;
+					if *scanner.parameters.read().await == Some(scanner.read_parameters().await) {
+						continue;
+					}
+					abort_scan.notify_waiters();
+					scanner.status.write().await.state = State::Pending;
+					while tokio::time::timeout(
+						Duration::from_secs(2),
+						config_manager.on_config_change(),
+					)
+					.await
+					.is_ok()
+					{}
+					scanner.pending_scan.notify_waiters();
+				}
+			}
+		});
+
+		tokio::spawn({
+			let scanner = scanner.clone();
 			async move {
 				loop {
 					scanner.pending_scan.notified().await;
-					if let Err(e) = scanner.update().await {
+					tokio::select! {
-						error!("Error while updating index: {}", e);
+						result = scanner.run_scan() => {
-					}
+							if let Err(e) = result {
+								error!("Error while updating index: {e}");
+							}
+						}
+						_ = abort_scan.notified() => {
+							info!("Interrupted index update");
+						}
+					};
 				}
 			}
 		});
@@ -91,28 +141,28 @@ impl Scanner {
 		Ok(scanner)
 	}
 
+	async fn read_parameters(&self) -> Parameters {
+		let album_art_pattern = self.config_manager.get_index_album_art_pattern().await;
+		let artwork_regex = Regex::new(&format!("(?i){}", &album_art_pattern)).ok();
+		Parameters {
+			artwork_regex,
+			mount_dirs: self.config_manager.get_mounts().await,
+		}
+	}
+
 	pub async fn get_status(&self) -> Status {
 		self.status.read().await.clone()
 	}
 
-	pub fn trigger_scan(&self) {
+	pub fn queue_scan(&self) {
 		self.pending_scan.notify_one();
 	}
 
-	pub fn begin_periodic_scans(&self) {
+	pub fn try_trigger_scan(&self) {
-		tokio::spawn({
+		self.pending_scan.notify_waiters();
-			let index = self.clone();
-			async move {
-				loop {
-					index.trigger_scan();
-					let sleep_duration = index.config_manager.get_index_sleep_duration().await;
-					tokio::time::sleep(sleep_duration).await;
-				}
-			}
-		});
 	}
 
-	pub async fn update(&mut self) -> Result<(), Error> {
+	pub async fn run_scan(&self) -> Result<(), Error> {
 		info!("Beginning collection scan");
 
 		let start = Instant::now();
@@ -125,24 +175,22 @@ impl Scanner {
 		let was_empty = self.index_manager.is_index_empty().await;
 		let mut partial_update_time = Instant::now();
 
-		let album_art_pattern = self.config_manager.get_index_album_art_pattern().await;
+		let new_parameters = self.read_parameters().await;
-		let album_art_regex = Regex::new(&format!("(?i){}", &album_art_pattern)).ok();
+		*self.parameters.write().await = Some(new_parameters.clone());
 
 		let (scan_directories_output, collection_directories_input) = channel();
 		let (scan_songs_output, collection_songs_input) = channel();
+		let scan = Scan::new(scan_directories_output, scan_songs_output, new_parameters);
 
-		let scan = Scan::new(
+		let mut scan_task_set = JoinSet::new();
-			scan_directories_output,
+		let mut index_task_set = JoinSet::new();
-			scan_songs_output,
+		let mut secondary_task_set = JoinSet::new();
-			self.config_manager.get_mounts().await,
-			album_art_regex,
-		);
 
-		let scan_task = tokio::task::spawn_blocking(|| scan.run());
+		scan_task_set.spawn_blocking(|| scan.run());
 
 		let partial_index_notify = Arc::new(tokio::sync::Notify::new());
 		let partial_index_mutex = Arc::new(tokio::sync::Mutex::new(index::Builder::default()));
-		let partial_application_task = tokio::task::spawn({
+		secondary_task_set.spawn({
 			let index_manager = self.index_manager.clone();
 			let partial_index_notify = partial_index_notify.clone();
 			let partial_index_mutex = partial_index_mutex.clone();
@@ -161,7 +209,7 @@ impl Scanner {
 		});
 
 		let (status_sender, mut status_receiver) = unbounded_channel();
-		let progress_monitor = tokio::task::spawn({
+		secondary_task_set.spawn({
 			let manager = self.clone();
 			async move {
 				loop {
@@ -175,7 +223,7 @@ impl Scanner {
 			}
 		});
 
-		let index_task = tokio::task::spawn_blocking(move || {
+		index_task_set.spawn_blocking(move || {
 			let mut index_builder = index::Builder::default();
 			let mut num_songs_scanned = 0;
 
@@ -219,9 +267,9 @@ impl Scanner {
 			index_builder.build()
 		});
 
-		let index = tokio::join!(scan_task, index_task).1?;
+		scan_task_set.join_next().await.unwrap()??;
-		partial_application_task.abort();
+		let index = index_task_set.join_next().await.unwrap()?;
-		progress_monitor.abort();
+		secondary_task_set.abort_all();
 
 		self.index_manager.persist_index(&index).await?;
 		self.index_manager.replace_index(index).await;
@@ -244,22 +292,19 @@ impl Scanner {
 struct Scan {
 	directories_output: Sender<Directory>,
 	songs_output: Sender<Song>,
-	mounts: Vec<config::MountDir>,
+	parameters: Parameters,
-	artwork_regex: Option<Regex>,
 }
 
 impl Scan {
 	pub fn new(
 		directories_output: Sender<Directory>,
 		songs_output: Sender<Song>,
-		mounts: Vec<config::MountDir>,
+		parameters: Parameters,
-		artwork_regex: Option<Regex>,
 	) -> Self {
 		Self {
 			directories_output,
 			songs_output,
-			mounts,
+			parameters,
-			artwork_regex,
 		}
 	}
 
@@ -273,12 +318,12 @@ impl Scan {
 
 		let directories_output = self.directories_output.clone();
 		let songs_output = self.songs_output.clone();
-		let artwork_regex = self.artwork_regex.clone();
+		let artwork_regex = self.parameters.artwork_regex.clone();
 
 		let thread_pool = ThreadPoolBuilder::new().num_threads(num_threads).build()?;
 		thread_pool.scope({
 			|scope| {
-				for mount in self.mounts {
+				for mount in self.parameters.mount_dirs {
 					scope.spawn(|scope| {
 						process_directory(
 							scope,
@@ -412,13 +457,15 @@ mod test {
 	async fn scan_finds_songs_and_directories() {
 		let (directories_sender, directories_receiver) = channel();
 		let (songs_sender, songs_receiver) = channel();
-		let mounts = vec![config::MountDir {
+		let parameters = Parameters {
-			source: ["test-data", "small-collection"].iter().collect(),
+			artwork_regex: None,
-			name: "root".to_owned(),
+			mount_dirs: vec![config::MountDir {
-		}];
+				source: ["test-data", "small-collection"].iter().collect(),
-		let artwork_regex = None;
+				name: "root".to_owned(),
+			}],
+		};
 
-		let scan = Scan::new(directories_sender, songs_sender, mounts, artwork_regex);
+		let scan = Scan::new(directories_sender, songs_sender, parameters);
 		scan.run().unwrap();
 
 		let directories = directories_receiver.iter().collect::<Vec<_>>();
@@ -432,13 +479,15 @@ mod test {
 	async fn scan_finds_embedded_artwork() {
 		let (directories_sender, _) = channel();
 		let (songs_sender, songs_receiver) = channel();
-		let mounts = vec![config::MountDir {
+		let parameters = Parameters {
-			source: ["test-data", "small-collection"].iter().collect(),
+			artwork_regex: None,
-			name: "root".to_owned(),
+			mount_dirs: vec![config::MountDir {
-		}];
+				source: ["test-data", "small-collection"].iter().collect(),
-		let artwork_regex = None;
+				name: "root".to_owned(),
+			}],
+		};
 
-		let scan = Scan::new(directories_sender, songs_sender, mounts, artwork_regex);
+		let scan = Scan::new(directories_sender, songs_sender, parameters);
 		scan.run().unwrap();
 
 		let songs = songs_receiver.iter().collect::<Vec<_>>();
@@ -455,13 +504,15 @@ mod test {
 		for pattern in patterns.into_iter() {
 			let (directories_sender, _) = channel();
 			let (songs_sender, songs_receiver) = channel();
-			let mounts = vec![config::MountDir {
+			let parameters = Parameters {
-				source: ["test-data", "small-collection"].iter().collect(),
+				artwork_regex: Some(Regex::new(pattern).unwrap()),
-				name: "root".to_owned(),
+				mount_dirs: vec![config::MountDir {
-			}];
+					source: ["test-data", "small-collection"].iter().collect(),
-			let artwork_regex = Some(Regex::new(pattern).unwrap());
+					name: "root".to_owned(),
+				}],
+			};
 
-			let scan = Scan::new(directories_sender, songs_sender, mounts, artwork_regex);
+			let scan = Scan::new(directories_sender, songs_sender, parameters);
 			scan.run().unwrap();
 
 			let songs = songs_receiver.iter().collect::<Vec<_>>();

src/main.rs

@@ -144,7 +144,7 @@ fn main() -> Result<(), Error> {
 async fn async_main(cli_options: CLIOptions, paths: paths::Paths) -> Result<(), Error> {
 	// Create and run app
 	let app = app::App::new(cli_options.port.unwrap_or(5050), paths).await?;
-	app.scanner.begin_periodic_scans();
+	app.scanner.queue_scan();
 	app.ddns_manager.begin_periodic_updates();
 
 	// Start server

src/server/axum/api.rs

@@ -252,7 +252,7 @@ async fn post_trigger_index(
 	_admin_rights: AdminRights,
 	State(scanner): State<scanner::Scanner>,
 ) -> Result<(), APIError> {
-	scanner.trigger_scan();
+	scanner.try_trigger_scan();
 	Ok(())
 }