Joren/qbz
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: optimize ML/reco module for home view loading

Browse Source 
- Add composite indexes on reco_events for GROUP BY + ORDER BY queries
  (event_type + album_id/track_id/artist_id + created_at DESC)
- Add composite index on reco_scores for scored lookups
  (score_type + item_type + score DESC)
- Eliminate duplicate fallback queries in get_home_seeds_internal
  by inlining the empty check into the scored branch
- Parallelize resolve_favorites with recent albums, tracks, and
  artists (4-way tokio::try_join instead of 3 + sequential)
- Batch cache writes: collect all Tier 3 API results first, then
  write to API cache and reco meta with single lock acquisitions
  instead of locking per-item inside spawned tasks
This commit is contained in:
vicrodh committed 2026-03-08 12:52:52 -06:00
1 parent 8a2c12ee41
commit 63b1161afe
2 files changed
+194 -135
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src-tauri/src/reco_store/commands.rs
+179 -135
View File
@@ -671,64 +671,74 @@ fn get_home_seeds_internal(
) -> Result<HomeSeeds, String> {
let has_scores = db.has_scores("all")?;
let recent_fresh_albums = db.get_recent_album_ids(4)?;
let recent_fresh_tracks = db.get_recent_track_ids(4)?;
let mut recently_played_album_ids = if has_scores {
// When scores exist, merge scored results with recent items.
// When no scores (or scored results empty), fall back to event-based queries.
// Avoid duplicate queries: only call fallback if scored path was taken but empty.
let recently_played_album_ids = if has_scores {
let recent_fresh = db.get_recent_album_ids(4)?;
let scored = db.get_scored_album_ids("all", limit_recent_albums + 4)?;
merge_unique_preserve_order(recent_fresh_albums, scored, limit_recent_albums as usize)
let merged = merge_unique_preserve_order(recent_fresh, scored, limit_recent_albums as usize);
if merged.is_empty() {
db.get_recent_album_ids(limit_recent_albums)?
} else {
merged
}
} else {
db.get_recent_album_ids(limit_recent_albums)?
};
let mut continue_listening_track_ids = if has_scores {
let continue_listening_track_ids = if has_scores {
let recent_fresh = db.get_recent_track_ids(4)?;
let scored = db.get_scored_track_ids("all", limit_continue_tracks + 4)?;
merge_unique_preserve_order(recent_fresh_tracks, scored, limit_continue_tracks as usize)
let merged = merge_unique_preserve_order(recent_fresh, scored, limit_continue_tracks as usize);
if merged.is_empty() {
db.get_recent_track_ids(limit_continue_tracks)?
} else {
merged
}
} else {
db.get_recent_track_ids(limit_continue_tracks)?
};
let mut top_artist_ids = if has_scores {
db.get_scored_artist_scores("all", limit_top_artists)?
let top_artist_ids = if has_scores {
let scored: Vec<TopArtistSeed> = db.get_scored_artist_scores("all", limit_top_artists)?
.into_iter()
.map(|(artist_id, score)| TopArtistSeed {
artist_id,
play_count: score.round().max(1.0) as u32,
})
.collect()
.collect();
if scored.is_empty() {
db.get_top_artist_ids(limit_top_artists)?
} else {
scored
}
} else {
Vec::new()
db.get_top_artist_ids(limit_top_artists)?
};
let mut favorite_album_ids = if has_scores {
db.get_scored_album_ids("favorite", limit_favorites)?
let favorite_album_ids = if has_scores {
let scored = db.get_scored_album_ids("favorite", limit_favorites)?;
if scored.is_empty() {
db.get_favorite_album_ids(limit_favorites)?
} else {
scored
}
} else {
Vec::new()
db.get_favorite_album_ids(limit_favorites)?
};
let mut favorite_track_ids = if has_scores {
db.get_scored_track_ids("favorite", limit_favorites)?
let favorite_track_ids = if has_scores {
let scored = db.get_scored_track_ids("favorite", limit_favorites)?;
if scored.is_empty() {
db.get_favorite_track_ids(limit_favorites)?
} else {
scored
}
} else {
Vec::new()
db.get_favorite_track_ids(limit_favorites)?
};
// Fallbacks for empty results
if recently_played_album_ids.is_empty() {
recently_played_album_ids = db.get_recent_album_ids(limit_recent_albums)?;
}
if continue_listening_track_ids.is_empty() {
continue_listening_track_ids = db.get_recent_track_ids(limit_continue_tracks)?;
}
if top_artist_ids.is_empty() {
top_artist_ids = db.get_top_artist_ids(limit_top_artists)?;
}
if favorite_album_ids.is_empty() {
favorite_album_ids = db.get_favorite_album_ids(limit_favorites)?;
}
if favorite_track_ids.is_empty() {
favorite_track_ids = db.get_favorite_track_ids(limit_favorites)?;
}
Ok(HomeSeeds {
recently_played_album_ids,
continue_listening_track_ids,
@@ -794,14 +804,20 @@ pub async fn reco_get_home_resolved(
.map(|s| (s.artist_id, s.play_count))
.collect();
// Step 3: Resolve albums, tracks, AND artists in parallel
// Artists are independent of albums/tracks, so start them concurrently.
let albums_fut = resolve_albums(
// Step 3: Resolve all entity types in parallel.
// Recent albums, favorite albums, tracks, and artists are all independent.
let recent_albums_fut = resolve_albums(
&seeds.recently_played_album_ids,
&reco_state,
&app_state,
&cache_state,
);
let favorite_albums_fut = resolve_albums(
&seeds.favorite_album_ids,
&reco_state,
&app_state,
&cache_state,
);
let tracks_fut = resolve_tracks(&all_track_ids, &reco_state, &app_state, &cache_state);
let artists_fut = resolve_artists(
&all_artist_ids,
@@ -811,8 +827,8 @@ pub async fn reco_get_home_resolved(
&cache_state,
);
let (recently_played_albums, all_tracks, top_artists) =
tokio::try_join!(albums_fut, tracks_fut, artists_fut)?;
let (recently_played_albums, resolved_favorites, all_tracks, top_artists) =
tokio::try_join!(recent_albums_fut, favorite_albums_fut, tracks_fut, artists_fut)?;
// Build track lookup
let track_map: HashMap<u64, &TrackDisplayMeta> =
@@ -824,18 +840,9 @@ pub async fn reco_get_home_resolved(
.filter_map(|id| track_map.get(id).map(|tr| (*tr).clone()))
.collect();
// Step 5: "More From Favorites" = discover albums by favorite artists,
// Step 4: "More From Favorites" = discover albums by favorite artists,
// excluding albums the user already has in favorites / recently played.
// 5a: Resolve favorite albums to get their artist IDs
let resolved_favorites = resolve_albums(
&seeds.favorite_album_ids,
&reco_state,
&app_state,
&cache_state,
)
.await?;
// 5b: Collect unique artist IDs from favorite albums + favorite tracks
let favorite_artist_ids: Vec<u64> = {
let mut ids = Vec::new();
@@ -989,6 +996,7 @@ async fn resolve_albums(
// Tier 2: API cache (24h TTL)
let mut api_cache_resolved: HashMap<String, AlbumCardMeta> = HashMap::new();
let mut tier2_album_metas: Vec<AlbumCardMeta> = Vec::new();
if !missing_ids.is_empty() {
let cached = {
let guard__ = cache_state.cache.lock().await;
@@ -1000,16 +1008,19 @@ async fn resolve_albums(
for (album_id, json_str) in cached {
if let Ok(album) = serde_json::from_str::<Album>(&json_str) {
let meta = album_to_card_meta(&album);
// Write through to reco meta for future instant hits
{
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
let _ = db.set_album_meta(&meta);
}
}
tier2_album_metas.push(meta.clone());
api_cache_resolved.insert(album_id, meta);
}
}
// Batch write tier-2 hits to reco meta (single lock)
if !tier2_album_metas.is_empty() {
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
for meta in &tier2_album_metas {
let _ = db.set_album_meta(meta);
}
}
}
}
let still_missing: Vec<String> = missing_ids
@@ -1018,7 +1029,7 @@ async fn resolve_albums(
.cloned()
.collect();
// Tier 3: Qobuz API (parallel with semaphore)
// Tier 3: Qobuz API — fetch in parallel, defer cache writes to after all complete
let mut api_resolved: HashMap<String, AlbumCardMeta> = HashMap::new();
if !still_missing.is_empty() {
log::info!(
@@ -1027,15 +1038,11 @@ async fn resolve_albums(
);
let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(10));
let client = app_state.client.clone();
let cache_arc = cache_state.cache.clone();
let reco_arc = reco_state.db.clone();
let mut handles = Vec::new();
for album_id in &still_missing {
let sem = sem.clone();
let client = client.clone();
let cache_arc = cache_arc.clone();
let reco_arc = reco_arc.clone();
let album_id = album_id.clone();
handles.push(tokio::spawn(async move {
@@ -1045,31 +1052,42 @@ async fn resolve_albums(
c.get_album(&album_id).await.ok()?
};
let meta = album_to_card_meta(&album);
// Cache the full API response
if let Ok(json) = serde_json::to_string(&album) {
let guard__ = cache_arc.lock().await;
if let Some(cache) = guard__.as_ref() {
let _ = cache.set_album(&album_id, &json);
}
}
// Write to reco meta
{
let guard__ = reco_arc.lock().await;
if let Some(db) = guard__.as_ref() {
let _ = db.set_album_meta(&meta);
}
}
Some((album_id, meta))
let json = serde_json::to_string(&album).ok();
Some((album_id, meta, json))
}));
}
// Collect results, then batch-write caches
let mut api_albums_to_cache: Vec<(String, String)> = Vec::new();
let mut api_album_metas: Vec<AlbumCardMeta> = Vec::new();
for handle in handles {
if let Ok(Some((id, meta))) = handle.await {
if let Ok(Some((id, meta, json))) = handle.await {
if let Some(j) = json {
api_albums_to_cache.push((id.clone(), j));
}
api_album_metas.push(meta.clone());
api_resolved.insert(id, meta);
}
}
// Batch write to API cache (single lock)
if !api_albums_to_cache.is_empty() {
let guard__ = cache_state.cache.lock().await;
if let Some(cache) = guard__.as_ref() {
for (album_id, json) in &api_albums_to_cache {
let _ = cache.set_album(album_id, json);
}
}
}
// Batch write to reco meta (single lock)
if !api_album_metas.is_empty() {
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
for meta in &api_album_metas {
let _ = db.set_album_meta(meta);
}
}
}
}
// Assemble in seed order
@@ -1117,6 +1135,7 @@ async fn resolve_tracks(
// Tier 2: API cache
let mut api_cache_resolved: HashMap<u64, TrackDisplayMeta> = HashMap::new();
let mut tier2_track_metas: Vec<TrackDisplayMeta> = Vec::new();
if !missing_ids.is_empty() {
let cached = {
let guard__ = cache_state.cache.lock().await;
@@ -1128,15 +1147,19 @@ async fn resolve_tracks(
for (track_id, json_str) in cached {
if let Ok(track) = serde_json::from_str::<Track>(&json_str) {
let meta = track_to_display_meta(&track);
{
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
let _ = db.set_track_meta(&meta);
}
}
tier2_track_metas.push(meta.clone());
api_cache_resolved.insert(track_id, meta);
}
}
// Batch write tier-2 hits to reco meta (single lock)
if !tier2_track_metas.is_empty() {
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
for meta in &tier2_track_metas {
let _ = db.set_track_meta(meta);
}
}
}
}
let still_missing: Vec<u64> = missing_ids
@@ -1145,7 +1168,7 @@ async fn resolve_tracks(
.copied()
.collect();
// Tier 3: Qobuz API
// Tier 3: Qobuz API — fetch in parallel, defer cache writes
let mut api_resolved: HashMap<u64, TrackDisplayMeta> = HashMap::new();
if !still_missing.is_empty() {
log::info!(
@@ -1154,15 +1177,11 @@ async fn resolve_tracks(
);
let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(10));
let client = app_state.client.clone();
let cache_arc = cache_state.cache.clone();
let reco_arc = reco_state.db.clone();
let mut handles = Vec::new();
for track_id in &still_missing {
let sem = sem.clone();
let client = client.clone();
let cache_arc = cache_arc.clone();
let reco_arc = reco_arc.clone();
let track_id = *track_id;
handles.push(tokio::spawn(async move {
@@ -1172,29 +1191,42 @@ async fn resolve_tracks(
c.get_track(track_id).await.ok()?
};
let meta = track_to_display_meta(&track);
if let Ok(json) = serde_json::to_string(&track) {
let guard__ = cache_arc.lock().await;
if let Some(cache) = guard__.as_ref() {
let _ = cache.set_track(track_id, &json);
}
}
{
let guard__ = reco_arc.lock().await;
if let Some(db) = guard__.as_ref() {
let _ = db.set_track_meta(&meta);
}
}
Some((track_id, meta))
let json = serde_json::to_string(&track).ok();
Some((track_id, meta, json))
}));
}
// Collect results, then batch-write caches
let mut api_tracks_to_cache: Vec<(u64, String)> = Vec::new();
let mut api_track_metas: Vec<TrackDisplayMeta> = Vec::new();
for handle in handles {
if let Ok(Some((id, meta))) = handle.await {
if let Ok(Some((id, meta, json))) = handle.await {
if let Some(j) = json {
api_tracks_to_cache.push((id, j));
}
api_track_metas.push(meta.clone());
api_resolved.insert(id, meta);
}
}
// Batch write to API cache (single lock)
if !api_tracks_to_cache.is_empty() {
let guard__ = cache_state.cache.lock().await;
if let Some(cache) = guard__.as_ref() {
for (track_id, json) in &api_tracks_to_cache {
let _ = cache.set_track(*track_id, json);
}
}
}
// Batch write to reco meta (single lock)
if !api_track_metas.is_empty() {
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
for meta in &api_track_metas {
let _ = db.set_track_meta(meta);
}
}
}
}
let result: Vec<TrackDisplayMeta> = ids
@@ -1246,6 +1278,7 @@ async fn resolve_artists(
};
let mut api_cache_resolved: HashMap<u64, ArtistCardMeta> = HashMap::new();
let mut tier2_metas_to_write: Vec<ArtistCardMeta> = Vec::new();
if !missing_ids.is_empty() {
let cached = {
let guard__ = cache_state.cache.lock().await;
@@ -1257,15 +1290,19 @@ async fn resolve_artists(
for (artist_id, json_str) in cached {
if let Ok(artist) = serde_json::from_str::<Artist>(&json_str) {
let meta = artist_to_card_meta(&artist, None);
{
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
let _ = db.set_artist_meta(&meta);
}
}
tier2_metas_to_write.push(meta.clone());
api_cache_resolved.insert(artist_id, meta);
}
}
// Batch write tier-2 hits to reco meta (single lock acquisition)
if !tier2_metas_to_write.is_empty() {
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
for meta in &tier2_metas_to_write {
let _ = db.set_artist_meta(meta);
}
}
}
}
let still_missing: Vec<u64> = missing_ids
@@ -1274,7 +1311,7 @@ async fn resolve_artists(
.copied()
.collect();
// Tier 3: Qobuz API
// Tier 3: Qobuz API — fetch in parallel, defer cache writes to after all complete
let mut api_resolved: HashMap<u64, ArtistCardMeta> = HashMap::new();
if !still_missing.is_empty() {
log::info!(
@@ -1283,17 +1320,11 @@ async fn resolve_artists(
);
let sem = std::sync::Arc::new(tokio::sync::Semaphore::new(10));
let client = app_state.client.clone();
let cache_arc = cache_state.cache.clone();
let reco_arc = reco_state.db.clone();
let locale_clone = locale.clone();
let mut handles = Vec::new();
for artist_id in &still_missing {
let sem = sem.clone();
let client = client.clone();
let cache_arc = cache_arc.clone();
let reco_arc = reco_arc.clone();
let locale = locale_clone.clone();
let artist_id = *artist_id;
handles.push(tokio::spawn(async move {
@@ -1303,29 +1334,42 @@ async fn resolve_artists(
c.get_artist_basic(artist_id).await.ok()?
};
let meta = artist_to_card_meta(&artist, None);
if let Ok(json) = serde_json::to_string(&artist) {
let guard__ = cache_arc.lock().await;
if let Some(cache) = guard__.as_ref() {
let _ = cache.set_artist(artist_id, &locale, &json);
}
}
{
let guard__ = reco_arc.lock().await;
if let Some(db) = guard__.as_ref() {
let _ = db.set_artist_meta(&meta);
}
}
Some((artist_id, meta))
let json = serde_json::to_string(&artist).ok();
Some((artist_id, meta, json))
}));
}
// Collect results without holding any locks
let mut api_artists_to_cache: Vec<(u64, String)> = Vec::new();
let mut api_metas_to_write: Vec<ArtistCardMeta> = Vec::new();
for handle in handles {
if let Ok(Some((id, meta))) = handle.await {
if let Ok(Some((id, meta, json))) = handle.await {
if let Some(j) = json {
api_artists_to_cache.push((id, j));
}
api_metas_to_write.push(meta.clone());
api_resolved.insert(id, meta);
}
}
// Batch write to API cache (single lock)
if !api_artists_to_cache.is_empty() {
let guard__ = cache_state.cache.lock().await;
if let Some(cache) = guard__.as_ref() {
for (artist_id, json) in &api_artists_to_cache {
let _ = cache.set_artist(*artist_id, &locale, json);
}
}
}
// Batch write to reco meta (single lock)
if !api_metas_to_write.is_empty() {
let guard__ = reco_state.db.lock().await;
if let Some(db) = guard__.as_ref() {
for meta in &api_metas_to_write {
let _ = db.set_artist_meta(meta);
}
}
}
}
// Assemble in seed order, attaching play_counts
src-tauri/src/reco_store/db.rs
+15
View File
@@ -64,6 +64,17 @@ impl RecoStoreDb {
CREATE INDEX IF NOT EXISTS idx_reco_events_artist ON reco_events(artist_id);
CREATE INDEX IF NOT EXISTS idx_reco_events_created ON reco_events(created_at);
-- Composite indexes for GROUP BY + ORDER BY queries in get_home_seeds
CREATE INDEX IF NOT EXISTS idx_reco_events_play_albums
ON reco_events(event_type, album_id, created_at DESC)
WHERE album_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_reco_events_play_tracks
ON reco_events(event_type, track_id, created_at DESC)
WHERE track_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_reco_events_play_artists
ON reco_events(event_type, artist_id, created_at DESC)
WHERE artist_id IS NOT NULL;
CREATE TABLE IF NOT EXISTS reco_scores (
id INTEGER PRIMARY KEY AUTOINCREMENT,
score_type TEXT NOT NULL,
@@ -79,6 +90,10 @@ impl RecoStoreDb {
CREATE INDEX IF NOT EXISTS idx_reco_scores_track ON reco_scores(track_id);
CREATE INDEX IF NOT EXISTS idx_reco_scores_album ON reco_scores(album_id);
CREATE INDEX IF NOT EXISTS idx_reco_scores_artist ON reco_scores(artist_id);
-- Composite index for scored lookups (score_type + item_type + ORDER BY score)
CREATE INDEX IF NOT EXISTS idx_reco_scores_lookup
ON reco_scores(score_type, item_type, score DESC);
"#,
)
.map_err(|e| format!("Failed to initialize reco database: {}", e))?;