feat(app): crawl shares in parallel, with a stop button

Each share is its own session against its own tree, so one large share
no longer holds the rest back. Stop ends every listing still running;
a share already past its listing finishes storing what it found, since
stopping halfway through replacing its rows would leave them disagreeing
with its downloads.

crates/pstr-app/src/app.rs

@@ -103,6 +103,8 @@ pub enum Action {
     RemoveDownload(DownloadKey, bool),
     /// Crawl one share, or every share.
     Crawl(Option<String>),
+    /// Stop the crawls still listing.
+    StopCrawl,
     AddShare {
         name: String,
         url: String,
@@ -422,7 +424,8 @@ pub struct App {
     pub connecting: bool,
     pub crawling: bool,
     /// Which share the crawl is on, and how far it has got.
-    crawl_progress: Option<(String, usize)>,
+    /// Files found so far, by share name, for each share still listing.
+    crawl_progress: std::collections::BTreeMap<String, usize>,
     pub downloads: Vec<DownloadItem>,
     pub offline_files: std::collections::HashSet<DownloadKey>,
     pub confirm_partial_delete: Option<DownloadKey>,
@@ -486,7 +489,7 @@ impl App {
             opening: None,
             connecting: true,
             crawling: false,
-            crawl_progress: None,
+            crawl_progress: Default::default(),
             downloads: Vec::new(),
             offline_files: std::collections::HashSet::new(),
             confirm_partial_delete: None,
@@ -551,6 +554,7 @@ impl App {
                     seconds,
                 } => {
                     let name = self.share_name(&share_id);
+                    self.crawl_progress.remove(&name);
                     self.note(
                         ctx,
                         format!("{name}: {files} playable of {nodes} nodes in {seconds:.0}s"),
@@ -559,11 +563,16 @@ impl App {
                 }
                 Event::CrawlFinished => {
                     self.crawling = false;
-                    self.crawl_progress = None;
+                    self.crawl_progress.clear();
                 }
                 Event::CrawlProgress { share, found } => {
                     self.crawling = true;
-                    self.crawl_progress = Some((share, found));
+                    self.crawl_progress.insert(share, found);
+                }
+                Event::CrawlStopped { share_id } => {
+                    let name = self.share_name(&share_id);
+                    self.crawl_progress.remove(&name);
+                    self.note(ctx, format!("stopped crawling {name}"), false);
                 }
                 Event::Thumbnail { key, image } => self.thumbs.insert(key, image),
                 Event::ThumbnailMissing { key } => self.thumbs.mark_missing(key),
@@ -1058,6 +1067,7 @@ impl App {
                 self.note(ctx, "crawling…", false);
                 self.engine.crawl(share);
             }
+            Action::StopCrawl => self.engine.stop_crawl(),
             Action::AddShare {
                 name,
                 url,
@@ -1266,12 +1276,15 @@ impl App {
             ui.with_layout(egui::Layout::right_to_left(egui::Align::Center), |ui| {
                 ui.add_space(4.0);
                 if self.crawling {
+                    if ui
+                        .small_button("Stop")
+                        .on_hover_text("Stop listing; shares already listed are still stored")
+                        .clicked()
+                    {
+                        actions.push(Action::StopCrawl);
+                    }
                     ui.add(egui::Spinner::new().size(16.0));
-                    ui.label(ui::muted(match &self.crawl_progress {
-                        Some((share, 0)) => format!("crawling {share}"),
-                        Some((share, found)) => format!("crawling {share} · {found} found"),
-                        None => "crawling".to_owned(),
-                    }));
+                    ui.label(ui::muted(crawl_label(&self.crawl_progress)));
                 } else if self.matching {
                     ui.add(egui::Spinner::new().size(16.0));
                     ui.label(ui::muted("matching"));
@@ -1621,6 +1634,22 @@ impl eframe::App for App {
     }
 }
 
+/// What the top bar says while shares are being listed: one share by name,
+/// several as a count, with the files found across all of them.
+fn crawl_label(progress: &std::collections::BTreeMap<String, usize>) -> String {
+    let found: usize = progress.values().sum();
+    let which = match progress.len() {
+        0 => return "crawling".to_owned(),
+        1 => progress.keys().next().cloned().unwrap_or_default(),
+        shares => format!("{shares} shares"),
+    };
+    if found == 0 {
+        format!("crawling {which}")
+    } else {
+        format!("crawling {which} · {found} found")
+    }
+}
+
 /// Titles whose tile art is not what it was.
 ///
 /// Compared by URL: the same URL is the same picture, and anything else —
@@ -1889,6 +1918,17 @@ mod tests {
         assert!(!Filter::Watching.admits(&film));
     }
 
+    #[test]
+    fn the_crawl_label_names_one_share_and_counts_several() {
+        let mut progress = std::collections::BTreeMap::new();
+        assert_eq!(crawl_label(&progress), "crawling");
+        progress.insert("anime".to_owned(), 0);
+        assert_eq!(crawl_label(&progress), "crawling anime");
+        progress.insert("films".to_owned(), 40);
+        progress.insert("anime".to_owned(), 2);
+        assert_eq!(crawl_label(&progress), "crawling 2 shares · 42 found");
+    }
+
     #[test]
     fn speed_steps_land_back_on_normal() {
         assert_eq!(step_speed(1.0, true), 1.25);

crates/pstr-app/src/engine.rs

@@ -186,6 +186,8 @@ pub enum Event {
     CrawlProgress { share: String, found: usize },
     /// Every requested crawl is done.
     CrawlFinished,
+    /// That share's crawl was stopped before it stored anything.
+    CrawlStopped { share_id: String },
     /// A decoded Proton thumbnail, keyed as [`thumbnail_key`].
     Thumbnail {
         key: String,
@@ -314,6 +316,8 @@ pub struct Engine {
     download_permits: Arc<tokio::sync::Semaphore>,
     /// Whether the player page is on screen. See [`Engine::wake`].
     picture_shown: Arc<AtomicBool>,
+    /// Bumped to stop the crawls in flight; each watches it from its start.
+    crawl_cancel: Arc<tokio::sync::watch::Sender<u64>>,
     events: Sender<Event>,
     ctx: egui::Context,
 }
@@ -410,6 +414,7 @@ impl Engine {
             downloads: Arc::new(Mutex::new(Downloads::default())),
             download_permits: Arc::new(tokio::sync::Semaphore::new(DOWNLOAD_CONCURRENCY)),
             picture_shown: Arc::new(AtomicBool::new(false)),
+            crawl_cancel: Arc::new(tokio::sync::watch::channel(0).0),
             events,
             ctx,
         };
@@ -798,117 +803,148 @@ impl Engine {
             None => library.share_ids().map(str::to_string).collect(),
         };
 
-        for share_id in targets {
-            let started = std::time::Instant::now();
-            let name = self
-                .store
-                .list()
-                .ok()
-                .and_then(|shares| shares.into_iter().find(|share| share.id == share_id))
-                .map_or_else(|| share_id.clone(), |share| share.name);
-            self.emit(Event::CrawlProgress {
-                share: name.clone(),
-                found: 0,
-            });
-            // At most a few times a second: a wide level comes back as a burst
-            // of listings, and a frame per listing is a frame per folder.
-            let last = parking_lot::Mutex::new(std::time::Instant::now());
-            let report = |found: usize| {
-                let mut last = last.lock();
-                if last.elapsed() >= std::time::Duration::from_millis(250) {
-                    *last = std::time::Instant::now();
-                    self.emit(Event::CrawlProgress {
-                        share: name.clone(),
-                        found,
-                    });
-                }
-            };
-            let nodes = match library.crawl_reporting(&share_id, report).await {
+        // Every share at once: each is its own session against its own
+        // tree, and one large share no longer holds the rest back behind it.
+        let cancel = self.crawl_cancel.subscribe();
+        futures::future::join_all(
+            targets
+                .into_iter()
+                .map(|share_id| self.crawl_share(&library, share_id, cancel.clone())),
+        )
+        .await;
+        self.emit(Event::CrawlFinished);
+    }
+
+    /// Stop every crawl still listing. Shares already past their listing
+    /// finish storing what they found.
+    pub fn stop_crawl(&self) {
+        self.crawl_cancel.send_modify(|generation| *generation += 1);
+    }
+
+    /// Crawl one share and store what it holds.
+    async fn crawl_share(
+        &self,
+        library: &SharedLibrary,
+        share_id: String,
+        mut cancel: tokio::sync::watch::Receiver<u64>,
+    ) {
+        let started = std::time::Instant::now();
+        let name = self
+            .store
+            .list()
+            .ok()
+            .and_then(|shares| shares.into_iter().find(|share| share.id == share_id))
+            .map_or_else(|| share_id.clone(), |share| share.name);
+        self.emit(Event::CrawlProgress {
+            share: name.clone(),
+            found: 0,
+        });
+        // At most a few times a second: a wide level comes back as a burst
+        // of listings, and a frame per listing is a frame per folder.
+        let last = parking_lot::Mutex::new(std::time::Instant::now());
+        let report = |found: usize| {
+            let mut last = last.lock();
+            if last.elapsed() >= std::time::Duration::from_millis(250) {
+                *last = std::time::Instant::now();
+                self.emit(Event::CrawlProgress {
+                    share: name.clone(),
+                    found,
+                });
+            }
+        };
+        // Only the listing can be stopped. Past it the crawl is replacing the
+        // share's rows and its stale downloads, and stopping halfway through
+        // that would leave the two disagreeing.
+        let nodes = tokio::select! {
+            result = library.crawl_reporting(&share_id, report) => match result {
                 Ok(nodes) => nodes,
                 Err(error) => {
                     self.fail(&format!("crawl {share_id}"), error);
-                    continue;
+                    return;
                 }
-            };
-            let rows = build_rows(&share_id, &nodes);
-            let files = rows.len();
-
-            // A writer opened before the crawl may be producing the revision
-            // this crawl is about to supersede. Quiesce all writers for the
-            // share before inspecting or deleting any revision paths.
-            let active_downloads: Vec<_> = self
-                .downloads
-                .lock()
-                .jobs
-                .keys()
-                .filter(|key| key.share_id == share_id)
-                .cloned()
-                .collect();
-            for key in &active_downloads {
-                self.cancel_and_join(key).await;
+            },
+            Ok(()) = cancel.changed() => {
+                self.emit(Event::CrawlStopped { share_id });
+                return;
             }
+        };
+        let rows = build_rows(&share_id, &nodes);
+        let files = rows.len();
 
-            // Paths are derived from the old revision id, so inspect and
-            // remove stale bytes before changing or dropping their index rows.
-            let retained = match self.catalog.lock().offline_files_for_share(&share_id) {
-                Ok(files) => files,
-                Err(error) => {
-                    self.fail(&format!("inspect downloads for {share_id}"), error);
-                    continue;
-                }
-            };
-            let revisions: HashMap<&str, Option<&str>> = rows
-                .iter()
-                .map(|row| (row.link_id.as_str(), row.active_revision_id.as_deref()))
-                .collect();
-            let stale: Vec<_> = retained
-                .iter()
-                .filter(|(link_id, file)| {
-                    revisions.get(link_id.as_str()).copied().flatten()
-                        != Some(file.revision_id.as_str())
-                })
-                .map(|(link_id, file)| (link_id.clone(), file.clone()))
-                .collect();
-            let mut cleanup_failed = false;
-            for (link_id, file) in &stale {
-                let path = self
-                    .dirs
-                    .offline_file(&share_id, link_id, &file.revision_id);
-                if let Err(error) = remove_if_present(&path).await {
-                    self.fail(&format!("delete stale download for {share_id}"), error);
-                    cleanup_failed = true;
-                    break;
-                }
-            }
-            if cleanup_failed {
-                continue;
-            }
-
-            let stored = {
-                let mut catalog = self.catalog.lock();
-                catalog
-                    .replace_share_retaining_offline(&share_id, &rows)
-                    .and_then(|()| {
-                        for (link_id, _) in &stale {
-                            catalog.remove_offline_file(&share_id, link_id)?;
-                        }
-                        Ok(())
-                    })
-            };
-            if let Err(error) = stored {
-                self.fail(&format!("store {share_id}"), error);
-                continue;
-            }
-
-            self.emit(Event::Crawled {
-                share_id,
-                nodes: nodes.len(),
-                files,
-                seconds: started.elapsed().as_secs_f64(),
-            });
-            self.load_library();
+        // A writer opened before the crawl may be producing the revision
+        // this crawl is about to supersede. Quiesce all writers for the
+        // share before inspecting or deleting any revision paths.
+        let active_downloads: Vec<_> = self
+            .downloads
+            .lock()
+            .jobs
+            .keys()
+            .filter(|key| key.share_id == share_id)
+            .cloned()
+            .collect();
+        for key in &active_downloads {
+            self.cancel_and_join(key).await;
         }
-        self.emit(Event::CrawlFinished);
+
+        // Paths are derived from the old revision id, so inspect and
+        // remove stale bytes before changing or dropping their index rows.
+        let retained = match self.catalog.lock().offline_files_for_share(&share_id) {
+            Ok(files) => files,
+            Err(error) => {
+                self.fail(&format!("inspect downloads for {share_id}"), error);
+                return;
+            }
+        };
+        let revisions: HashMap<&str, Option<&str>> = rows
+            .iter()
+            .map(|row| (row.link_id.as_str(), row.active_revision_id.as_deref()))
+            .collect();
+        let stale: Vec<_> = retained
+            .iter()
+            .filter(|(link_id, file)| {
+                revisions.get(link_id.as_str()).copied().flatten()
+                    != Some(file.revision_id.as_str())
+            })
+            .map(|(link_id, file)| (link_id.clone(), file.clone()))
+            .collect();
+        let mut cleanup_failed = false;
+        for (link_id, file) in &stale {
+            let path = self
+                .dirs
+                .offline_file(&share_id, link_id, &file.revision_id);
+            if let Err(error) = remove_if_present(&path).await {
+                self.fail(&format!("delete stale download for {share_id}"), error);
+                cleanup_failed = true;
+                break;
+            }
+        }
+        if cleanup_failed {
+            return;
+        }
+
+        let stored = {
+            let mut catalog = self.catalog.lock();
+            catalog
+                .replace_share_retaining_offline(&share_id, &rows)
+                .and_then(|()| {
+                    for (link_id, _) in &stale {
+                        catalog.remove_offline_file(&share_id, link_id)?;
+                    }
+                    Ok(())
+                })
+        };
+        if let Err(error) = stored {
+            self.fail(&format!("store {share_id}"), error);
+            return;
+        }
+
+        self.emit(Event::Crawled {
+            share_id,
+            nodes: nodes.len(),
+            files,
+            seconds: started.elapsed().as_secs_f64(),
+        });
+        self.load_library();
     }
 
     /// Record where playback got to.