Kept artwork off the async threads, and one fetch per missing image (#345 #349)

#345: a cache hit made four blocking std::fs calls on a tokio worker,
and the cache was on shfs (FUSE). A page asks for about fifty images at
once, so hits ran 11 ms at p50 and over a second at p90. Reading,
keeping and the reap job's trim now run in spawn_blocking. Production's
cache also moves to /mnt/fast in the compose file, outside the repo.

#349: two requests for the same uncached image both fetched it from its
publisher. A per-address lock now makes the second wait for the first,
then find the image kept. The lock is forgotten once its last holder is
done, whichever way the request ended.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/ipx/CHANGELOG.md

@@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - A page of items finds its files by feed as well as item: it read every file's row in the database, about a quarter of a second a page (#344).
 - A page of items asks for its count beside the items, and the feed list asks its five questions at once, where each waited for the one before (#346).
 - A Signal's count of items and of unread ones are worked out in one pass, not two (#348).
+- Artwork kept on disk is read and written off the threads that answer requests, and two requests for the same missing artwork fetch it once (#345, #349).
 
 ### Fixed
 

crates/ipx/src/main.rs

@@ -1238,8 +1238,9 @@ async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
     span.record("entries", r.entries_pruned);
     if !dry_run {
         tracing::Instrument::instrument(clean_directory(ctx), db::step("clean_directory")).await?;
-        let gone =
+        // Off the async threads: it walks every file kept, on FUSE (#345).
-            db::step("trim_artwork").in_scope(|| art::trim(&art::dir(), ctx.cfg().general.art_cache_mb * 1_048_576));
+        let (span, limit) = (db::step("trim_artwork"), ctx.cfg().general.art_cache_mb * 1_048_576);
+        let gone = tokio::task::spawn_blocking(move || span.in_scope(|| art::trim(&art::dir(), limit))).await?;
         if gone > 0 {
             tracing::debug!(gone, "trimmed the artwork kept on disk");
         }

crates/ipx/src/web.rs

@@ -2287,7 +2287,7 @@ async fn art(State(state): State<WebState>, Query(q): Query<ArtQuery>) -> Respon
     // Kept only once it was let through below, so a kept image needs no asking again (#272).
     let keep = state.ctx.cfg().general.art_cache_mb > 0;
     let file = crate::art::path(&q.u);
-    if keep && let Some((kind, body)) = step("art_cached").in_scope(|| crate::art::read(&file)) {
+    if keep && let Some((kind, body)) = kept_art(&file).await {
         crate::metrics::STATS.art_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         return art_response(kind, body);
     }
@@ -2301,6 +2301,15 @@ async fn art(State(state): State<WebState>, Query(q): Query<ArtQuery>) -> Respon
     if let Some(why) = crate::art::failed(&q.u) {
         return (StatusCode::BAD_GATEWAY, why).into_response();
     }
+    // One fetch an address at a time: a page asks for fifty images at once, and a cover two of
+    // them name was fetched twice on a cold cache (#349). The second waits, then finds it kept.
+    let gate = ART_FETCHES.lock().unwrap().entry(q.u.clone()).or_default().clone();
+    let _turn = gate.lock().await;
+    let _done = ArtFetchDone(&q.u);
+    if keep && let Some((kind, body)) = kept_art(&file).await {
+        crate::metrics::STATS.art_hits.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        return art_response(kind, body);
+    }
     crate::metrics::STATS.art_misses.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
     let got = async {
         let mut r = state.ctx.client.get(&q.u).timeout(crate::feed::FEED_TIMEOUT).send().await?.error_for_status()?;
@@ -2316,8 +2325,15 @@ async fn art(State(state): State<WebState>, Query(q): Query<ArtQuery>) -> Respon
     };
     match got.instrument(step("art_fetch")).await {
         Ok((kind, body)) => {
-            if keep && let Err(e) = step("art_keep").in_scope(|| crate::art::write(&file, &kind, &body)) {
+            if keep {
-                tracing::debug!(error = %e, "could not keep the artwork");
+                let (span, file, kind, body) = (step("art_keep"), file.clone(), kind.clone(), body.clone());
+                match tokio::task::spawn_blocking(move || span.in_scope(|| crate::art::write(&file, &kind, &body)))
+                    .await
+                {
+                    Ok(Ok(())) => {}
+                    Ok(Err(e)) => tracing::debug!(error = %e, "could not keep the artwork"),
+                    Err(e) => tracing::debug!(error = %e, "could not keep the artwork"),
+                }
             }
             art_response(kind, body)
         }
@@ -2331,6 +2347,31 @@ async fn art(State(state): State<WebState>, Query(q): Query<ArtQuery>) -> Respon
     }
 }
 
+/// The fetches of artwork under way, by address, each a lock the requests for it take in turn.
+static ART_FETCHES: std::sync::LazyLock<
+    std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
+> = std::sync::LazyLock::new(Default::default);
+
+/// Forgets an address's lock once nobody else holds it, however the request ended.
+struct ArtFetchDone<'a>(&'a str);
+
+impl Drop for ArtFetchDone<'_> {
+    fn drop(&mut self) {
+        let mut fetches = ART_FETCHES.lock().unwrap();
+        // Two: the map's and this request's own.
+        if fetches.get(self.0).is_some_and(|g| Arc::strong_count(g) <= 2) {
+            fetches.remove(self.0);
+        }
+    }
+}
+
+/// Artwork kept on disk, read off the async threads: the cache is on FUSE, and four calls a
+/// hit, fifty images a page, held a worker each, 1 s at p90 for a hit (#345).
+async fn kept_art(file: &std::path::Path) -> Option<(String, Vec<u8>)> {
+    let (span, file) = (step("art_cached"), file.to_owned());
+    tokio::task::spawn_blocking(move || span.in_scope(|| crate::art::read(&file))).await.ok().flatten()
+}
+
 /// A type that says nothing about what the body is, which some publishers send with real artwork:
 /// Spotify's CloudFront `application/octet-stream`, pippa.io none at all (#279).
 fn untyped(said: &str) -> bool {