Harden the backend: read pool, tighter limits, crawler guards

Feed, likes, artist, track and the session lookup run on four
read-only SQLite connections, so a slow feed no longer blocks writes.
Reads are capped at 120 a minute per user and the audio and image
proxy at 3000 a minute per IP. Session rotation happens once when
requests race, and the page, assets and legal pages skip the session
lookup. Events and follows for unknown tracks or artists return 404,
and radio only explores artists already in the catalogue.

The crawler queue is capped at 20k jobs, an empty home page counts as
a failure, and 25 failures in a row log an error that pl8list may
have changed its API. A guest who signs in with "stay connected" gets
up to 200 likes and follows saved to pl8list. The old Python server is
removed; it lives on in 54a436a.

backend/src/crawler.rs

@@ -52,6 +52,11 @@ const REFRESH_AFTER: f64 = 86400.0; // re-check an artist's newest page for uplo
 const RESEED_EVERY: Duration = Duration::from_secs(600);
 /// An artist radio crawls its artist's neighbourhood at most this often.
 const EXPLORE_EVERY: Duration = Duration::from_secs(600);
+/// Jobs waiting at most. Past this only seed and feedback jobs get in, so nothing
+/// from outside can grow the queue without end.
+const MAX_QUEUE: usize = 20_000;
+/// This many failed jobs in a row means pl8list is down or has changed its API.
+const FAILURES_ALARM: u32 = 25;
 
 #[derive(Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Debug)]
 enum Job {
@@ -95,6 +100,9 @@ impl Crawler {
 
     fn push(&self, priority: i64, job: Job) {
         let mut q = self.queue.lock().unwrap_or_else(|e| e.into_inner());
+        if q.heap.len() >= MAX_QUEUE && priority > P_SEED {
+            return;
+        }
         if !q.queued.insert(job.clone()) {
             return;
         }
@@ -153,7 +161,12 @@ impl Crawler {
 
     /// An artist radio was opened: crawl the artist, and the related pages of their
     /// best tracks that were not read yet, to fill in the artist graph around them.
+    /// Only artists already in the catalogue: a handle typed into a URL queues nothing.
     pub async fn explore_artist(&self, handle: String) {
+        let h = handle.clone();
+        if !self.db.read(move |c| db::artist_known(c, &h)).await.unwrap_or(false) {
+            return;
+        }
         {
             let mut explored = self.explored.lock().unwrap_or_else(|e| e.into_inner());
             let now = Instant::now();
@@ -187,10 +200,26 @@ impl Crawler {
     async fn run(&self) {
         self.seed();
         let mut last_seed = Instant::now();
+        let mut failures = 0u32;
         loop {
             let job = self.pop().await;
-            if let Err(e) = self.run_job(job.clone()).await {
-                tracing::warn!("crawl {job:?} failed: {e:#}");
+            match self.run_job(job.clone()).await {
+                Ok(()) => {
+                    if failures >= FAILURES_ALARM {
+                        tracing::info!("crawler: pl8list answers again after {failures} failed jobs");
+                    }
+                    failures = 0;
+                }
+                Err(e) => {
+                    tracing::warn!("crawl {job:?} failed: {e:#}");
+                    failures += 1;
+                    if failures == FAILURES_ALARM {
+                        tracing::error!(
+                            "crawler: {failures} jobs in a row failed. pl8list is down, or its routes changed \
+                             (a move from Remix `?_data=` to React Router `.data` would look like this)"
+                        );
+                    }
+                }
             }
             if !self.busy().await {
                 tokio::time::sleep(IDLE_DELAY).await;
@@ -308,10 +337,18 @@ impl Crawler {
         match job {
             Job::Home => {
                 let d = self.pl8.home().await?.unwrap_or(Value::Null);
+                let mut found = 0;
                 for section in ["bestSellers", "recommended"] {
-                    self.upsert(normalize::items(d.get(section).and_then(|s| s.get("items"))), None).await?;
+                    let rows = normalize::items(d.get(section).and_then(|s| s.get("items")));
+                    found += rows.len();
+                    self.upsert(rows, None).await?;
                 }
-                self.ticker_events(d.get("tickerEvents")).await
+                self.ticker_events(d.get("tickerEvents")).await?;
+                // The home page always lists tracks; none means its shape changed.
+                if found == 0 {
+                    anyhow::bail!("home page has no tracks in bestSellers or recommended; did pl8list change its data?");
+                }
+                Ok(())
             }
             Job::Ticker => {
                 let d = self.pl8.activity_ticker().await?.unwrap_or(Value::Null);

backend/src/db.rs

@@ -5,13 +5,16 @@
 
 use std::{
     path::Path,
-    sync::{Arc, Mutex},
+    sync::{
+        Arc, Mutex,
+        atomic::{AtomicUsize, Ordering},
+    },
     time::{SystemTime, UNIX_EPOCH},
 };
 
 use base64::Engine;
 use rand::Rng;
-use rusqlite::{Connection, OptionalExtension, Row, params, params_from_iter};
+use rusqlite::{Connection, OpenFlags, OptionalExtension, Row, params, params_from_iter};
 use serde::Serialize;
 use sha2::{Digest, Sha256};
 
@@ -134,8 +137,17 @@ pub fn now() -> f64 {
     SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64()
 }
 
+/// Read-only connections next to the one that writes. With WAL they read while it
+/// writes, so a slow feed query does not hold up every other request.
+const READERS: usize = 4;
+
+/// One connection that writes, and a few that only read.
 #[derive(Clone)]
-pub struct Db(Arc<Mutex<Connection>>);
+pub struct Db {
+    writer: Arc<Mutex<Connection>>,
+    readers: Arc<Vec<Mutex<Connection>>>,
+    next: Arc<AtomicUsize>,
+}
 
 impl Db {
     pub fn open(path: &Path) -> anyhow::Result<Self> {
@@ -148,16 +160,22 @@ impl Db {
         if tagged > 0 {
             tracing::info!("tagged {tagged} tracks");
         }
-        Ok(Db(Arc::new(Mutex::new(conn))))
+        let mut readers = Vec::with_capacity(READERS);
+        for _ in 0..READERS {
+            let r = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX)?;
+            r.busy_timeout(std::time::Duration::from_secs(5))?;
+            readers.push(Mutex::new(r));
+        }
+        Ok(Db { writer: Arc::new(Mutex::new(conn)), readers: Arc::new(readers), next: Arc::new(AtomicUsize::new(0)) })
     }
 
-    /// Run `f` on the connection off the async runtime.
+    /// Run `f` on the writing connection off the async runtime.
     pub async fn call<T, F>(&self, f: F) -> anyhow::Result<T>
     where
         T: Send + 'static,
         F: FnOnce(&mut Connection) -> rusqlite::Result<T> + Send + 'static,
     {
-        let db = self.0.clone();
+        let db = self.writer.clone();
         let out = tokio::task::spawn_blocking(move || {
             let mut conn = db.lock().unwrap_or_else(|e| e.into_inner());
             f(&mut conn)
@@ -165,6 +183,25 @@ impl Db {
         .await??;
         Ok(out)
     }
+
+    /// Run a read-only `f` on one of the reading connections: a free one if there
+    /// is one, else the next in turn.
+    pub async fn read<T, F>(&self, f: F) -> anyhow::Result<T>
+    where
+        T: Send + 'static,
+        F: FnOnce(&Connection) -> rusqlite::Result<T> + Send + 'static,
+    {
+        let readers = self.readers.clone();
+        let start = self.next.fetch_add(1, Ordering::Relaxed);
+        let out = tokio::task::spawn_blocking(move || {
+            let n = readers.len();
+            let free = (0..n).find_map(|i| readers[(start + i) % n].try_lock().ok());
+            let conn = free.unwrap_or_else(|| readers[start % n].lock().unwrap_or_else(|e| e.into_inner()));
+            f(&conn)
+        })
+        .await??;
+        Ok(out)
+    }
 }
 
 fn migrate(conn: &Connection) -> rusqlite::Result<()> {
@@ -298,6 +335,14 @@ pub fn artist(conn: &Connection, handle: &str) -> rusqlite::Result<Option<Artist
     .optional()
 }
 
+/// Whether the catalogue knows this artist, from a track or a listing.
+pub fn artist_known(conn: &Connection, handle: &str) -> rusqlite::Result<bool> {
+    conn.prepare_cached(
+        "SELECT 1 WHERE EXISTS (SELECT 1 FROM tracks WHERE handle=?1) OR EXISTS (SELECT 1 FROM artists WHERE handle=?1)",
+    )?
+    .exists([handle])
+}
+
 /// An artist's most common style tags; tempo tags left out.
 pub fn artist_tags(conn: &Connection, handle: &str, n: usize) -> rusqlite::Result<Vec<String>> {
     conn.prepare_cached(
@@ -564,6 +609,25 @@ pub fn create_guest(conn: &Connection) -> rusqlite::Result<(User, String)> {
     Ok((user, token))
 }
 
+/// What a guest would add to a pl8list library: their liked tracks (`[handle, slug]`,
+/// newest first, not unliked since) and followed artists (`[handle]`), at most `max`.
+pub fn guest_library(conn: &Connection, guest_id: i64, max: usize) -> rusqlite::Result<Vec<Vec<String>>> {
+    let liked: Vec<Vec<String>> = conn
+        .prepare(
+            "SELECT t.handle, t.slug FROM events e JOIN tracks t ON t.id = e.track_id \
+             WHERE e.user_id=?1 AND e.kind='like' AND NOT EXISTS (SELECT 1 FROM events u \
+               WHERE u.user_id=?1 AND u.track_id=e.track_id AND u.kind='unlike' AND u.ts >= e.ts) \
+             GROUP BY e.track_id ORDER BY MAX(e.ts) DESC",
+        )?
+        .query_map([guest_id], |r| Ok(vec![r.get(0)?, r.get(1)?]))?
+        .collect::<rusqlite::Result<_>>()?;
+    let followed: Vec<Vec<String>> = conn
+        .prepare("SELECT handle FROM follows WHERE user_id=? ORDER BY ts DESC")?
+        .query_map([guest_id], |r| Ok(vec![r.get(0)?]))?
+        .collect::<rusqlite::Result<_>>()?;
+    Ok(followed.into_iter().chain(liked).take(max).collect())
+}
+
 /// A guest signed in with pl8list: what they played, liked and followed moves to
 /// the account, and the guest is gone.
 pub fn merge_guest(conn: &mut Connection, guest_id: i64, user_id: i64) -> rusqlite::Result<()> {
@@ -601,6 +665,7 @@ pub fn create_session(conn: &Connection, user_id: i64) -> rusqlite::Result<Strin
     Ok(token)
 }
 
+#[derive(Clone)]
 pub struct SessionHit {
     pub user: User,
     pub created_at: f64,
@@ -609,9 +674,10 @@ pub struct SessionHit {
     pub synced_at: Option<f64>,
 }
 
+/// The session behind a token, read-only. `touch_session` then records the use.
 pub fn session_user(conn: &Connection, token: &str) -> rusqlite::Result<Option<SessionHit>> {
     let now = now();
-    let hit = conn
+    conn
         .query_row(
             "SELECT u.*, s.created_at AS s_created, s.expires_at AS s_expires, s.pl8_link, s.pl8_synced_at \
              FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.token_hash=? AND s.expires_at > ?",
@@ -626,31 +692,46 @@ pub fn session_user(conn: &Connection, token: &str) -> rusqlite::Result<Option<S
                 })
             },
         )
-        .optional()?;
-    if let Some(h) = &hit {
-        if h.user.last_seen.unwrap_or(0.0) < now - 60.0 {
-            conn.execute("UPDATE users SET last_seen=? WHERE id=?", params![now, h.user.id])?;
-        }
-        // Sliding expiry, written at most once a day.
-        if h.expires_at < now + (SESSION_IDLE_DAYS - 1.0) * 86400.0 {
-            conn.execute(
-                "UPDATE sessions SET expires_at=? WHERE token_hash=? AND expires_at > ?",
-                params![now + SESSION_IDLE_DAYS * 86400.0, hash_token(token), now + 120.0],
-            )?;
-        }
+        .optional()
+}
+
+impl SessionHit {
+    /// Whether `touch_session` has anything to write: last seen at most once a
+    /// minute, the sliding expiry at most once a day.
+    pub fn stale(&self) -> bool {
+        let now = now();
+        self.user.last_seen.unwrap_or(0.0) < now - 60.0 || self.expires_at < now + (SESSION_IDLE_DAYS - 1.0) * 86400.0
     }
-    Ok(hit)
+}
+
+pub fn touch_session(conn: &Connection, token: &str, hit: &SessionHit) -> rusqlite::Result<()> {
+    let now = now();
+    if hit.user.last_seen.unwrap_or(0.0) < now - 60.0 {
+        conn.execute("UPDATE users SET last_seen=? WHERE id=?", params![now, hit.user.id])?;
+    }
+    if hit.expires_at < now + (SESSION_IDLE_DAYS - 1.0) * 86400.0 {
+        // Not a session that was just rotated away: that one stays short.
+        conn.execute(
+            "UPDATE sessions SET expires_at=? WHERE token_hash=? AND expires_at > ?",
+            params![now + SESSION_IDLE_DAYS * 86400.0, hash_token(token), now + 120.0],
+        )?;
+    }
+    Ok(())
 }
 
 /// Replace a session with a new token. The old one keeps working for a minute so
-/// requests already in flight do not fail.
-pub fn rotate_session(conn: &Connection, old_token: &str, user_id: i64) -> rusqlite::Result<String> {
-    let token = create_session(conn, user_id)?;
-    conn.execute(
-        "UPDATE sessions SET expires_at=MIN(expires_at, ?) WHERE token_hash=?",
-        params![now() + 60.0, hash_token(old_token)],
+/// requests already in flight do not fail. Returns `None` when another request
+/// rotated it already: then that request's response carries the new token.
+pub fn rotate_session(conn: &Connection, old_token: &str, user_id: i64) -> rusqlite::Result<Option<String>> {
+    let now = now();
+    let retired = conn.execute(
+        "UPDATE sessions SET expires_at=? WHERE token_hash=? AND user_id=? AND expires_at > ?",
+        params![now + 60.0, hash_token(old_token), user_id, now + 60.0],
     )?;
-    Ok(token)
+    if retired == 0 {
+        return Ok(None);
+    }
+    create_session(conn, user_id).map(Some)
 }
 
 /// Store (or replace) the encrypted pl8list session of one device.
@@ -856,8 +937,11 @@ mod tests {
         assert_eq!(stored, hash_token(&token));
         assert_eq!(session_user(&c, &token).unwrap().unwrap().user.id, 7);
 
-        let new = rotate_session(&c, &token, 7).unwrap();
+        let new = rotate_session(&c, &token, 7).unwrap().unwrap();
         assert!(session_user(&c, &new).unwrap().is_some());
+        // A second request with the old token, running alongside, does not rotate again.
+        assert_eq!(rotate_session(&c, &token, 7).unwrap(), None);
+        assert_eq!(c.query_row("SELECT COUNT(*) FROM sessions", [], |r| r.get::<_, i64>(0)).unwrap(), 2);
         let old_expiry: f64 = c
             .query_row("SELECT expires_at FROM sessions WHERE token_hash=?", [hash_token(&token)], |r| r.get(0))
             .unwrap();

backend/src/http/app.rs

@@ -52,13 +52,22 @@ async fn resync_if_due(app: &App, current: &CurrentUser) {
     }
 }
 
+/// The signed-in user, within the limit for catalogue reads.
+fn reader(app: &App, current: &CurrentUser) -> Result<db::User, Response> {
+    let user = current.require()?;
+    if !app.limits.reads.allow(&user.id.to_string()) {
+        return Err(json_error(StatusCode::TOO_MANY_REQUESTS, "slow down a little"));
+    }
+    Ok(user.clone())
+}
+
 fn num(q: &HashMap<String, String>, name: &str) -> Option<f64> {
     q.get(name).and_then(|v| v.parse::<f64>().ok()).filter(|v| v.is_finite())
 }
 
 pub async fn feed(State(app): St, Extension(current): Extension<CurrentUser>, Query(q): Query<HashMap<String, String>>) -> Res {
-    let user = match current.require() {
-        Ok(u) => u.clone(),
+    let user = match reader(&app, &current) {
+        Ok(u) => u,
         Err(r) => return Ok(r),
     };
     let artist = q.get("artist").map(String::as_str).filter(|h| valid_handle(h));
@@ -78,11 +87,18 @@ pub async fn feed(State(app): St, Extension(current): Extension<CurrentUser>, Qu
     resync_if_due(&app, &current).await;
     let deadline = Instant::now() + FEED_WAIT;
     let uid = user.id;
-    let mut tracks = loop {
+    let (mut tracks, liked) = loop {
         let mode = mode.clone();
-        let tracks = app.db.call(move |c| ranker::pick(c, uid, n, filters, &*mode.ranker(c)?)).await?;
+        let (tracks, liked) = app
+            .db
+            .read(move |c| {
+                let taste = ranker::taste(c, uid)?;
+                let tracks = ranker::pick_with(c, uid, n, filters, &*mode.ranker(c)?, &taste)?;
+                Ok((tracks, taste.liked_ids.len()))
+            })
+            .await?;
         if !tracks.is_empty() || Instant::now() >= deadline {
-            break tracks;
+            break (tracks, liked);
         }
         tokio::time::sleep(Duration::from_millis(1500)).await;
     };
@@ -95,7 +111,7 @@ pub async fn feed(State(app): St, Extension(current): Extension<CurrentUser>, Qu
                 db::add_event(c, uid, id, "served", None)?;
             }
             let mut s = db::stats(c, uid)?;
-            s.liked = ranker::taste(c, uid)?.liked_ids.len();
+            s.liked = liked;
             Ok(s)
         })
         .await?;
@@ -103,11 +119,11 @@ pub async fn feed(State(app): St, Extension(current): Extension<CurrentUser>, Qu
 }
 
 pub async fn likes(State(app): St, Extension(current): Extension<CurrentUser>) -> Res {
-    let uid = match current.require() {
+    let uid = match reader(&app, &current) {
         Ok(u) => u.id,
         Err(r) => return Ok(r),
     };
-    let mut tracks = app.db.call(move |c| ranker::likes(c, uid)).await?;
+    let mut tracks = app.db.read(move |c| ranker::likes(c, uid)).await?;
     mark_followed(&app, uid, &mut tracks).await?;
     Ok((no_store(), Json(json!({ "tracks": tracks }))).into_response())
 }
@@ -129,24 +145,27 @@ pub async fn event(State(app): St, Extension(current): Extension<CurrentUser>, J
     if !app.limits.events.allow(&uid.to_string()) {
         return Ok(StatusCode::TOO_MANY_REQUESTS.into_response());
     }
+    // Only tracks in the catalogue: anything else would be a row nothing reads.
+    let Some((handle, slug)) = app.db.read(move |c| db::track_path(c, track_id)).await? else {
+        return Ok(StatusCode::NOT_FOUND.into_response());
+    };
     let kind = kind.to_string();
     let k = kind.clone();
     app.db.call(move |c| db::add_event(c, uid, track_id, &k, seconds)).await?;
     if kind == "like" || kind == "complete" {
         app.crawler.on_positive(track_id).await;
     }
-    if kind == "like" && app.limits.pl8_writes.allow(&uid.to_string()) {
-        let row = app.db.call(move |c| db::track_path(c, track_id)).await?;
-        if let (Some((handle, slug)), Some((sealer, session))) = (row, app.connected(&current).await) {
-            app.sync.push_like(sealer, session, handle, slug);
-        }
+    if kind == "like" && app.limits.pl8_writes.allow(&uid.to_string())
+        && let Some((sealer, session)) = app.connected(&current).await
+    {
+        app.sync.push_like(sealer, session, handle, slug);
     }
     Ok((no_store(), Json(json!({ "ok": true }))).into_response())
 }
 
 /// One track, for a link into the app (`/?t=handle/slug`).
 pub async fn track(State(app): St, Extension(current): Extension<CurrentUser>, Query(q): Query<HashMap<String, String>>) -> Res {
-    let uid = match current.require() {
+    let uid = match reader(&app, &current) {
         Ok(u) => u.id,
         Err(r) => return Ok(r),
     };
@@ -155,7 +174,7 @@ pub async fn track(State(app): St, Extension(current): Extension<CurrentUser>, Q
     let (Some(handle), Some(slug)) = (handle, slug) else {
         return Ok(StatusCode::BAD_REQUEST.into_response());
     };
-    let Some(track) = app.db.call(move |c| ranker::track_by_path(c, &handle, &slug)).await? else {
+    let Some(track) = app.db.read(move |c| ranker::track_by_path(c, &handle, &slug)).await? else {
         return Ok(json_error(StatusCode::NOT_FOUND, "no such track here (yet)"));
     };
     let mut tracks = [track];
@@ -172,7 +191,7 @@ const FEW_NEIGHBOURS: usize = 3;
 /// Starts crawling the neighbourhood when it is new, so the sheet can poll while
 /// `exploring` is true.
 pub async fn artist(State(app): St, Extension(current): Extension<CurrentUser>, Query(q): Query<HashMap<String, String>>) -> Res {
-    let uid = match current.require() {
+    let uid = match reader(&app, &current) {
         Ok(u) => u.id,
         Err(r) => return Ok(r),
     };
@@ -182,7 +201,7 @@ pub async fn artist(State(app): St, Extension(current): Extension<CurrentUser>,
     app.crawler.explore_artist(handle.clone()).await;
     let body = app
         .db
-        .call(move |c| {
+        .read(move |c| {
             let Some(artist) = db::artist(c, &handle)? else { return Ok(None) };
             let mut neighbours = vec![];
             for (h, weight) in ranker::scaled_neighbours(c, &handle)?.into_iter().take(SHEET_NEIGHBOURS) {
@@ -226,6 +245,10 @@ pub async fn follow(State(app): St, Extension(current): Extension<CurrentUser>,
         return Ok(StatusCode::TOO_MANY_REQUESTS.into_response());
     }
     let h = handle.clone();
+    if on && !app.db.read(move |c| db::artist_known(c, &h)).await? {
+        return Ok(json_error(StatusCode::NOT_FOUND, "unknown artist"));
+    }
+    let h = handle.clone();
     let added = app
         .db
         .call(move |c| if on { db::add_follows(c, uid, &[h], "local") } else { db::unfollow(c, uid, &h).map(|_| 0) })

backend/src/http/auth.rs

@@ -18,7 +18,7 @@ use axum::{
 use serde_json::{Value, json};
 
 use super::{App, CurrentUser, LOGIN_COOKIE, Peer, Res, SESSION_COOKIE, St, client_ip, json_error, no_store, security};
-use crate::{db, pl8::login::LoginFlow};
+use crate::{db, pl8::login::LoginFlow, sync};
 
 const LOGIN_TTL: Duration = Duration::from_secs(900);
 const MAX_PENDING: usize = 500;
@@ -164,17 +164,22 @@ pub async fn verify(State(app): St, Extension(current): Extension<CurrentUser>,
     }
     // A guest signing in keeps what they played, liked and followed.
     let guest = current.user.as_ref().filter(|g| g.is_guest()).map(|g| g.id);
-    let token = app
+    let (token, guest_library) = app
         .db
         .call(move |c| {
             if let Some(t) = previous {
                 db::delete_session(c, &t)?;
             }
             db::upsert_user(c, u.id, &u.handle, u.name.as_deref(), u.img.as_deref())?;
+            // Staying connected, the guest's likes and follows go to pl8list too.
+            let mut library = vec![];
             if let Some(g) = guest {
+                if stay {
+                    library = db::guest_library(c, g, sync::MAX_GUEST_PUSH)?;
+                }
                 db::merge_guest(c, g, u.id)?;
             }
-            db::create_session(c, u.id)
+            Ok((db::create_session(c, u.id)?, library))
         })
         .await?;
     if stay {
@@ -182,7 +187,7 @@ pub async fn verify(State(app): St, Extension(current): Extension<CurrentUser>,
         let blob = sealer.seal(&pl8_session);
         let hash = sealer.token_hash.clone();
         app.db.call(move |c| db::set_link(c, &hash, &blob, Some(db::now()))).await?;
-        app.sync.linked(sealer, pl8_session);
+        app.sync.linked_with(sealer, pl8_session, guest_library);
     } else {
         app.sync.once(user.id, pl8_session);
     }

backend/src/http/mod.rs

@@ -49,7 +49,12 @@ pub struct Limits {
     /// Every start and resend makes pl8list send a mail; cap them for the whole server.
     pub login_global: RateLimiter,
     pub events: RateLimiter,
+    /// Feed, artist, track and likes: each runs queries over the catalogue.
+    pub reads: RateLimiter,
     pub proxy: RateLimiter,
+    /// The proxy per address too: guests are cheap to make, so a per-user limit
+    /// alone would not hold.
+    pub proxy_ip: RateLimiter,
     /// Saves to a user's pl8list library (likes and follows pushed from here).
     pub pl8_writes: RateLimiter,
     /// The resync button.
@@ -86,7 +91,9 @@ impl App {
                 login_email: RateLimiter::new(3, 900),
                 login_global: RateLimiter::new(30, 600),
                 events: RateLimiter::new(600, 600),
+                reads: RateLimiter::new(120, 60),
                 proxy: RateLimiter::new(1200, 60),
+                proxy_ip: RateLimiter::new(3000, 60),
                 pl8_writes: RateLimiter::new(60, 600),
                 resync: RateLimiter::new(1, 120),
                 guests: RateLimiter::new(10, 3600),
@@ -112,7 +119,9 @@ impl App {
         }
         self.logins.prune();
         self.segments.prune();
-        self.limits.guests.prune();
+        for l in [&self.limits.guests, &self.limits.reads, &self.limits.proxy, &self.limits.proxy_ip, &self.limits.events] {
+            l.prune();
+        }
     }
 
     /// This device's stored pl8list session, if it stays connected. A blob that no
@@ -190,7 +199,13 @@ async fn session(State(app): St, mut req: Request, next: Next) -> Response {
     let mut current = CurrentUser { user: None, token: token.clone(), link: None, synced_at: None };
     if let Some(t) = token {
         let lookup = t.clone();
-        if let Ok(Some(hit)) = app.db.call(move |c| db::session_user(c, &lookup)).await {
+        // Looked up on a reader; the writer only sees the rare request that has to
+        // record something.
+        if let Ok(Some(hit)) = app.db.read(move |c| db::session_user(c, &lookup)).await {
+            if hit.stale() {
+                let (tt, h) = (t.clone(), hit.clone());
+                let _ = app.db.call(move |c| db::touch_session(c, &tt, &h)).await;
+            }
             let mut new_token = None;
             current.link = hit.link.clone();
             current.synced_at = hit.synced_at;
@@ -201,7 +216,7 @@ async fn session(State(app): St, mut req: Request, next: Next) -> Response {
                 let key = &app.sync.key;
                 let synced = hit.synced_at;
                 new_token = match app.db.call(move |c| db::rotate_session(c, &old, uid)).await {
-                    Ok(nt) => {
+                    Ok(Some(nt)) => {
                         if let Some(session) = reopened {
                             let new = key.sealer(&nt, uid);
                             let blob = new.seal(&session);
@@ -220,7 +235,9 @@ async fn session(State(app): St, mut req: Request, next: Next) -> Response {
                         }
                         Some(nt)
                     }
-                    Err(_) => None,
+                    // Rotated by a request running alongside this one, or failed: the
+                    // old token works for another minute either way.
+                    Ok(None) | Err(_) => None,
                 };
             } else if legacy {
                 new_token = Some(t.clone());
@@ -311,14 +328,17 @@ async fn healthz() -> Response {
 
 pub fn router(app: Arc<App>) -> Router {
     let secure = app.cfg.secure;
-    Router::new()
+    // The page itself, its assets and the legal pages need no session: no database
+    // lookup for each of them.
+    let public = Router::new()
         .route("/", get(index))
         .route("/static/{*path}", get(static_route))
         .route("/sw.js", get(service_worker))
         .route("/healthz", get(healthz))
         .route("/impressum", get(legal::impressum))
         .route("/datenschutz", get(legal::datenschutz))
-        .route("/privacy", get(legal::datenschutz))
+        .route("/privacy", get(legal::datenschutz));
+    let api = Router::new()
         .route("/auth/start", post(auth::start))
         .route("/auth/resend", post(auth::resend))
         .route("/auth/verify", post(auth::verify))
@@ -339,7 +359,9 @@ pub fn router(app: Arc<App>) -> Router {
         .route("/api/{handle}/{slug}/hls/playlist.m3u8", get(proxy::playlist))
         .route("/api/{handle}/{slug}/hls/segment/{n}", get(proxy::segment))
         .route("/img/{id}", get(proxy::image))
-        .layer(middleware::from_fn_with_state(app.clone(), session))
+        .layer(middleware::from_fn_with_state(app.clone(), session));
+    public
+        .merge(api)
         .layer(middleware::from_fn_with_state(app.clone(), origin_check))
         .layer(DefaultBodyLimit::max(4096))
         .layer(TimeoutLayer::with_status_code(StatusCode::REQUEST_TIMEOUT, Duration::from_secs(30)))

backend/src/http/proxy.rs

@@ -12,6 +12,7 @@
 
 use std::{
     collections::HashMap,
+    net::SocketAddr,
     sync::Mutex,
     time::{Duration, Instant},
 };
@@ -19,7 +20,7 @@ use std::{
 use axum::{
     Extension,
     extract::{Path, Query, State},
-    http::{StatusCode, header},
+    http::{HeaderMap, StatusCode, header},
     response::{IntoResponse, Response},
 };
 use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
@@ -28,7 +29,7 @@ use futures_util::StreamExt;
 use serde::Deserialize;
 use serde_json::Value;
 
-use super::{App, CurrentUser, Res, St};
+use super::{App, CurrentUser, Peer, Res, St, client_ip};
 use crate::cache::{self, IMAGE_TYPES, SEGMENT_TYPES};
 
 const MAX_PLAYLIST: usize = 256 << 10;
@@ -82,9 +83,9 @@ fn token_upload(token: &str) -> Option<String> {
     v.get("u")?.as_str().filter(|u| safe_segment(u, 128)).map(String::from)
 }
 
-fn require_user(app: &App, current: &CurrentUser) -> Result<(), Response> {
+fn require_user(app: &App, current: &CurrentUser, headers: &HeaderMap, peer: SocketAddr) -> Result<(), Response> {
     let user = current.require()?;
-    if !app.limits.proxy.allow(&user.id.to_string()) {
+    if !app.limits.proxy.allow(&user.id.to_string()) || !app.limits.proxy_ip.allow(&client_ip(app, headers, peer)) {
         return Err(StatusCode::TOO_MANY_REQUESTS.into_response());
     }
     Ok(())
@@ -133,10 +134,12 @@ pub struct PlaylistQuery {
 pub async fn playlist(
     State(app): St,
     Extension(current): Extension<CurrentUser>,
+    peer: Peer,
+    headers: HeaderMap,
     Path((handle, slug)): Path<(String, String)>,
     Query(q): Query<PlaylistQuery>,
 ) -> Res {
-    if let Err(r) = require_user(&app, &current) {
+    if let Err(r) = require_user(&app, &current, &headers, peer.0) {
         return Ok(r);
     }
     if !safe_segment(&handle, 64) || !safe_segment(&slug, 160) || q.v.as_deref().is_some_and(|v| !safe_segment(v, 128)) {
@@ -180,10 +183,12 @@ pub struct SegmentQuery {
 pub async fn segment(
     State(app): St,
     Extension(current): Extension<CurrentUser>,
+    peer: Peer,
+    headers: HeaderMap,
     Path((handle, slug, n)): Path<(String, String, String)>,
     Query(q): Query<SegmentQuery>,
 ) -> Res {
-    if let Err(r) = require_user(&app, &current) {
+    if let Err(r) = require_user(&app, &current, &headers, peer.0) {
         return Ok(r);
     }
     let token = q.st.unwrap_or_default();
@@ -219,8 +224,8 @@ pub async fn segment(
     Ok(media(ctype, body, "private, max-age=86400"))
 }
 
-pub async fn image(State(app): St, Extension(current): Extension<CurrentUser>, Path(id): Path<String>) -> Res {
-    if let Err(r) = require_user(&app, &current) {
+pub async fn image(State(app): St, Extension(current): Extension<CurrentUser>, peer: Peer, headers: HeaderMap, Path(id): Path<String>) -> Res {
+    if let Err(r) = require_user(&app, &current, &headers, peer.0) {
         return Ok(r);
     }
     if id.len() > 64 || id.is_empty() || !id.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-') {

backend/src/http/tests.rs

@@ -301,6 +301,7 @@ async fn connected_device_syncs_library_and_pushes_likes_and_follows() {
 
     let res = r.clone().oneshot(req("POST", "/local/event", Some(r#"{"track_id":77,"kind":"like"}"#), Some(&session), "203.0.113.9")).await.unwrap();
     assert_eq!(res.status(), StatusCode::OK);
+    known_track(&app, 78, "artist3").await;
     let res = r.clone().oneshot(req("POST", "/local/follow", Some(r#"{"handle":"artist3","follow":true}"#), Some(&session), "203.0.113.9")).await.unwrap();
     assert_eq!(res.status(), StatusCode::OK);
     assert!(wait_for(&pl8, |q| q.url.path() == "/api/artist1/song/add-to-library").await);
@@ -507,10 +508,17 @@ async fn artist_sheet_and_radio_reasons() {
     assert_eq!(why[&3], serde_json::json!({ "kind": "via", "handle": "b", "name": "B" }));
 }
 
+/// A track (and so its artist) in the catalogue; events and follows only take known ones.
+async fn known_track(app: &crate::http::App, id: i64, handle: &str) {
+    let t = crate::db::Track { id, handle: handle.into(), artist_name: handle.into(), slug: format!("track-{id}"), title: "T".into(), ..Default::default() };
+    app.db.call(move |c| crate::db::upsert_tracks(c, &[t], None)).await.unwrap();
+}
+
 #[tokio::test]
 async fn guests_listen_and_keep_it_all_when_they_sign_in() {
     let pl8 = fake_pl8list(1).await;
     let (r, app, _) = setup("guest", &pl8.uri()).await;
+    known_track(&app, 7, "motus").await;
     let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.20")).await.unwrap();
     assert_eq!(res.status(), StatusCode::OK);
     let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf");
@@ -521,6 +529,11 @@ async fn guests_listen_and_keep_it_all_when_they_sign_in() {
         let res = r.clone().oneshot(req("POST", uri, Some(body), Some(&guest), "203.0.113.20")).await.unwrap();
         assert_eq!(res.status(), StatusCode::OK, "{uri}");
     }
+    // Nothing the catalogue does not know.
+    for (uri, body) in [("/local/event", r#"{"track_id":999,"kind":"like"}"#), ("/local/follow", r#"{"handle":"nobody-here","follow":true}"#)] {
+        let res = r.clone().oneshot(req("POST", uri, Some(body), Some(&guest), "203.0.113.20")).await.unwrap();
+        assert_eq!(res.status(), StatusCode::NOT_FOUND, "{uri}");
+    }
 
     // Signing in on the same device takes the guest's likes and follows along.
     let res = r.clone().oneshot(req("POST", "/auth/start", Some(r#"{"email":"a@b.de"}"#), None, "203.0.113.20")).await.unwrap();
@@ -550,6 +563,7 @@ async fn guests_listen_and_keep_it_all_when_they_sign_in() {
 #[tokio::test]
 async fn guests_leave_nothing_behind_and_are_capped_per_ip() {
     let (r, app, _) = setup("guest-gone", "http://127.0.0.1:9").await;
+    known_track(&app, 7, "motus").await;
     let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.21")).await.unwrap();
     let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf");
     // A device with a session gets no second guest.
@@ -576,3 +590,32 @@ async fn guests_leave_nothing_behind_and_are_capped_per_ip() {
     }
     assert_eq!(statuses.iter().filter(|s| **s == StatusCode::TOO_MANY_REQUESTS).count(), 2);
 }
+
+#[tokio::test]
+async fn guest_staying_connected_saves_what_they_liked_to_pl8list() {
+    let pl8 = fake_pl8list(0).await;
+    let (r, app, _) = setup("guest-push", &pl8.uri()).await;
+    known_track(&app, 7, "motus").await;
+    let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.22")).await.unwrap();
+    let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf");
+    for (uri, body) in [("/local/event", r#"{"track_id":7,"kind":"like"}"#), ("/local/follow", r#"{"handle":"motus","follow":true}"#)] {
+        let res = r.clone().oneshot(req("POST", uri, Some(body), Some(&guest), "203.0.113.22")).await.unwrap();
+        assert_eq!(res.status(), StatusCode::OK, "{uri}");
+    }
+    let res = r.clone().oneshot(req("POST", "/auth/start", Some(r#"{"email":"a@b.de"}"#), None, "203.0.113.22")).await.unwrap();
+    let login = cookie_pair(&set_cookies(&res), "__Host-pl8inf_login");
+    let both = format!("{login}; {guest}");
+    let res = r.clone().oneshot(req("POST", "/auth/verify", Some(r#"{"code":"123456","connect":true}"#), Some(&both), "203.0.113.22")).await.unwrap();
+    assert_eq!(res.status(), StatusCode::OK);
+    assert!(wait_for(&pl8, |q| q.url.path() == "/api/motus/add-to-library").await, "follow not saved");
+    assert!(wait_for(&pl8, |q| q.url.path() == "/api/motus/track-7/add-to-library").await, "like not saved");
+}
+
+#[tokio::test]
+async fn radio_of_an_unknown_artist_is_not_found() {
+    let (r, _app, _) = setup("explore-unknown", "http://127.0.0.1:9").await;
+    let res = r.clone().oneshot(req("POST", "/auth/guest", Some("{}"), None, "203.0.113.23")).await.unwrap();
+    let guest = cookie_pair(&set_cookies(&res), "__Host-pl8inf");
+    let res = r.clone().oneshot(req("GET", "/local/artist?handle=made-up-handle", None, Some(&guest), "203.0.113.23")).await.unwrap();
+    assert_eq!(res.status(), StatusCode::NOT_FOUND);
+}

backend/src/ranker.rs

@@ -349,7 +349,13 @@ pub struct Filters {
     pub bpm_max: Option<f64>,
 }
 
+#[cfg(test)]
 pub fn pick(conn: &Connection, user_id: i64, n: usize, f: Filters, ranker: &dyn Ranker) -> rusqlite::Result<Vec<Value>> {
+    pick_with(conn, user_id, n, f, ranker, &taste(conn, user_id)?)
+}
+
+/// `pick` with the user's taste worked out already.
+pub fn pick_with(conn: &Connection, user_id: i64, n: usize, f: Filters, ranker: &dyn Ranker, t: &Taste) -> rusqlite::Result<Vec<Value>> {
     let mut sql = String::from(
         "SELECT * FROM tracks t WHERE (t.full_access=1 OR t.preview_upload_id IS NOT NULL) \
          AND NOT EXISTS (SELECT 1 FROM events e WHERE e.user_id=? AND e.kind='served' AND e.track_id=t.id)",
@@ -377,7 +383,6 @@ pub fn pick(conn: &Connection, user_id: i64, n: usize, f: Filters, ranker: &dyn
         return Ok(vec![]);
     }
 
-    let t = taste(conn, user_id)?;
     let recent: Vec<String> = conn
         .prepare_cached(
             "SELECT t.handle FROM events e JOIN tracks t ON t.id=e.track_id \

backend/src/sync.rs

@@ -29,6 +29,8 @@ use crate::{
 /// A connected device resyncs on use once its last sync is this old.
 pub const RESYNC_AFTER: f64 = 6.0 * 3600.0;
 const MAX_PAGES: i64 = 20;
+/// A guest's likes and follows saved to pl8list when they sign in and stay connected.
+pub const MAX_GUEST_PUSH: usize = 200;
 const DEADLINE: Duration = Duration::from_secs(120);
 
 pub struct Syncer {
@@ -76,15 +78,21 @@ impl Syncer {
 
     /// Sync once with a fresh sign-in session, then log it out at pl8list.
     pub fn once(self: &Arc<Self>, user_id: i64, session: SecretString) {
-        self.spawn_sync(user_id, session, After::Logout);
+        self.spawn_sync(user_id, session, After::Logout, vec![]);
     }
 
     /// Sync with a connected device's session. Returns false if one is already running.
     pub fn linked(self: &Arc<Self>, sealer: Sealer, session: SecretString) -> bool {
-        self.spawn_sync(sealer.user_id, session, After::Store(sealer))
+        self.spawn_sync(sealer.user_id, session, After::Store(sealer), vec![])
     }
 
-    fn spawn_sync(self: &Arc<Self>, user_id: i64, session: SecretString, after: After) -> bool {
+    /// Like `linked`, saving `push` (paths like `[handle]` or `[handle, slug]`) to the
+    /// pl8list library first: what a guest liked and followed before signing in.
+    pub fn linked_with(self: &Arc<Self>, sealer: Sealer, session: SecretString, push: Vec<Vec<String>>) -> bool {
+        self.spawn_sync(sealer.user_id, session, After::Store(sealer), push)
+    }
+
+    fn spawn_sync(self: &Arc<Self>, user_id: i64, session: SecretString, after: After, push: Vec<Vec<String>>) -> bool {
         if !self.claim(user_id) {
             if let After::Logout = after {
                 // Never keep a one-time session around just because a sync is running.
@@ -102,7 +110,11 @@ impl Syncer {
                 }
             }
             let mut s = UserSession::new(session);
-            let res = tokio::time::timeout(DEADLINE, me.read_account(user_id, &mut s)).await;
+            let res = tokio::time::timeout(DEADLINE, async {
+                me.push_all(&mut s, &push).await?;
+                me.read_account(user_id, &mut s).await
+            })
+            .await;
             let expired = match &res {
                 Ok(Ok((feed, likes, follows))) => {
                     tracing::info!("synced pl8list: {feed} feed pages, {likes} new likes, {follows} new follows");
@@ -180,6 +192,24 @@ impl Syncer {
         }
     }
 
+    /// Save each path to the library, one after the other. A failed save is skipped;
+    /// an expired session ends it.
+    async fn push_all(&self, s: &mut UserSession, push: &[Vec<String>]) -> anyhow::Result<()> {
+        let mut saved = 0;
+        for path in push {
+            let parts: Vec<&str> = path.iter().map(String::as_str).collect();
+            match self.pl8.save(s, &parts).await {
+                Ok(()) => saved += 1,
+                Err(e) if e.is::<SessionExpired>() => return Err(e),
+                Err(e) => tracing::debug!("pl8list save failed: {e:#}"),
+            }
+        }
+        if !push.is_empty() {
+            tracing::info!("saved {saved} of {} guest likes and follows to pl8list", push.len());
+        }
+        Ok(())
+    }
+
     /// Feed, saved tracks and saved artists. Returns (feed pages, new likes, new follows).
     async fn read_account(&self, user_id: i64, s: &mut UserSession) -> anyhow::Result<(i64, usize, usize)> {
         let mut feed_pages = 0;