Optimize Last.fm popularity lookups with light and full modes
Build and Publish / Build and Publish Docker Image (push) Successful in 5m22s

This commit is contained in:
Aleksandr Bogomiakov committed 2026-10-16 00:59:09 +01:00
1 parent 2b3b50f7af
commit bd1d3a4ebc
11 files changed
+930 -160

No files matched your search

+23
View File
@@ -131,6 +131,22 @@ The browser interface and JSON API are served by the same application. Import,
artwork, metadata enrichment, similarity indexing, and maintenance run as
durable background jobs rather than blocking playback requests.
The `lastfm_popularity` job defaults to **light** mode: it requests popularity
only for tracks without a completed lookup or with a changed title or primary
artist. Both matches (including zero counts) and explicit Last.fm “Track not
found” results are skipped on subsequent light runs. HTTP errors, timeouts,
invalid responses, and other API errors remain eligible for retry. A failed
full refresh also makes that track eligible for the next light run.
Choose **Full: refresh all visible tracks** in the task's admin parameters to
refresh existing statistics and retry previous misses. The options endpoint
`POST /admin/v2/api/jobs/lastfm_popularity/run-options` accepts
`{"refresh_existing": true}`; an empty object, scheduled runs, and the ordinary
run endpoint use light mode. Neither mode limits the number of tracks processed.
An authoritative miss clears current statistics while preserving their history.
The migration preserves existing popularity; legacy misses need one more lookup
because older versions did not persist them.
## Contributing
Bug reports, design discussions, and patches are welcome. Before submitting a
@@ -141,3 +157,10 @@ cargo fmt --all -- --check
cargo clippy --all-targets -- -D warnings
cargo test --all-targets
```
Last.fm database tests use temporary tables on a disposable PostgreSQL database:
```bash
FURUMUSIC_LASTFM_TEST_DATABASE_URL=postgres://user:pass@localhost/test_db \
cargo test lastfm_popularity -- --include-ignored
```
+96
View File
@@ -388,6 +388,32 @@ impl App for AdminApp {
}),
"admin_v2_artwork_backfill_run_options",
),
Route::with_handler_and_name(
"/v2/api/jobs/lastfm_popularity/run-options",
cot::router::method::post({
let pool = Arc::clone(&pool);
let pool_config = Arc::clone(&pool_config);
move |session: Session,
db: Database,
json: Json<v2::LastfmPopularityRunRequest>| {
let pool = Arc::clone(&pool);
let pool_config = Arc::clone(&pool_config);
async move {
let pg_pool = pool
.get_or_init(|| async {
sqlx::postgres::PgPoolOptions::new()
.max_connections(5)
.connect(&pool_config.database_url)
.await
.expect("admin pool")
})
.await;
v2::run_lastfm_popularity(session, db, pg_pool, json).await
}
}
}),
"admin_v2_lastfm_popularity_run_options",
),
Route::with_handler_and_name(
"/v2/api/jobs/{name}/run",
cot::router::method::post({
@@ -1237,6 +1263,33 @@ impl App for AdminApp {
}),
"admin_metadata_backfill_run",
),
Route::with_handler_and_name(
"/jobs/lastfm_popularity/run-options",
cot::router::method::post({
let pool = Arc::clone(&pool);
let pool_config = Arc::clone(&pool_config);
move |session: Session, db: Database,
form: RequestForm<views::LastfmPopularityForm>| {
let pool = Arc::clone(&pool);
let pool_config = Arc::clone(&pool_config);
async move {
let admin = match auth::require_admin_or_redirect(&session, &db).await {
Ok(u) => u,
Err(resp) => return Ok(resp),
};
let pg_pool = pool.get_or_init(|| async {
sqlx::postgres::PgPoolOptions::new()
.max_connections(3)
.connect(&pool_config.database_url)
.await
.expect("admin pool")
}).await;
views::lastfm_popularity_run(admin, &db, pg_pool, form).await
}
}
}),
"admin_lastfm_popularity_run",
),
Route::with_handler_and_name(
"/jobs/{name}/run",
cot::router::method::post({
@@ -1490,3 +1543,46 @@ impl App for AdminApp {
all
}
}
/// Start an options-based run for both admin interfaces. Ordinary scheduler
/// entry points continue to call Job::run with light-mode defaults.
async fn start_lastfm_popularity(
db: &Database,
pool: &sqlx::PgPool,
options: crate::jobs::lastfm_popularity::LastfmPopularityOptions,
) -> cot::Result<i64> {
use crate::scheduler::{JobContext, JobLog, JobRun};
let mut run = JobRun::create_running(db, "lastfm_popularity", "manual")
.await
.map_err(|err| cot::Error::internal(format!("failed to create job run: {err}")))?;
let run_id = run.id_val();
let (live_config, _) = AppConfig::load_with_db(db).await;
let db = db.clone();
let pool = pool.clone();
tokio::spawn(async move {
let start = std::time::Instant::now();
let ctx = JobContext {
config: Arc::new(live_config),
db: db.clone(),
pool: pool.clone(),
run_id,
registry: Arc::new(JobRegistry::new()),
};
let mut log = JobLog::with_live_flush(pool, run_id);
let result =
crate::jobs::lastfm_popularity::run_with_options(&ctx, &mut log, options).await;
let duration_ms = start.elapsed().as_millis() as i64;
match result {
Ok(()) => {
let _ = run.set_completed(&db, duration_ms, &log.output()).await;
}
Err(err) => {
let _ = run
.set_failed(&db, duration_ms, &log.output(), &err.to_string())
.await;
}
}
});
Ok(run_id)
}
+15
View File
@@ -110,6 +110,8 @@ pub struct ArtworkBackfillRunRequest {
overwrite_existing: bool,
}
pub type LastfmPopularityRunRequest = crate::jobs::lastfm_popularity::LastfmPopularityOptions;
#[derive(Debug, Deserialize)]
pub(super) struct UpdateLibraryItemRequest {
kind: String,
@@ -1734,6 +1736,19 @@ pub async fn run_metadata_backfill(
Json(JobRunStartedDto { ok: true, run_id }).into_response()
}
pub async fn run_lastfm_popularity(
session: Session,
db: Database,
pool: &PgPool,
Json(options): Json<LastfmPopularityRunRequest>,
) -> cot::Result<cot::response::Response> {
if let Err(response) = require_admin_json(&session, &db).await {
return Ok(response);
}
let run_id = super::start_lastfm_popularity(&db, pool, options).await?;
Json(JobRunStartedDto { ok: true, run_id }).into_response()
}
pub async fn run_artwork_backfill(
session: Session,
db: Database,
+24
View File
@@ -1453,6 +1453,30 @@ pub async fn job_run_now(
Ok(auth::redirect(&format!("/admin/jobs/{job_name}")))
}
#[derive(Debug, Form)]
pub struct LastfmPopularityForm {
refresh_existing: Option<String>,
}
pub async fn lastfm_popularity_run(
_admin: AuthenticatedUser,
db: &Database,
pool: &sqlx::PgPool,
form: RequestForm<LastfmPopularityForm>,
) -> cot::Result<cot::http::Response<Body>> {
let RequestForm(result) = form;
let FormResult::Ok(data) = result else {
return Ok(auth::redirect("/admin/jobs/lastfm_popularity"));
};
let options = crate::jobs::lastfm_popularity::LastfmPopularityOptions {
refresh_existing: data.refresh_existing.is_some(),
};
let run_id = super::start_lastfm_popularity(db, pool, options).await?;
Ok(auth::redirect(&format!(
"/admin/jobs/lastfm_popularity/runs/{run_id}"
)))
}
pub async fn metadata_backfill_run(
_admin: AuthenticatedUser,
db: &Database,
+2
View File
@@ -194,6 +194,8 @@ translations! {
jobs_back_to_list: "Back to jobs" , "Назад к заданиям";
jobs_run_detail: "Run detail" , "Детали запуска";
jobs_back_to_job: "Back to job" , "Назад к заданию";
jobs_lastfm_light_help: "Light mode checks only unchecked tracks, changed titles or artists, and failed lookups. Previous matches and confirmed misses are skipped." , "Лайт-режим проверяет только непроверенные треки, изменённые названия или исполнителей и ошибки прошлых запросов. Найденные и подтверждённо отсутствующие треки пропускаются.";
jobs_lastfm_full: "Full refresh: check all visible tracks, including previous matches and misses" , "Полное обновление: проверить все видимые треки, включая ранее найденные и отсутствующие";
jobs_metadata_backfill_options: "Metadata backfill options" , "Параметры обновления метадаты";
jobs_metadata_backfill_fields: "Fields to update" , "Поля для обновления";
jobs_metadata_backfill_fill_missing: "Fill missing only" , "Заполнить только пустые";
+346 -157
View File
@@ -1,19 +1,54 @@
use std::time::Duration;
use anyhow::Context;
use reqwest::{Client, StatusCode};
use serde::Deserialize;
use sqlx::PgConnection;
use crate::scheduler::{Job, JobContext, JobLog};
pub struct LastfmPopularityJob;
const LASTFM_REQUEST_DELAY: std::time::Duration = std::time::Duration::from_millis(1200);
const LASTFM_API_URL: &str = "https://ws.audioscrobbler.com/2.0/";
const LASTFM_REQUEST_DELAY: Duration = Duration::from_millis(1200);
const TRACK_PAGE_SIZE: i64 = 500;
const ADVISORY_LOCK_ID: i64 = 0x4c41_5354_464d_504f; // "LASTFMPO"
/// Scheduled and ordinary manual runs use light mode. Full mode is opt-in.
#[derive(Debug, Clone, Copy, Default, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct LastfmPopularityOptions {
pub refresh_existing: bool,
}
#[derive(Debug, sqlx::FromRow)]
struct TrackLookupRow {
id: i64,
title: String,
artist_name: Option<String>,
lastfm_updated_at: Option<String>,
}
// Store the lookup identity alongside the result so every metadata editing path
// (including artist renames and link changes) automatically invalidates it.
const TRACK_CANDIDATES_SQL: &str = r#"
SELECT t.id, t.title::text AS title, primary_artist.name AS artist_name
FROM furumusic__track t
LEFT JOIN LATERAL (
SELECT a.name::text AS name
FROM furumusic__track_artist ta
JOIN furumusic__artist a ON a.id = ta.artist_id
WHERE ta.track_id = t.id AND ta.role <> 'featuring'
ORDER BY ta.position, ta.artist_id
LIMIT 1
) primary_artist ON true
WHERE t.is_hidden = false AND t.id > $1 AND t.id <= $2
AND ($3 OR t.lastfm_checked_at IS NULL
OR t.lastfm_lookup_title IS DISTINCT FROM t.title::text
OR t.lastfm_lookup_artist IS DISTINCT FROM primary_artist.name)
ORDER BY t.id
LIMIT $4
"#;
#[derive(Debug, Deserialize)]
struct LastfmTrackInfoResponse {
track: Option<LastfmTrack>,
@@ -23,8 +58,69 @@ struct LastfmTrackInfoResponse {
#[derive(Debug, Deserialize)]
struct LastfmTrack {
listeners: Option<String>,
playcount: Option<String>,
listeners: String,
playcount: String,
}
#[derive(Debug, PartialEq)]
enum LookupOutcome {
Found { listeners: i64, playcount: i64 },
NotFound,
}
#[derive(Debug)]
enum LookupError {
Http(StatusCode),
Api { code: i32, message: String },
Transport(reqwest::Error),
InvalidResponse(String),
}
impl LookupError {
fn should_stop_run(&self) -> bool {
match self {
Self::Http(status) => matches!(status.as_u16(), 401 | 403 | 429),
Self::Api { code, .. } => matches!(code, 2 | 3 | 4 | 5 | 9 | 10 | 11 | 13 | 26 | 29),
_ => false,
}
}
}
impl std::fmt::Display for LookupError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Http(status) => write!(f, "Last.fm HTTP {status}"),
Self::Api { code, message } => write!(f, "Last.fm API error {code}: {message}"),
Self::Transport(err) => write!(f, "Last.fm request failed: {err}"),
Self::InvalidResponse(message) => write!(f, "Invalid Last.fm response: {message}"),
}
}
}
impl std::error::Error for LookupError {}
#[derive(Default)]
struct PopularityStats {
considered: u64,
requested: u64,
updated: u64,
not_found: u64,
skipped: u64,
failed: u64,
}
impl PopularityStats {
fn summary(&self) -> String {
format!(
"{} considered, {} requested, {} updated, {} not found, {} skipped, {} failed",
self.considered,
self.requested,
self.updated,
self.not_found,
self.skipped,
self.failed
)
}
}
#[async_trait::async_trait]
@@ -34,7 +130,7 @@ impl Job for LastfmPopularityJob {
}
fn description(&self) -> &'static str {
"Update Last.fm playcount/listener popularity for library tracks"
"Fetch Last.fm popularity for unchecked tracks; optionally refresh all tracks"
}
fn default_cron(&self) -> &'static str {
@@ -43,159 +139,236 @@ impl Job for LastfmPopularityJob {
}
async fn run(&self, ctx: &JobContext, log: &mut JobLog) -> anyhow::Result<()> {
run_with_options(ctx, log, LastfmPopularityOptions::default()).await
}
}
pub async fn run_with_options(
ctx: &JobContext,
log: &mut JobLog,
options: LastfmPopularityOptions,
) -> anyhow::Result<()> {
let api_key = ctx.config.lastfm_api_key.trim();
if api_key.is_empty() {
log.warn("lastfm_api_key is not configured, skipping Last.fm popularity update");
return Ok(());
}
let tracks = sqlx::query_as::<_, TrackLookupRow>(
r#"SELECT t.id,
t.title::text AS title,
t.lastfm_updated_at::text AS lastfm_updated_at,
(
SELECT a.name::text
FROM furumusic__track_artist ta
JOIN furumusic__artist a ON a.id = ta.artist_id
WHERE ta.track_id = t.id AND ta.role <> 'featuring'
ORDER BY ta.position
LIMIT 1
) AS artist_name
FROM furumusic__track t
WHERE t.is_hidden = false
ORDER BY t.lastfm_updated_at IS NOT NULL, t.lastfm_updated_at ASC, t.id ASC"#,
)
.fetch_all(&ctx.pool)
.await?;
if tracks.is_empty() {
log.info("No visible tracks found for Last.fm popularity update");
let mut conn = ctx.pool.acquire().await?;
// A session lock must never return to the pool, including on cancellation.
// All database work below uses this connection; no transaction spans HTTP.
conn.close_on_drop();
if !try_lock(&mut conn).await? {
log.info("Last.fm popularity update is already running; skipping this run");
return Ok(());
}
let client = Client::builder()
.user_agent(format!(
"furumusic-lastfm-popularity/{}",
env!("CARGO_PKG_VERSION")
))
.timeout(Duration::from_secs(15))
.build()?;
log.info(&format!(
"Starting Last.fm popularity update for {} visible tracks; oldest or missing ratings are processed first; request delay is {} ms; rating formula is ln(playcount + 1) * ln(listeners + 1)",
tracks.len(),
"Last.fm popularity options: mode={}; request delay={} ms",
if options.refresh_existing {
"full"
} else {
"light"
},
LASTFM_REQUEST_DELAY.as_millis()
));
let result = process_tracks(&mut conn, &client, api_key, options, log).await;
// Release promptly on normal completion; close_on_drop also covers errors
// and task cancellation without leaking the lock into another run.
conn.close().await?;
result
}
let client = reqwest::Client::builder()
.user_agent("furumusic-lastfm-popularity/0.1")
.timeout(std::time::Duration::from_secs(15))
.build()?;
let mut updated = 0u64;
let mut skipped = 0u64;
let mut failed = 0u64;
async fn try_lock(conn: &mut PgConnection) -> anyhow::Result<bool> {
Ok(sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
.bind(ADVISORY_LOCK_ID)
.fetch_one(conn)
.await?)
}
for (index, track) in tracks.iter().enumerate() {
let Some(artist) = track
.artist_name
.as_deref()
.map(str::trim)
.filter(|v| !v.is_empty())
else {
skipped += 1;
log.warn(&format!(
"Skipping track {} \"{}\": no primary artist",
track.id, track.title
));
continue;
};
async fn fetch_candidates(
conn: &mut PgConnection,
after_id: i64,
max_id: i64,
options: LastfmPopularityOptions,
) -> anyhow::Result<Vec<TrackLookupRow>> {
Ok(sqlx::query_as(TRACK_CANDIDATES_SQL)
.bind(after_id)
.bind(max_id)
.bind(options.refresh_existing)
.bind(TRACK_PAGE_SIZE)
.fetch_all(conn)
.await?)
}
log.info(&format!(
"Last.fm lookup {}/{}: track {} \"{}\" by \"{}\" (previous update: {})",
index + 1,
tracks.len(),
track.id,
track.title,
artist,
track.lastfm_updated_at.as_deref().unwrap_or("never")
));
let result = fetch_track_info(&client, api_key, artist, &track.title).await;
match result {
Ok(Some((listeners, playcount))) => {
let rating = popularity_rating(listeners, playcount);
let fetched_at = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
sqlx::query(
r#"UPDATE furumusic__track
SET lastfm_listeners = $2,
lastfm_playcount = $3,
lastfm_rating = $4,
lastfm_updated_at = $5
WHERE id = $1"#,
)
.bind(track.id)
.bind(listeners)
.bind(playcount)
.bind(rating)
.bind(&fetched_at)
.execute(&ctx.pool)
async fn process_tracks(
conn: &mut PgConnection,
client: &Client,
api_key: &str,
options: LastfmPopularityOptions,
log: &mut JobLog,
) -> anyhow::Result<()> {
// Bound the snapshot by id, not by count: every eligible track at the start
// can be processed, while concurrent imports wait for the next run.
let max_id: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(id), 0) FROM furumusic__track")
.fetch_one(&mut *conn)
.await?;
sqlx::query(
r#"INSERT INTO furumusic__track_popularity_history
(track_id, source, listeners, playcount, rating, fetched_at)
VALUES ($1, 'lastfm', $2, $3, $4, $5)"#,
)
.bind(track.id)
.bind(listeners)
.bind(playcount)
.bind(rating)
.bind(&fetched_at)
.execute(&ctx.pool)
.await?;
updated += 1;
log.info(&format!(
"Updated track {} \"{}\" by \"{}\": listeners={listeners}, playcount={playcount}, rating={rating:.4}",
track.id, track.title, artist
));
}
Ok(None) => {
skipped += 1;
log.warn(&format!(
"Last.fm has no usable match for track {} \"{}\" by \"{}\"",
track.id, track.title, artist
));
}
Err(err) if err.to_string().contains("Last.fm rate limit exceeded") => {
failed += 1;
log.error("Last.fm rate limit exceeded; stopping this run early");
let mut after_id = 0;
let mut stats = PopularityStats::default();
let mut next_request_at = tokio::time::Instant::now();
let mut last_progress = std::time::Instant::now();
loop {
let tracks = fetch_candidates(conn, after_id, max_id, options).await?;
if tracks.is_empty() {
break;
}
// Invalidate the page before HTTP, preserving its statistics. A failed
// write or cancellation during a full refresh then also leaves work for
// the next light run. This costs one update per page, not per request.
let ids: Vec<i64> = tracks.iter().map(|track| track.id).collect();
mark_pending(conn, &ids).await?;
for track in tracks {
// Advance even after an error so a failing track is tried only once
// per run and cannot starve the remaining candidates.
after_id = track.id;
stats.considered += 1;
let artist = track.artist_name.as_deref().unwrap_or_default();
if artist.trim().is_empty() || track.title.trim().is_empty() {
stats.skipped += 1;
// Missing local metadata says nothing about the Last.fm catalog.
continue;
}
tokio::time::sleep_until(next_request_at).await;
stats.requested += 1;
let outcome = fetch_track_info(client, api_key, artist, &track.title).await;
next_request_at = tokio::time::Instant::now() + LASTFM_REQUEST_DELAY;
match outcome {
Ok(outcome) => {
if save_outcome(conn, &track, &outcome).await? {
match outcome {
LookupOutcome::Found { .. } => stats.updated += 1,
LookupOutcome::NotFound => stats.not_found += 1,
}
} else {
// Metadata changed or the track was removed during HTTP.
stats.skipped += 1;
}
}
Err(err) => {
failed += 1;
stats.failed += 1;
tracing::warn!(track_id = track.id, error = %err, "Last.fm popularity lookup failed");
if stats.failed <= 5 || err.should_stop_run() {
log.warn(&format!(
"Last.fm lookup failed for track {} \"{}\" / \"{}\": {err}",
track.id, artist, track.title
"Last.fm lookup failed for track {}: {err}",
track.id
));
}
if err.should_stop_run() {
log.error(&format!("Last.fm popularity stopped: {}", stats.summary()));
return Err(err.into());
}
if (index + 1) % 50 == 0 {
log.info(&format!(
"Last.fm progress: {}/{} tracks, {updated} updated, {skipped} skipped, {failed} failed",
index + 1,
tracks.len()
));
}
tokio::time::sleep(LASTFM_REQUEST_DELAY).await;
}
log.info(&format!(
"Last.fm popularity update finished: {updated} updated, {skipped} skipped, {failed} failed, {} considered",
tracks.len()
));
// Keep live logs bounded: the shared logger rewrites its entire
// accumulated output whenever a new line is added.
if last_progress.elapsed() >= Duration::from_secs(30) {
log.info(&format!("Last.fm progress: {}", stats.summary()));
last_progress = std::time::Instant::now();
}
}
}
log.info(&format!("Last.fm popularity finished: {}", stats.summary()));
anyhow::ensure!(
stats.failed == 0,
"{} Last.fm lookups failed; they will be retried in light mode",
stats.failed
);
Ok(())
}
}
async fn mark_pending(conn: &mut PgConnection, track_ids: &[i64]) -> anyhow::Result<()> {
sqlx::query(
"UPDATE furumusic__track SET lastfm_checked_at = NULL \
WHERE id = ANY($1) AND lastfm_checked_at IS NOT NULL",
)
.bind(track_ids)
.execute(conn)
.await?;
Ok(())
}
async fn save_outcome(
conn: &mut PgConnection,
track: &TrackLookupRow,
outcome: &LookupOutcome,
) -> anyhow::Result<bool> {
let (listeners, playcount, rating, updated_at) = match outcome {
LookupOutcome::Found {
listeners,
playcount,
} => (
Some(*listeners),
Some(*playcount),
Some(popularity_rating(*listeners, *playcount)),
Some(chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()),
),
// An authoritative miss replaces the current result, but preserves
// historical observations. Unknown popularity is not a zero rating.
LookupOutcome::NotFound => (None, None, None, None),
};
// The data-modifying CTE atomically saves the result and its history in one
// round trip. Ignore responses for a lookup identity that is already stale.
let saved: bool = sqlx::query_scalar(
r#"
WITH updated AS (
UPDATE furumusic__track t
SET lastfm_listeners = $2, lastfm_playcount = $3,
lastfm_rating = $4, lastfm_updated_at = $5,
lastfm_checked_at = CURRENT_TIMESTAMP,
lastfm_lookup_title = $6, lastfm_lookup_artist = $7
WHERE t.id = $1 AND t.is_hidden = false AND t.title::text = $6
AND (SELECT a.name::text
FROM furumusic__track_artist ta
JOIN furumusic__artist a ON a.id = ta.artist_id
WHERE ta.track_id = t.id AND ta.role <> 'featuring'
ORDER BY ta.position, ta.artist_id LIMIT 1) IS NOT DISTINCT FROM $7
RETURNING id
), history AS (
INSERT INTO furumusic__track_popularity_history
(track_id, source, listeners, playcount, rating, fetched_at)
SELECT id, 'lastfm', $2::bigint, $3::bigint, $4::double precision, $5::varchar
FROM updated WHERE $2::bigint IS NOT NULL
)
SELECT EXISTS (SELECT 1 FROM updated)
"#,
)
.bind(track.id)
.bind(listeners)
.bind(playcount)
.bind(rating)
.bind(updated_at)
.bind(&track.title)
.bind(&track.artist_name)
.fetch_one(conn)
.await?;
Ok(saved)
}
async fn fetch_track_info(
client: &reqwest::Client,
client: &Client,
api_key: &str,
artist: &str,
track: &str,
) -> anyhow::Result<Option<(i64, i64)>> {
) -> Result<LookupOutcome, LookupError> {
let response = client
.get("https://ws.audioscrobbler.com/2.0/")
.get(LASTFM_API_URL)
.query(&[
("method", "track.getInfo"),
("api_key", api_key),
@@ -205,45 +378,61 @@ async fn fetch_track_info(
("format", "json"),
])
.send()
.await?;
.await
.map_err(|err| LookupError::Transport(err.without_url()))?;
let status = response.status();
if !status.is_success() {
return Err(LookupError::Http(status));
}
let body = response
.bytes()
.await
.map_err(|err| LookupError::Transport(err.without_url()))?;
interpret_response(status, &body)
}
if response.status() == reqwest::StatusCode::NOT_FOUND {
return Ok(None);
fn interpret_response(status: StatusCode, body: &[u8]) -> Result<LookupOutcome, LookupError> {
// A bare HTTP 404 (or any other HTTP failure) never proves catalog absence.
if !status.is_success() {
return Err(LookupError::Http(status));
}
let response = response.error_for_status()?;
let body: LastfmTrackInfoResponse = response.json().await?;
let body: LastfmTrackInfoResponse = serde_json::from_slice(body)
.map_err(|err| LookupError::InvalidResponse(err.to_string()))?;
if let Some(code) = body.error {
if code == 29 {
anyhow::bail!("Last.fm rate limit exceeded");
let message = body.message.unwrap_or_else(|| "unknown error".to_owned());
// Last.fm uses error 6 for both missing tracks and invalid parameters.
// Only an explicit track-not-found message is a cacheable miss.
if matches!(code, 6 | 7)
&& message
.trim()
.trim_end_matches('.')
.eq_ignore_ascii_case("track not found")
{
return Ok(LookupOutcome::NotFound);
}
if code == 6 || code == 7 {
return Ok(None);
return Err(LookupError::Api { code, message });
}
anyhow::bail!(
"Last.fm API error {code}: {}",
body.message.unwrap_or_else(|| "unknown error".to_string())
);
}
let Some(info) = body.track else {
return Ok(None);
let info = body
.track
.ok_or_else(|| LookupError::InvalidResponse("missing track".to_owned()))?;
let parse_count = |value: &str, field: &str| {
value
.parse::<i64>()
.ok()
.filter(|value| *value >= 0)
.with_context(|| format!("invalid {field}"))
.map_err(|err| LookupError::InvalidResponse(err.to_string()))
};
let listeners = info
.listeners
.as_deref()
.unwrap_or("0")
.parse::<i64>()
.unwrap_or(0);
let playcount = info
.playcount
.as_deref()
.unwrap_or("0")
.parse::<i64>()
.unwrap_or(0);
Ok(Some((listeners.max(0), playcount.max(0))))
Ok(LookupOutcome::Found {
listeners: parse_count(&info.listeners, "listeners")?,
playcount: parse_count(&info.playcount, "playcount")?,
})
}
fn popularity_rating(listeners: i64, playcount: i64) -> f64 {
let listeners = listeners.max(0) as f64;
let playcount = playcount.max(0) as f64;
playcount.ln_1p() * listeners.ln_1p()
(playcount as f64).ln_1p() * (listeners as f64).ln_1p()
}
#[cfg(test)]
#[path = "lastfm_popularity_tests.rs"]
mod tests;
+337
View File
@@ -0,0 +1,337 @@
use super::*;
use sqlx::Connection;
#[test]
fn run_options_default_to_light_and_require_explicit_full_mode() {
assert!(!LastfmPopularityOptions::default().refresh_existing);
let light: LastfmPopularityOptions = serde_json::from_str("{}").unwrap();
assert!(!light.refresh_existing);
let full: LastfmPopularityOptions =
serde_json::from_str(r#"{"refresh_existing":true}"#).unwrap();
assert!(full.refresh_existing);
assert!(
serde_json::from_str::<LastfmPopularityOptions>(r#"{"refresh_existing":"true"}"#).is_err()
);
assert!(
serde_json::from_str::<LastfmPopularityOptions>(r#"{"refresh_exising":true}"#).is_err()
);
}
#[test]
fn only_explicit_catalog_misses_are_cacheable() {
for code in [6, 7] {
for message in ["Track not found", "Track not found.", " track NOT found "] {
let body = serde_json::json!({"error": code, "message": message}).to_string();
assert_eq!(
interpret_response(StatusCode::OK, body.as_bytes()).unwrap(),
LookupOutcome::NotFound
);
}
}
for (code, message) in [
(6, "Invalid parameters"),
(6, "Track parameter is required"),
(7, "Invalid resource specified"),
(8, "Operation failed"),
(11, "Service Offline"),
(16, "Temporary error"),
(29, "Track not found"),
] {
let body = serde_json::json!({"error": code, "message": message}).to_string();
assert!(interpret_response(StatusCode::OK, body.as_bytes()).is_err());
}
for status in [
StatusCode::NOT_FOUND,
StatusCode::TOO_MANY_REQUESTS,
StatusCode::INTERNAL_SERVER_ERROR,
StatusCode::SERVICE_UNAVAILABLE,
] {
// Even a JSON not-found message cannot turn an HTTP error into a miss.
assert!(interpret_response(status, br#"{"error":6,"message":"Track not found"}"#).is_err());
}
}
#[test]
fn invalid_or_incomplete_popularity_is_retryable_and_zero_is_valid() {
for body in [
"not json",
"{}",
r#"{"track":null}"#,
r#"{"track":{}}"#,
r#"{"track":{"listeners":"1"}}"#,
r#"{"track":{"listeners":"bad","playcount":"2"}}"#,
r#"{"track":{"listeners":"1","playcount":"-1"}}"#,
r#"{"track":{"listeners":"1","playcount":"9223372036854775808"}}"#,
r#"{"track":{"listeners":null,"playcount":"2"}}"#,
] {
assert!(
interpret_response(StatusCode::OK, body.as_bytes()).is_err(),
"{body}"
);
}
assert_eq!(
interpret_response(
StatusCode::OK,
br#"{"track":{"listeners":"0","playcount":"0"}}"#
)
.unwrap(),
LookupOutcome::Found {
listeners: 0,
playcount: 0
}
);
assert_eq!(popularity_rating(0, 0), 0.0);
assert!(popularity_rating(10, 100) > popularity_rating(1, 10));
}
#[test]
fn global_errors_stop_the_run_without_caching_absence() {
assert!(LookupError::Http(StatusCode::TOO_MANY_REQUESTS).should_stop_run());
for code in [10, 11, 26, 29] {
assert!(
LookupError::Api {
code,
message: String::new()
}
.should_stop_run()
);
}
assert!(!LookupError::Http(StatusCode::INTERNAL_SERVER_ERROR).should_stop_run());
assert!(!LookupError::InvalidResponse("missing track".into()).should_stop_run());
}
// These tests use temporary tables on isolated connections, never application
// tables. Run explicitly against a disposable PostgreSQL instance:
// FURUMUSIC_LASTFM_TEST_DATABASE_URL=... cargo test lastfm_popularity -- --include-ignored
async fn database_fixture() -> PgConnection {
let url = std::env::var("FURUMUSIC_LASTFM_TEST_DATABASE_URL")
.expect("set FURUMUSIC_LASTFM_TEST_DATABASE_URL to a disposable PostgreSQL database");
let mut conn = PgConnection::connect(&url).await.unwrap();
sqlx::raw_sql(
r#"
CREATE TEMP TABLE furumusic__track (
id BIGINT PRIMARY KEY, title TEXT NOT NULL,
is_hidden BOOLEAN NOT NULL DEFAULT false,
lastfm_listeners BIGINT, lastfm_playcount BIGINT,
lastfm_rating DOUBLE PRECISION, lastfm_updated_at VARCHAR(32)
);
CREATE TEMP TABLE furumusic__artist (id BIGINT PRIMARY KEY, name TEXT NOT NULL);
CREATE TEMP TABLE furumusic__track_artist (
track_id BIGINT, artist_id BIGINT, role TEXT, position INTEGER
);
CREATE TEMP TABLE furumusic__track_popularity_history (
id BIGSERIAL PRIMARY KEY, track_id BIGINT NOT NULL, source VARCHAR(32) NOT NULL,
listeners BIGINT NOT NULL, playcount BIGINT NOT NULL,
rating DOUBLE PRECISION NOT NULL, fetched_at VARCHAR(32) NOT NULL
);
INSERT INTO furumusic__artist VALUES (1, 'Artist'), (2, 'Guest'), (3, 'Other');
INSERT INTO furumusic__track (id, title, lastfm_rating, lastfm_listeners, lastfm_playcount)
VALUES (1, 'Legacy zero', 0, 0, 0), (2, 'New', NULL, NULL, NULL),
(3, 'Hidden', NULL, NULL, NULL), (4, 'No artist', NULL, NULL, NULL);
UPDATE furumusic__track SET is_hidden = true WHERE id = 3;
INSERT INTO furumusic__track_artist VALUES
(1, 1, 'main', 0), (2, 1, 'main', 0), (3, 1, 'main', 0), (2, 2, 'featuring', -1);
"#,
)
.execute(&mut conn)
.await
.unwrap();
// Exercise precisely the same prepared statements as the real migration.
for statement in include_str!("../music/lastfm_popularity_lookup.sql").split(';') {
if !statement.trim().is_empty() {
sqlx::query(statement).execute(&mut conn).await.unwrap();
}
}
conn
}
async fn candidates(conn: &mut PgConnection, full: bool) -> Vec<TrackLookupRow> {
fetch_candidates(
conn,
0,
i64::MAX,
LastfmPopularityOptions {
refresh_existing: full,
},
)
.await
.unwrap()
}
fn ids(rows: &[TrackLookupRow]) -> Vec<i64> {
rows.iter().map(|row| row.id).collect()
}
#[tokio::test]
#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"]
async fn database_modes_misses_retries_and_metadata_changes() {
let mut conn = database_fixture().await;
assert_eq!(ids(&candidates(&mut conn, false).await), [2, 4]);
assert_eq!(ids(&candidates(&mut conn, true).await), [1, 2, 4]);
let new = candidates(&mut conn, false).await.remove(0);
assert_eq!(new.artist_name.as_deref(), Some("Artist"));
assert!(
save_outcome(&mut conn, &new, &LookupOutcome::NotFound)
.await
.unwrap()
);
assert_eq!(ids(&candidates(&mut conn, false).await), [4]);
assert_eq!(ids(&candidates(&mut conn, true).await), [1, 2, 4]);
// A failed full refresh must become eligible for light mode again.
mark_pending(&mut conn, &[new.id]).await.unwrap();
assert_eq!(ids(&candidates(&mut conn, false).await), [2, 4]);
let found = LookupOutcome::Found {
listeners: 5,
playcount: 10,
};
assert!(save_outcome(&mut conn, &new, &found).await.unwrap());
assert_eq!(ids(&candidates(&mut conn, false).await), [4]);
// Unlike a network error, a confirmed miss replaces old current statistics.
assert!(
save_outcome(&mut conn, &new, &LookupOutcome::NotFound)
.await
.unwrap()
);
let rating: Option<f64> =
sqlx::query_scalar("SELECT lastfm_rating FROM furumusic__track WHERE id=2")
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(rating, None);
let history: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM furumusic__track_popularity_history")
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(history, 1);
sqlx::query("UPDATE furumusic__artist SET name='Renamed' WHERE id=1")
.execute(&mut conn)
.await
.unwrap();
assert_eq!(ids(&candidates(&mut conn, false).await), [1, 2, 4]);
// A response fetched before a concurrent rename cannot mark the new identity checked.
assert!(!save_outcome(&mut conn, &new, &found).await.unwrap());
let renamed = candidates(&mut conn, false).await.remove(1);
assert!(save_outcome(&mut conn, &renamed, &found).await.unwrap());
sqlx::query("UPDATE furumusic__track SET title='New title' WHERE id=2")
.execute(&mut conn)
.await
.unwrap();
assert_eq!(ids(&candidates(&mut conn, false).await), [1, 2, 4]);
assert!(!save_outcome(&mut conn, &renamed, &found).await.unwrap());
let retitled = candidates(&mut conn, false).await.remove(1);
assert!(save_outcome(&mut conn, &retitled, &found).await.unwrap());
sqlx::query("UPDATE furumusic__track_artist SET artist_id=3 WHERE track_id=2 AND role='main'")
.execute(&mut conn)
.await
.unwrap();
assert_eq!(ids(&candidates(&mut conn, false).await), [1, 2, 4]);
assert!(!save_outcome(&mut conn, &retitled, &found).await.unwrap());
let relinked = candidates(&mut conn, false).await.remove(1);
assert_eq!(relinked.artist_name.as_deref(), Some("Other"));
assert!(save_outcome(&mut conn, &relinked, &found).await.unwrap());
sqlx::query("DELETE FROM furumusic__track_artist WHERE track_id=2 AND role='main'")
.execute(&mut conn)
.await
.unwrap();
assert_eq!(candidates(&mut conn, false).await[1].artist_name, None);
}
#[tokio::test]
#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"]
async fn database_pagination_does_not_limit_runs_or_repeat_failures() {
let mut conn = database_fixture().await;
sqlx::query("INSERT INTO furumusic__track (id, title) SELECT n, 'Track ' || n FROM generate_series(5, 1105) n")
.execute(&mut conn).await.unwrap();
sqlx::query("INSERT INTO furumusic__track_artist SELECT id, 1, 'main', 0 FROM furumusic__track WHERE id>=5")
.execute(&mut conn).await.unwrap();
let mut after_id = 0;
let mut seen = Vec::new();
loop {
let page = fetch_candidates(
&mut conn,
after_id,
1105,
LastfmPopularityOptions::default(),
)
.await
.unwrap();
if page.is_empty() {
break;
}
assert!(page.len() <= TRACK_PAGE_SIZE as usize);
for row in page {
after_id = row.id;
// Simulate failed requests: all candidates remain pending in the DB.
mark_pending(&mut conn, &[row.id]).await.unwrap();
seen.push(row.id);
}
}
assert_eq!(seen.len(), 1103);
assert!(seen.windows(2).all(|pair| pair[0] < pair[1]));
assert_eq!(seen.last(), Some(&1105));
let bounded = fetch_candidates(&mut conn, 1000, 1010, LastfmPopularityOptions::default())
.await
.unwrap();
assert_eq!(ids(&bounded), (1001..=1010).collect::<Vec<_>>());
}
#[tokio::test]
#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"]
async fn database_result_and_history_are_atomic() {
let mut conn = database_fixture().await;
let track = candidates(&mut conn, false).await.remove(0);
sqlx::query("ALTER TABLE furumusic__track_popularity_history ADD CHECK (playcount > 100)")
.execute(&mut conn)
.await
.unwrap();
assert!(
save_outcome(
&mut conn,
&track,
&LookupOutcome::Found {
listeners: 5,
playcount: 10
}
)
.await
.is_err()
);
let state: (bool, Option<f64>) = sqlx::query_as(
"SELECT lastfm_checked_at IS NULL, lastfm_rating FROM furumusic__track WHERE id=2",
)
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(state, (true, None));
}
#[tokio::test]
#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"]
async fn database_interrupted_full_refresh_remains_pending_without_losing_statistics() {
let mut conn = database_fixture().await;
let page = candidates(&mut conn, true).await;
let track_ids = ids(&page);
mark_pending(&mut conn, &track_ids).await.unwrap();
assert_eq!(track_ids, vec![1, 2, 4]);
assert_eq!(ids(&candidates(&mut conn, false).await), track_ids);
let state: (bool, Option<f64>) = sqlx::query_as(
"SELECT lastfm_checked_at IS NULL, lastfm_rating FROM furumusic__track WHERE id=1",
)
.fetch_one(&mut conn)
.await
.unwrap();
assert_eq!(state, (true, Some(0.0)));
}
#[tokio::test]
#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"]
async fn database_lock_is_shared_by_light_and_full_runs() {
let url = std::env::var("FURUMUSIC_LASTFM_TEST_DATABASE_URL").unwrap();
let mut first = PgConnection::connect(&url).await.unwrap();
let mut second = PgConnection::connect(&url).await.unwrap();
assert!(try_lock(&mut first).await.unwrap());
assert!(!try_lock(&mut second).await.unwrap());
first.close().await.unwrap();
assert!(try_lock(&mut second).await.unwrap());
}
+27
View File
@@ -0,0 +1,27 @@
ALTER TABLE furumusic__track
ADD COLUMN IF NOT EXISTS lastfm_checked_at TIMESTAMPTZ,
ADD COLUMN IF NOT EXISTS lastfm_lookup_title TEXT,
ADD COLUMN IF NOT EXISTS lastfm_lookup_artist TEXT;
CREATE INDEX IF NOT EXISTS idx_track_artist_lastfm_lookup
ON furumusic__track_artist (track_id, position, artist_id)
WHERE role <> 'featuring';
-- Preserve existing observations, including valid zero-valued popularity.
-- Legacy misses cannot be backfilled because they were only written to logs.
UPDATE furumusic__track t
SET lastfm_checked_at = CURRENT_TIMESTAMP,
lastfm_lookup_title = t.title::text,
lastfm_lookup_artist = (
SELECT a.name::text
FROM furumusic__track_artist ta
JOIN furumusic__artist a ON a.id = ta.artist_id
WHERE ta.track_id = t.id AND ta.role <> 'featuring'
ORDER BY ta.position, ta.artist_id
LIMIT 1
)
WHERE t.lastfm_checked_at IS NULL
AND (t.lastfm_updated_at IS NOT NULL
OR t.lastfm_rating IS NOT NULL
OR t.lastfm_listeners IS NOT NULL
OR t.lastfm_playcount IS NOT NULL);
+30
View File
@@ -2823,6 +2823,35 @@ pub mod db_migrations {
&[Operation::custom(add_playback_coordination_columns).build()];
}
// -- M0049: persist Last.fm lookup results and their metadata identity --
#[cot::db::migrations::migration_op]
async fn add_lastfm_lookup_state(ctx: migrations::MigrationContext<'_>) -> cot::db::Result<()> {
// Cot prepares raw queries, so execute these static DDL statements
// separately rather than sending a multi-statement prepared query.
for statement in include_str!("lastfm_popularity_lookup.sql").split(';') {
if !statement.trim().is_empty() {
ctx.db.raw(statement).await?;
}
}
Ok(())
}
#[derive(Debug, Copy, Clone)]
pub struct M0049AddLastfmLookupState;
impl migrations::Migration for M0049AddLastfmLookupState {
const APP_NAME: &'static str = "furumusic";
const MIGRATION_NAME: &'static str = "m_0049_add_lastfm_lookup_state";
const DEPENDENCIES: &'static [migrations::MigrationDependency] =
&[migrations::MigrationDependency::migration(
"furumusic",
"m_0048_add_playback_coordination_columns",
)];
const OPERATIONS: &'static [Operation] =
&[Operation::custom(add_lastfm_lookup_state).build()];
}
pub const MIGRATIONS: &[&SyncDynMigration] = &[
&M0006CreateMediaFile,
&M0007CreateArtist,
@@ -2862,5 +2891,6 @@ pub mod db_migrations {
&M0046CreateLocalUploadHistory,
&M0047CreateYouTubeImportMediaLinks,
&M0048AddPlaybackCoordinationColumns,
&M0049AddLastfmLookupState,
];
}
+11 -1
View File
@@ -13,7 +13,7 @@
</table>
<div style="margin: 1rem 0; display: flex; gap: .5rem; align-items: flex-end;">
{% if job.name_str() != "metadata_backfill" %}
{% if job.name_str() != "metadata_backfill" && job.name_str() != "lastfm_popularity" %}
<form method="post" action="/admin/jobs/{{ job.name_str() }}/run" style="margin:0;">
<button type="submit" style="padding:.4rem 1rem; background:#007bff; color:#fff; border:none; border-radius:4px; cursor:pointer;">{{ t.jobs_run_now }}</button>
</form>
@@ -29,6 +29,16 @@
{% endif %}
</div>
{% if job.name_str() == "lastfm_popularity" %}
<form method="post" action="/admin/jobs/lastfm_popularity/run-options" style="margin:1rem 0;">
<p>{{ t.jobs_lastfm_light_help }}</p>
<label style="display:block; margin-bottom:1rem;">
<input type="checkbox" name="refresh_existing"> {{ t.jobs_lastfm_full }}
</label>
<button type="submit">{{ t.jobs_run_now }}</button>
</form>
{% endif %}
{% if job.name_str() == "metadata_backfill" %}
<h2>{{ t.jobs_metadata_backfill_options }}</h2>
<form method="post" action="/admin/jobs/metadata_backfill/run-options" style="margin:0 0 1.5rem; padding:1rem; background:#fff; border:1px solid #e0e0e0; border-radius:6px;">
+19 -2
View File
@@ -2292,13 +2292,20 @@ tbody tr:hover {
<label><input type="radio" value="overwrite" x-model="artworkBackfillOptions.mode" /> Search all and replace existing</label>
</div>
</div>
<div class="metadata-backfill-options" x-show="isLastfmPopularityJob(activeJob)">
<div class="mode-row">
<label><input type="radio" value="light" x-model="lastfmPopularityOptions.mode" /> Light: unchecked tracks only</label>
<label><input type="radio" value="full" x-model="lastfmPopularityOptions.mode" /> Full: refresh all visible tracks</label>
</div>
<p class="job-param-note">Light mode skips previous matches and tracks confirmed missing. Changed titles or artists and failed lookups are checked again. Full mode also retries previous matches and misses.</p>
</div>
<div class="job-param-note" x-show="activeJob && !jobHasParameterForm(activeJob)">
This task has no manual parameters.
</div>
<div class="toolbar" style="margin-bottom:12px">
<button class="btn primary" type="button" @click="runJob(activeJob)">
<i data-lucide="play"></i>
<span x-text="isMetadataBackfillJob(activeJob) ? 'Run with options' : 'Run now'"></span>
<span x-text="jobHasParameterForm(activeJob) ? 'Run with options' : 'Run now'"></span>
</button>
<button class="btn" type="button" @click="toggleJob(activeJob)">
<i data-lucide="power"></i>
@@ -3811,6 +3818,7 @@ function adminV2() {
musicbrainz_tags: true,
mode: 'fill_missing'
},
lastfmPopularityOptions: { mode: 'light' },
artworkBackfillOptions: {
mode: 'missing'
},
@@ -4624,6 +4632,11 @@ function adminV2() {
method: 'POST',
body: JSON.stringify(this.artworkBackfillPayload())
})
: this.isLastfmPopularityJob(job)
? await this.request(`${this.apiBase}/jobs/lastfm_popularity/run-options`, {
method: 'POST',
body: JSON.stringify({ refresh_existing: this.lastfmPopularityOptions.mode === 'full' })
})
: await this.request(`${this.apiBase}/jobs/${encodeURIComponent(job.name)}/run`, { method: 'POST' });
this.showToast(`Run #${result.run_id} started`);
this.activeJobName = job.name;
@@ -4647,8 +4660,12 @@ function adminV2() {
return job && job.name === 'artwork_backfill';
},
isLastfmPopularityJob(job) {
return job && job.name === 'lastfm_popularity';
},
jobHasParameterForm(job) {
return this.isMetadataBackfillJob(job) || this.isArtworkBackfillJob(job);
return this.isMetadataBackfillJob(job) || this.isArtworkBackfillJob(job) || this.isLastfmPopularityJob(job);
},
metadataBackfillPayload() {