danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge pull request #286 from kettui/fix/improve-graceful-shutdown

Browse source 
Improve shutdown procedures so that the application can close gracefully
This commit is contained in:
BoulderBadgeDad 2026-04-12 12:42:00 -07:00 committed by GitHub
commit 4f25cf4661
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 579 additions and 168 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
core/audiodb_worker.py
View file

@ -8,6 +8,7 @@ from datetime import datetime, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.audiodb_client import AudioDBClient from core.audiodb_client import AudioDBClient
from core.worker_utils import interruptible_sleep
logger = get_logger("audiodb_worker") logger = get_logger("audiodb_worker")
@ -24,6 +25,7 @@ class AudioDBWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -52,6 +54,7 @@ class AudioDBWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("AudioDB background worker started") logger.info("AudioDB background worker started")
@ -64,9 +67,10 @@ class AudioDBWorker:
logger.info("Stopping AudioDB worker...") logger.info("Stopping AudioDB worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("AudioDB worker stopped") logger.info("AudioDB worker stopped")
@ -115,7 +119,7 @@ class AudioDBWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
self.current_item = None self.current_item = None
@ -124,7 +128,7 @@ class AudioDBWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -143,11 +147,11 @@ class AudioDBWorker:
self._process_item(item) self._process_item(item)
time.sleep(2) interruptible_sleep(self._stop_event, 2)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
logger.info("AudioDB worker thread finished") logger.info("AudioDB worker thread finished")

14
core/deezer_worker.py
View file

@ -8,6 +8,7 @@ from datetime import datetime, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.deezer_client import DeezerClient from core.deezer_client import DeezerClient
from core.worker_utils import interruptible_sleep
logger = get_logger("deezer_worker") logger = get_logger("deezer_worker")
@ -24,6 +25,7 @@ class DeezerWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -52,6 +54,7 @@ class DeezerWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Deezer background worker started") logger.info("Deezer background worker started")
@ -64,9 +67,10 @@ class DeezerWorker:
logger.info("Stopping Deezer worker...") logger.info("Stopping Deezer worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Deezer worker stopped") logger.info("Deezer worker stopped")
@ -115,7 +119,7 @@ class DeezerWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
self.current_item = None self.current_item = None
@ -124,7 +128,7 @@ class DeezerWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -143,11 +147,11 @@ class DeezerWorker:
self._process_item(item) self._process_item(item)
time.sleep(2) interruptible_sleep(self._stop_event, 2)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
logger.info("Deezer worker thread finished") logger.info("Deezer worker thread finished")

14
core/discogs_worker.py
View file

@ -18,6 +18,7 @@ from datetime import datetime, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.discogs_client import DiscogsClient from core.discogs_client import DiscogsClient
from core.worker_utils import interruptible_sleep
logger = get_logger("discogs_worker") logger = get_logger("discogs_worker")
@ -34,6 +35,7 @@ class DiscogsWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -62,6 +64,7 @@ class DiscogsWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Discogs background worker started") logger.info("Discogs background worker started")
@ -73,8 +76,9 @@ class DiscogsWorker:
logger.info("Stopping Discogs worker...") logger.info("Stopping Discogs worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Discogs worker stopped") logger.info("Discogs worker stopped")
def pause(self): def pause(self):
@ -113,14 +117,14 @@ class DiscogsWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
self.current_item = None self.current_item = None
item = self._get_next_item() item = self._get_next_item()
if not item: if not item:
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item.get('name', '') self.current_item = item.get('name', '')
@ -132,11 +136,11 @@ class DiscogsWorker:
continue continue
self._process_item(item) self._process_item(item)
time.sleep(2) interruptible_sleep(self._stop_event, 2)
except Exception as e: except Exception as e:
logger.error(f"Error in Discogs worker loop: {e}") logger.error(f"Error in Discogs worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
logger.info("Discogs worker thread finished") logger.info("Discogs worker thread finished")

16
core/genius_worker.py
View file

@ -9,6 +9,7 @@ from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.genius_client import GeniusClient from core.genius_client import GeniusClient
from config.settings import config_manager from config.settings import config_manager
from core.worker_utils import interruptible_sleep
logger = get_logger("genius_worker") logger = get_logger("genius_worker")
@ -31,6 +32,7 @@ class GeniusWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -64,6 +66,7 @@ class GeniusWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Genius background worker started") logger.info("Genius background worker started")
@ -76,9 +79,10 @@ class GeniusWorker:
logger.info("Stopping Genius worker...") logger.info("Stopping Genius worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Genius worker stopped") logger.info("Genius worker stopped")
@ -123,14 +127,14 @@ class GeniusWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# Check if access token is configured # Check if access token is configured
if not self.client.access_token: if not self.client.access_token:
self._init_client() self._init_client()
if not self.client.access_token: if not self.client.access_token:
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
self.current_item = None self.current_item = None
@ -138,7 +142,7 @@ class GeniusWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -157,11 +161,11 @@ class GeniusWorker:
self._process_item(item) self._process_item(item)
# Genius rate limiting is conservative (500ms per call) + lyrics scraping # Genius rate limiting is conservative (500ms per call) + lyrics scraping
time.sleep(1) interruptible_sleep(self._stop_event, 1)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
logger.info("Genius worker thread finished") logger.info("Genius worker thread finished")

12
core/hydrabase_worker.py
View file

@ -15,6 +15,7 @@ import time
import uuid import uuid
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
from core.worker_utils import interruptible_sleep
class HydrabaseWorker: class HydrabaseWorker:
@ -31,6 +32,7 @@ class HydrabaseWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Queue with cap # Queue with cap
self.queue = queue.Queue(maxsize=1000) self.queue = queue.Queue(maxsize=1000)
@ -47,6 +49,7 @@ class HydrabaseWorker:
return return
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Hydrabase P2P mirror worker started") logger.info("Hydrabase P2P mirror worker started")
@ -56,8 +59,9 @@ class HydrabaseWorker:
return return
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Hydrabase P2P mirror worker stopped") logger.info("Hydrabase P2P mirror worker stopped")
def pause(self): def pause(self):
@ -102,7 +106,7 @@ class HydrabaseWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# Non-blocking dequeue with timeout # Non-blocking dequeue with timeout
@ -112,12 +116,12 @@ class HydrabaseWorker:
continue continue
self._process_item(item) self._process_item(item)
time.sleep(0.5) # Rate limit interruptible_sleep(self._stop_event, 0.5) # Rate limit
except Exception as e: except Exception as e:
logger.error(f"Error in Hydrabase worker loop: {e}") logger.error(f"Error in Hydrabase worker loop: {e}")
self.stats['errors'] += 1 self.stats['errors'] += 1
time.sleep(2) interruptible_sleep(self._stop_event, 2)
def _process_item(self, item): def _process_item(self, item):
ws, lock = self.get_ws_and_lock() ws, lock = self.get_ws_and_lock()

20
core/itunes_worker.py
View file

@ -8,6 +8,7 @@ from datetime import datetime, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.itunes_client import iTunesClient from core.itunes_client import iTunesClient
from core.worker_utils import interruptible_sleep
logger = get_logger("itunes_worker") logger = get_logger("itunes_worker")
@ -34,6 +35,7 @@ class iTunesWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -66,6 +68,7 @@ class iTunesWorker:
return return
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("iTunes background worker started") logger.info("iTunes background worker started")
@ -76,8 +79,9 @@ class iTunesWorker:
logger.info("Stopping iTunes worker...") logger.info("Stopping iTunes worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("iTunes worker stopped") logger.info("iTunes worker stopped")
def pause(self): def pause(self):
@ -116,7 +120,7 @@ class iTunesWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# No auth check needed — iTunes API requires no authentication # No auth check needed — iTunes API requires no authentication
@ -126,7 +130,7 @@ class iTunesWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -147,13 +151,13 @@ class iTunesWorker:
# Sleep depends on item type — search items need more delay # Sleep depends on item type — search items need more delay
item_type = item.get('type', '') item_type = item.get('type', '')
if item_type in ('album_batch', 'track_batch'): if item_type in ('album_batch', 'track_batch'):
time.sleep(self.batch_inter_item_sleep) interruptible_sleep(self._stop_event, self.batch_inter_item_sleep)
else: else:
time.sleep(self.inter_item_sleep) interruptible_sleep(self._stop_event, self.inter_item_sleep)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
self.current_item = None self.current_item = None
logger.info("iTunes worker thread finished") logger.info("iTunes worker thread finished")
@ -444,7 +448,7 @@ class iTunesWorker:
self._mark_status('album', db_id, 'not_found') self._mark_status('album', db_id, 'not_found')
self.stats['not_found'] += 1 self.stats['not_found'] += 1
time.sleep(self.batch_inter_item_sleep) interruptible_sleep(self._stop_event, self.batch_inter_item_sleep)
logger.info(f"Album batch for '{artist_name}': {matched_count}/{len(db_albums)} matched") logger.info(f"Album batch for '{artist_name}': {matched_count}/{len(db_albums)} matched")
@ -516,7 +520,7 @@ class iTunesWorker:
self._mark_status('track', db_id, 'not_found') self._mark_status('track', db_id, 'not_found')
self.stats['not_found'] += 1 self.stats['not_found'] += 1
time.sleep(self.batch_inter_item_sleep) interruptible_sleep(self._stop_event, self.batch_inter_item_sleep)
logger.info(f"Track batch for '{album_name}': {matched_count}/{len(db_tracks)} matched") logger.info(f"Track batch for '{album_name}': {matched_count}/{len(db_tracks)} matched")

16
core/lastfm_worker.py
View file

@ -9,6 +9,7 @@ from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.lastfm_client import LastFMClient from core.lastfm_client import LastFMClient
from config.settings import config_manager from config.settings import config_manager
from core.worker_utils import interruptible_sleep
logger = get_logger("lastfm_worker") logger = get_logger("lastfm_worker")
@ -31,6 +32,7 @@ class LastFMWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -65,6 +67,7 @@ class LastFMWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Last.fm background worker started") logger.info("Last.fm background worker started")
@ -77,9 +80,10 @@ class LastFMWorker:
logger.info("Stopping Last.fm worker...") logger.info("Stopping Last.fm worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Last.fm worker stopped") logger.info("Last.fm worker stopped")
@ -124,14 +128,14 @@ class LastFMWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# Check if API key is configured # Check if API key is configured
if not self.client.api_key: if not self.client.api_key:
self._init_client() self._init_client()
if not self.client.api_key: if not self.client.api_key:
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
self.current_item = None self.current_item = None
@ -139,7 +143,7 @@ class LastFMWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -158,11 +162,11 @@ class LastFMWorker:
self._process_item(item) self._process_item(item)
# Last.fm allows 5 req/sec but we use multiple calls per item # Last.fm allows 5 req/sec but we use multiple calls per item
time.sleep(1) interruptible_sleep(self._stop_event, 1)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
logger.info("Last.fm worker thread finished") logger.info("Last.fm worker thread finished")

24
core/listening_stats_worker.py
View file

@ -11,6 +11,7 @@ import time
from typing import Dict, Any from typing import Dict, Any
from utils.logging_config import get_logger from utils.logging_config import get_logger
from core.worker_utils import interruptible_sleep
logger = get_logger("listening_stats_worker") logger = get_logger("listening_stats_worker")
@ -32,6 +33,7 @@ class ListeningStatsWorker:
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self.current_item = None self.current_item = None
self._stop_event = threading.Event()
# Stats # Stats
self.stats = { self.stats = {
@ -52,6 +54,7 @@ class ListeningStatsWorker:
return return
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Listening stats worker started") logger.info("Listening stats worker started")
@ -61,8 +64,9 @@ class ListeningStatsWorker:
return return
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Listening stats worker stopped") logger.info("Listening stats worker stopped")
def pause(self): def pause(self):
@ -86,25 +90,30 @@ class ListeningStatsWorker:
logger.info("Listening stats worker thread started") logger.info("Listening stats worker thread started")
# Build cache from existing data immediately (before first poll) # Build cache from existing data immediately (before first poll)
time.sleep(5) if interruptible_sleep(self._stop_event, 5):
return
try: try:
self._build_stats_cache() self._build_stats_cache()
logger.info("Initial stats cache built from existing data") logger.info("Initial stats cache built from existing data")
except Exception as e: except Exception as e:
logger.debug(f"Initial cache build skipped: {e}") logger.debug(f"Initial cache build skipped: {e}")
if self.should_stop:
return
# Wait before first poll # Wait before first poll
time.sleep(10) if interruptible_sleep(self._stop_event, 10):
return
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(5) interruptible_sleep(self._stop_event, 5)
continue continue
# Check if enabled # Check if enabled
if not self.config_manager.get('listening_stats.enabled', True): if not self.config_manager.get('listening_stats.enabled', True):
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
# Update poll interval from config # Update poll interval from config
@ -119,12 +128,13 @@ class ListeningStatsWorker:
for _ in range(int(self.poll_interval)): for _ in range(int(self.poll_interval)):
if self.should_stop: if self.should_stop:
break break
time.sleep(1) if interruptible_sleep(self._stop_event, 1):
break
except Exception as e: except Exception as e:
logger.error(f"Error in listening stats worker: {e}", exc_info=True) logger.error(f"Error in listening stats worker: {e}", exc_info=True)
self.stats['errors'] += 1 self.stats['errors'] += 1
time.sleep(60) interruptible_sleep(self._stop_event, 60)
self.current_item = None self.current_item = None
logger.info("Listening stats worker thread finished") logger.info("Listening stats worker thread finished")

19
core/media_scan_manager.py
View file

@ -39,6 +39,7 @@ class MediaScanManager:
self._periodic_update_timer = None # Timer for 5-minute periodic updates self._periodic_update_timer = None # Timer for 5-minute periodic updates
self._periodic_update_interval = 300 # 5 minutes in seconds self._periodic_update_interval = 300 # 5 minutes in seconds
self._is_doing_periodic_updates = False # Track if we're in periodic update mode self._is_doing_periodic_updates = False # Track if we're in periodic update mode
self._shutting_down = False
logger.info(f"MediaScanManager initialized with {delay_seconds}s debounce delay") logger.info(f"MediaScanManager initialized with {delay_seconds}s debounce delay")
@ -119,6 +120,9 @@ class MediaScanManager:
""" """
logger.info(f"DEBUG: Media scan requested - reason: {reason}") logger.info(f"DEBUG: Media scan requested - reason: {reason}")
with self._lock: with self._lock:
if self._shutting_down:
logger.debug("Media scan request ignored during shutdown")
return
if self._scan_in_progress: if self._scan_in_progress:
# Server is currently scanning - mark that we need another scan later # Server is currently scanning - mark that we need another scan later
self._downloads_during_scan = True self._downloads_during_scan = True
@ -134,6 +138,7 @@ class MediaScanManager:
# Start the debounce timer # Start the debounce timer
self._timer = threading.Timer(self.delay, self._execute_scan) self._timer = threading.Timer(self.delay, self._execute_scan)
self._timer.daemon = True
self._timer.start() self._timer.start()
def add_scan_completion_callback(self, callback): def add_scan_completion_callback(self, callback):
@ -164,6 +169,9 @@ class MediaScanManager:
def _execute_scan(self): def _execute_scan(self):
"""Execute the actual media library scan""" """Execute the actual media library scan"""
with self._lock: with self._lock:
if self._shutting_down:
logger.debug("Media scan execution skipped during shutdown")
return
if self._scan_in_progress: if self._scan_in_progress:
logger.warning("Scan already in progress - skipping duplicate execution") logger.warning("Scan already in progress - skipping duplicate execution")
return return
@ -211,6 +219,7 @@ class MediaScanManager:
# Schedule first periodic update after 5 minutes # Schedule first periodic update after 5 minutes
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update) self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start() self._periodic_update_timer.start()
except Exception as e: except Exception as e:
@ -247,15 +256,22 @@ class MediaScanManager:
if is_scanning: if is_scanning:
# Still scanning - trigger database update and continue periodic updates # Still scanning - trigger database update and continue periodic updates
logger.info(f"{server_type.upper()} still scanning - triggering database update") logger.info(f"{server_type.upper()} still scanning - triggering database update")
if self._shutting_down:
return
self._call_completion_callbacks() self._call_completion_callbacks()
# Schedule next periodic update # Schedule next periodic update
if self._shutting_down:
return
logger.info(f"Scheduling next periodic update in {self._periodic_update_interval//60} minutes") logger.info(f"Scheduling next periodic update in {self._periodic_update_interval//60} minutes")
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update) self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start() self._periodic_update_timer.start()
else: else:
# Scanning stopped - final update and cleanup # Scanning stopped - final update and cleanup
logger.info(f"{server_type.upper()} scanning completed - doing final database update") logger.info(f"{server_type.upper()} scanning completed - doing final database update")
if self._shutting_down:
return
self._call_completion_callbacks() self._call_completion_callbacks()
self._stop_periodic_updates() self._stop_periodic_updates()
@ -360,6 +376,7 @@ class MediaScanManager:
def shutdown(self): def shutdown(self):
"""Clean shutdown - cancel any pending timers""" """Clean shutdown - cancel any pending timers"""
with self._lock: with self._lock:
self._shutting_down = True
if self._timer: if self._timer:
self._timer.cancel() self._timer.cancel()
self._timer = None self._timer = None
@ -369,4 +386,4 @@ class MediaScanManager:
self._periodic_update_timer = None self._periodic_update_timer = None
self._is_doing_periodic_updates = False self._is_doing_periodic_updates = False
logger.info("MediaScanManager shutdown - cancelled all pending timers") logger.info("MediaScanManager shutdown - cancelled all pending timers")

14
core/musicbrainz_worker.py
View file

@ -5,6 +5,7 @@ from datetime import datetime, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.musicbrainz_service import MusicBrainzService from core.musicbrainz_service import MusicBrainzService
from core.worker_utils import interruptible_sleep
logger = get_logger("musicbrainz_worker") logger = get_logger("musicbrainz_worker")
@ -20,6 +21,7 @@ class MusicBrainzWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -45,6 +47,7 @@ class MusicBrainzWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("MusicBrainz background worker started") logger.info("MusicBrainz background worker started")
@ -57,9 +60,10 @@ class MusicBrainzWorker:
logger.info("Stopping MusicBrainz worker...") logger.info("Stopping MusicBrainz worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Music Brainz worker stopped") logger.info("Music Brainz worker stopped")
@ -112,7 +116,7 @@ class MusicBrainzWorker:
try: try:
# Check if paused # Check if paused
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# Clear previous item before getting next # Clear previous item before getting next
@ -124,7 +128,7 @@ class MusicBrainzWorker:
if not item: if not item:
# No more items - sleep for a bit # No more items - sleep for a bit
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
# Set current item for UI tracking # Set current item for UI tracking
@ -147,11 +151,11 @@ class MusicBrainzWorker:
# Keep current_item set during sleep so UI can see what was just processed # Keep current_item set during sleep so UI can see what was just processed
# Rate limit: 1 request per second # Rate limit: 1 request per second
time.sleep(1) interruptible_sleep(self._stop_event, 1)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) # Back off on errors interruptible_sleep(self._stop_event, 5) # Back off on errors
logger.info("MusicBrainz worker thread finished") logger.info("MusicBrainz worker thread finished")

23
core/plex_scan_manager.py
View file

@ -39,6 +39,7 @@ class PlexScanManager:
self._periodic_update_timer = None # Timer for 5-minute periodic updates self._periodic_update_timer = None # Timer for 5-minute periodic updates
self._periodic_update_interval = 300 # 5 minutes in seconds self._periodic_update_interval = 300 # 5 minutes in seconds
self._is_doing_periodic_updates = False # Track if we're in periodic update mode self._is_doing_periodic_updates = False # Track if we're in periodic update mode
self._shutting_down = False
logger.info(f"PlexScanManager initialized with {delay_seconds}s debounce delay") logger.info(f"PlexScanManager initialized with {delay_seconds}s debounce delay")
@ -51,6 +52,9 @@ class PlexScanManager:
""" """
logger.info(f"DEBUG: Plex scan requested - reason: {reason}") logger.info(f"DEBUG: Plex scan requested - reason: {reason}")
with self._lock: with self._lock:
if self._shutting_down:
logger.debug("Plex scan request ignored during shutdown")
return
if self._scan_in_progress: if self._scan_in_progress:
# Plex is currently scanning - mark that we need another scan later # Plex is currently scanning - mark that we need another scan later
self._downloads_during_scan = True self._downloads_during_scan = True
@ -66,6 +70,7 @@ class PlexScanManager:
# Start the debounce timer # Start the debounce timer
self._timer = threading.Timer(self.delay, self._execute_scan) self._timer = threading.Timer(self.delay, self._execute_scan)
self._timer.daemon = True
self._timer.start() self._timer.start()
def add_scan_completion_callback(self, callback): def add_scan_completion_callback(self, callback):
@ -96,6 +101,9 @@ class PlexScanManager:
def _execute_scan(self): def _execute_scan(self):
"""Execute the actual Plex library scan""" """Execute the actual Plex library scan"""
with self._lock: with self._lock:
if self._shutting_down:
logger.debug("Plex scan execution skipped during shutdown")
return
if self._scan_in_progress: if self._scan_in_progress:
logger.warning("Scan already in progress - skipping duplicate execution") logger.warning("Scan already in progress - skipping duplicate execution")
return return
@ -136,6 +144,7 @@ class PlexScanManager:
# Schedule first periodic update after 5 minutes # Schedule first periodic update after 5 minutes
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update) self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start() self._periodic_update_timer.start()
except Exception as e: except Exception as e:
@ -165,15 +174,22 @@ class PlexScanManager:
if is_scanning: if is_scanning:
# Still scanning - trigger database update and continue periodic updates # Still scanning - trigger database update and continue periodic updates
logger.info("Plex still scanning - triggering database update") logger.info("Plex still scanning - triggering database update")
if self._shutting_down:
return
self._call_completion_callbacks() self._call_completion_callbacks()
# Schedule next periodic update # Schedule next periodic update
if self._shutting_down:
return
logger.info(f"Scheduling next periodic update in {self._periodic_update_interval//60} minutes") logger.info(f"Scheduling next periodic update in {self._periodic_update_interval//60} minutes")
self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update) self._periodic_update_timer = threading.Timer(self._periodic_update_interval, self._do_periodic_update)
self._periodic_update_timer.daemon = True
self._periodic_update_timer.start() self._periodic_update_timer.start()
else: else:
# Scanning stopped - final update and cleanup # Scanning stopped - final update and cleanup
logger.info("Plex scanning completed - doing final database update") logger.info("Plex scanning completed - doing final database update")
if self._shutting_down:
return
self._call_completion_callbacks() self._call_completion_callbacks()
self._stop_periodic_updates() self._stop_periodic_updates()
@ -218,7 +234,9 @@ class PlexScanManager:
if is_scanning: if is_scanning:
# Still scanning, poll again in 30 seconds # Still scanning, poll again in 30 seconds
logger.info("DEBUG: Plex library still scanning, will check again in 30 seconds") logger.info("DEBUG: Plex library still scanning, will check again in 30 seconds")
threading.Timer(30, self._poll_scan_status).start() timer = threading.Timer(30, self._poll_scan_status)
timer.daemon = True
timer.start()
else: else:
# Scan completed! # Scan completed!
elapsed_time = time.time() - self._scan_start_time if self._scan_start_time else 0 elapsed_time = time.time() - self._scan_start_time if self._scan_start_time else 0
@ -311,6 +329,7 @@ class PlexScanManager:
def shutdown(self): def shutdown(self):
"""Clean shutdown - cancel any pending timers""" """Clean shutdown - cancel any pending timers"""
with self._lock: with self._lock:
self._shutting_down = True
if self._timer: if self._timer:
self._timer.cancel() self._timer.cancel()
self._timer = None self._timer = None
@ -320,4 +339,4 @@ class PlexScanManager:
self._periodic_update_timer = None self._periodic_update_timer = None
self._is_doing_periodic_updates = False self._is_doing_periodic_updates = False
logger.info("PlexScanManager shutdown - cancelled all pending timers") logger.info("PlexScanManager shutdown - cancelled all pending timers")

22
core/qobuz_worker.py
View file

@ -8,6 +8,7 @@ from datetime import datetime, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.qobuz_client import _qobuz_is_rate_limited from core.qobuz_client import _qobuz_is_rate_limited
from core.worker_utils import interruptible_sleep
logger = get_logger("qobuz_worker") logger = get_logger("qobuz_worker")
@ -24,6 +25,7 @@ class QobuzWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -52,6 +54,7 @@ class QobuzWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Qobuz background worker started") logger.info("Qobuz background worker started")
@ -64,9 +67,10 @@ class QobuzWorker:
logger.info("Stopping Qobuz worker...") logger.info("Stopping Qobuz worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Qobuz worker stopped") logger.info("Qobuz worker stopped")
@ -120,24 +124,24 @@ class QobuzWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# Auth guard: sleep if not authenticated # Auth guard: sleep if not authenticated
try: try:
if not self.client or not self.client.is_authenticated(): if not self.client or not self.client.is_authenticated():
self.current_item = None self.current_item = None
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
except Exception: except Exception:
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
# Rate limit guard: back off if globally rate limited # Rate limit guard: back off if globally rate limited
if _qobuz_is_rate_limited(): if _qobuz_is_rate_limited():
self.current_item = None self.current_item = None
logger.debug("Qobuz rate limited, backing off...") logger.debug("Qobuz rate limited, backing off...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = None self.current_item = None
@ -146,7 +150,7 @@ class QobuzWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -166,11 +170,11 @@ class QobuzWorker:
self._process_item(item) self._process_item(item)
# Throttle between API calls # Throttle between API calls
time.sleep(2) interruptible_sleep(self._stop_event, 2)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
logger.info("Qobuz worker thread finished") logger.info("Qobuz worker thread finished")
@ -351,7 +355,7 @@ class QobuzWorker:
error_str = str(e).lower() error_str = str(e).lower()
if '429' in error_str or 'rate limit' in error_str: if '429' in error_str or 'rate limit' in error_str:
logger.warning(f"Rate limited while processing {item['type']} #{item['id']}, backing off 30s") logger.warning(f"Rate limited while processing {item['type']} #{item['id']}, backing off 30s")
time.sleep(30) interruptible_sleep(self._stop_event, 30)
return return
logger.error(f"Error processing {item['type']} #{item['id']}: {e}") logger.error(f"Error processing {item['type']} #{item['id']}: {e}")
self.stats['errors'] += 1 self.stats['errors'] += 1

3
core/repair_jobs/acoustid_scanner.py
View file

@ -141,7 +141,8 @@ class AcoustIDScannerJob(RepairJob):
if batch_count >= batch_size: if batch_count >= batch_size:
batch_count = 0 batch_count = 0
self._save_checkpoint(context, fpath) self._save_checkpoint(context, fpath)
time.sleep(2) if context.sleep_or_stop(2):
return result
if context.update_progress and (i + 1) % 10 == 0: if context.update_progress and (i + 1) % 10 == 0:
context.update_progress(i + 1, total) context.update_progress(i + 1, total)

28
core/repair_jobs/base.py
View file

@ -2,6 +2,7 @@
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from dataclasses import dataclass, field from dataclasses import dataclass, field
import threading
from typing import Any, Callable, Dict, List, Optional from typing import Any, Callable, Dict, List, Optional
@ -29,6 +30,7 @@ class JobContext:
mb_client: Any = None mb_client: Any = None
acoustid_client: Any = None acoustid_client: Any = None
metadata_cache: Any = None metadata_cache: Any = None
stop_event: Optional[threading.Event] = None
# Callbacks # Callbacks
create_finding: Optional[Callable] = None create_finding: Optional[Callable] = None
@ -39,6 +41,8 @@ class JobContext:
def check_stop(self) -> bool: def check_stop(self) -> bool:
"""Return True if the worker should stop.""" """Return True if the worker should stop."""
if self.stop_event and self.stop_event.is_set():
return True
return self.should_stop() if self.should_stop else False return self.should_stop() if self.should_stop else False
def is_spotify_rate_limited(self) -> bool: def is_spotify_rate_limited(self) -> bool:
@ -55,11 +59,31 @@ class JobContext:
def wait_if_paused(self): def wait_if_paused(self):
"""Block until unpaused or stopped. Returns True if should stop.""" """Block until unpaused or stopped. Returns True if should stop."""
import time
while self.is_paused and self.is_paused(): while self.is_paused and self.is_paused():
if self.check_stop(): if self.check_stop():
return True return True
time.sleep(1) if self.stop_event:
self.stop_event.wait(0.2)
else:
import time
time.sleep(0.2)
return self.check_stop()
def sleep_or_stop(self, seconds: float, step: float = 0.2) -> bool:
"""Sleep in small increments so stop requests can interrupt quickly."""
if seconds <= 0:
return self.check_stop()
remaining = seconds
while remaining > 0:
if self.check_stop():
return True
chunk = min(step, remaining)
if self.stop_event:
self.stop_event.wait(chunk)
else:
import time
time.sleep(chunk)
remaining -= chunk
return self.check_stop() return self.check_stop()

3
core/repair_jobs/library_reorganize.py
View file

@ -821,7 +821,8 @@ class LibraryReorganizeJob(RepairJob):
years[key] = year_str years[key] = year_str
break break
import time import time
time.sleep(0.1) # Rate limit courtesy if context.sleep_or_stop(0.1): # Rate limit courtesy
break
except Exception as e: except Exception as e:
logger.debug("API year lookup failed for %s - %s: %s", artist, album, e) logger.debug("API year lookup failed for %s - %s: %s", artist, album, e)

3
core/repair_jobs/mbid_mismatch_detector.py
View file

@ -294,7 +294,8 @@ class MbidMismatchDetectorJob(RepairJob):
try: try:
# Rate limit: MusicBrainz allows ~1 req/sec # Rate limit: MusicBrainz allows ~1 req/sec
time.sleep(1.1) if context.sleep_or_stop(1.1):
return result
recording = mb_client.get_recording(mbid, includes=['artist-credits']) recording = mb_client.get_recording(mbid, includes=['artist-credits'])
if not recording: if not recording:

3
core/repair_jobs/metadata_gap_filler.py
View file

@ -173,7 +173,8 @@ class MetadataGapFillerJob(RepairJob):
# Rate limit API calls # Rate limit API calls
if spotify_track_id: if spotify_track_id:
time.sleep(0.5) if context.sleep_or_stop(0.5):
return result
if context.update_progress and (i + 1) % 10 == 0: if context.update_progress and (i + 1) % 10 == 0:
context.update_progress(i + 1, total) context.update_progress(i + 1, total)

3
core/repair_jobs/unknown_artist_fixer.py
View file

@ -337,7 +337,8 @@ class UnknownArtistFixerJob(RepairJob):
except Exception as e: except Exception as e:
logger.debug(f"Title search failed for '{title}': {e}") logger.debug(f"Title search failed for '{title}': {e}")
# Rate limit courtesy # Rate limit courtesy
time.sleep(0.2) if context.sleep_or_stop(0.2):
return None
return None return None

34
core/repair_worker.py
View file

@ -67,6 +67,7 @@ class RepairWorker:
self.running = False self.running = False
self.enabled = False # Master toggle (replaces 'paused') self.enabled = False # Master toggle (replaces 'paused')
self.should_stop = False self.should_stop = False
self._stop_event = threading.Event()
self.thread = None self.thread = None
# Current job being executed # Current job being executed
@ -293,6 +294,7 @@ class RepairWorker:
return return
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Repair worker started") logger.info("Repair worker started")
@ -303,8 +305,9 @@ class RepairWorker:
logger.info("Stopping repair worker...") logger.info("Stopping repair worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=2)
logger.info("Repair worker stopped") logger.info("Repair worker stopped")
def toggle(self) -> bool: def toggle(self) -> bool:
@ -418,7 +421,7 @@ class RepairWorker:
logger.info("Repair worker thread started") logger.info("Repair worker thread started")
self._ensure_jobs_loaded() self._ensure_jobs_loaded()
while not self.should_stop: while not self._stop_event.is_set():
try: try:
# Check force-run queue even when disabled (user explicitly requested) # Check force-run queue even when disabled (user explicitly requested)
forced_job = None forced_job = None
@ -428,13 +431,15 @@ class RepairWorker:
if forced_job: if forced_job:
self._run_job(forced_job) self._run_job(forced_job)
time.sleep(2) if self._sleep_or_stop(2):
break
continue continue
if not self.enabled: if not self.enabled:
self._current_job_id = None self._current_job_id = None
self._current_job_name = None self._current_job_name = None
time.sleep(2) if self._sleep_or_stop(2):
break
continue continue
# Find the next job to run based on staleness # Find the next job to run based on staleness
@ -444,20 +449,23 @@ class RepairWorker:
# Nothing due — sleep and re-check # Nothing due — sleep and re-check
self._current_job_id = None self._current_job_id = None
self._current_job_name = None self._current_job_name = None
time.sleep(10) if self._sleep_or_stop(10):
break
continue continue
# Run the selected job # Run the selected job
self._run_job(next_job) self._run_job(next_job)
# Brief pause between jobs # Brief pause between jobs
time.sleep(5) if self._sleep_or_stop(5):
break
except Exception as e: except Exception as e:
logger.error("Error in repair worker loop: %s", e, exc_info=True) logger.error("Error in repair worker loop: %s", e, exc_info=True)
self._current_job_id = None self._current_job_id = None
self._current_job_name = None self._current_job_name = None
time.sleep(30) if self._sleep_or_stop(30):
break
logger.info("Repair worker thread finished") logger.info("Repair worker thread finished")
@ -552,6 +560,7 @@ class RepairWorker:
metadata_cache=self.metadata_cache, metadata_cache=self.metadata_cache,
create_finding=self._create_finding, create_finding=self._create_finding,
should_stop=lambda: self.should_stop, should_stop=lambda: self.should_stop,
stop_event=self._stop_event,
is_paused=lambda: not self.enabled, is_paused=lambda: not self.enabled,
update_progress=self._update_progress, update_progress=self._update_progress,
report_progress=_report_progress, report_progress=_report_progress,
@ -595,6 +604,17 @@ class RepairWorker:
self._current_job_name = None self._current_job_name = None
self._current_progress = {'scanned': 0, 'total': 0, 'percent': 0} self._current_progress = {'scanned': 0, 'total': 0, 'percent': 0}
def _sleep_or_stop(self, seconds: float, step: float = 0.2) -> bool:
"""Sleep in small chunks so shutdown interrupts quickly."""
if seconds <= 0:
return self._stop_event.is_set()
remaining = seconds
while remaining > 0 and not self._stop_event.is_set():
chunk = min(step, remaining)
self._stop_event.wait(chunk)
remaining -= chunk
return self._stop_event.is_set()
def run_job_now(self, job_id: str): def run_job_now(self, job_id: str):
"""Queue a job for immediate execution by the main worker loop. """Queue a job for immediate execution by the main worker loop.

20
core/soulid_worker.py
View file

@ -24,6 +24,7 @@ import unicodedata
from typing import Dict, Any, List, Optional from typing import Dict, Any, List, Optional
from utils.logging_config import get_logger from utils.logging_config import get_logger
from core.worker_utils import interruptible_sleep
logger = get_logger("soulid_worker") logger = get_logger("soulid_worker")
@ -79,6 +80,7 @@ class SoulIDWorker:
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self.current_item = None self.current_item = None
self._stop_event = threading.Event()
# API clients (lazy-initialized) # API clients (lazy-initialized)
self._itunes_client = None self._itunes_client = None
@ -135,6 +137,7 @@ class SoulIDWorker:
return return
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("SoulID worker started") logger.info("SoulID worker started")
@ -144,8 +147,9 @@ class SoulIDWorker:
return return
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("SoulID worker stopped") logger.info("SoulID worker stopped")
def pause(self): def pause(self):
@ -178,7 +182,7 @@ class SoulIDWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
processed = 0 processed = 0
@ -188,16 +192,16 @@ class SoulIDWorker:
if processed == 0: if processed == 0:
self.current_item = None self.current_item = None
time.sleep(self.idle_sleep) interruptible_sleep(self._stop_event, self.idle_sleep)
else: else:
# Albums/tracks get inter_batch_sleep, artists get their # Albums/tracks get inter_batch_sleep, artists get their
# own sleep inside _process_next_artist # own sleep inside _process_next_artist
time.sleep(self.inter_batch_sleep) interruptible_sleep(self._stop_event, self.inter_batch_sleep)
except Exception as e: except Exception as e:
logger.error(f"Error in SoulID worker loop: {e}", exc_info=True) logger.error(f"Error in SoulID worker loop: {e}", exc_info=True)
self.stats['errors'] += 1 self.stats['errors'] += 1
time.sleep(5) interruptible_sleep(self._stop_event, 5)
self.current_item = None self.current_item = None
logger.info("SoulID worker thread finished") logger.info("SoulID worker thread finished")
@ -267,7 +271,7 @@ class SoulIDWorker:
logger.info(f"Generated soul ID for artist: {name}" + (f" (canonical id: {canonical_id})" if canonical_id else "")) logger.info(f"Generated soul ID for artist: {name}" + (f" (canonical id: {canonical_id})" if canonical_id else ""))
# Rate limit courtesy for API calls # Rate limit courtesy for API calls
time.sleep(self.artist_sleep) interruptible_sleep(self._stop_event, self.artist_sleep)
return 1 return 1
except Exception as e: except Exception as e:
@ -324,7 +328,7 @@ class SoulIDWorker:
deezer_artist_id = int(raw_id) deezer_artist_id = int(raw_id)
logger.debug(f"Deezer artist ID for '{artist_name}': {deezer_artist_id}") logger.debug(f"Deezer artist ID for '{artist_name}': {deezer_artist_id}")
break break
time.sleep(0.3) interruptible_sleep(self._stop_event, 0.3)
except Exception as e: except Exception as e:
logger.debug(f"Deezer track search failed for '{artist_name}': {e}") logger.debug(f"Deezer track search failed for '{artist_name}': {e}")
@ -344,7 +348,7 @@ class SoulIDWorker:
itunes_artist_id = int(raw_id) itunes_artist_id = int(raw_id)
logger.debug(f"iTunes artist ID for '{artist_name}': {itunes_artist_id}") logger.debug(f"iTunes artist ID for '{artist_name}': {itunes_artist_id}")
break break
time.sleep(0.3) interruptible_sleep(self._stop_event, 0.3)
except Exception as e: except Exception as e:
logger.debug(f"iTunes track search failed for '{artist_name}': {e}") logger.debug(f"iTunes track search failed for '{artist_name}': {e}")

28
core/spotify_worker.py
View file

@ -8,6 +8,7 @@ from datetime import datetime, date, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.spotify_client import SpotifyClient, SpotifyRateLimitError from core.spotify_client import SpotifyClient, SpotifyRateLimitError
from core.worker_utils import interruptible_sleep
logger = get_logger("spotify_worker") logger = get_logger("spotify_worker")
@ -31,6 +32,7 @@ class SpotifyWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -66,6 +68,7 @@ class SpotifyWorker:
return return
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Spotify background worker started") logger.info("Spotify background worker started")
@ -76,8 +79,9 @@ class SpotifyWorker:
logger.info("Stopping Spotify worker...") logger.info("Stopping Spotify worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Spotify worker stopped") logger.info("Spotify worker stopped")
def pause(self): def pause(self):
@ -167,7 +171,7 @@ class SpotifyWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# Rate limit guard — if globally rate limited, sleep until ban expires # Rate limit guard — if globally rate limited, sleep until ban expires
@ -175,7 +179,7 @@ class SpotifyWorker:
info = self.client.get_rate_limit_info() info = self.client.get_rate_limit_info()
remaining = info['remaining_seconds'] if info else 60 remaining = info['remaining_seconds'] if info else 60
logger.debug(f"Spotify globally rate limited, sleeping {remaining}s...") logger.debug(f"Spotify globally rate limited, sleeping {remaining}s...")
time.sleep(min(remaining, 60)) # Check again every 60s max interruptible_sleep(self._stop_event, min(remaining, 60)) # Check again every 60s max
continue continue
# Daily budget guard — worker-only cap to avoid saturating Spotify rate limits # Daily budget guard — worker-only cap to avoid saturating Spotify rate limits
@ -184,7 +188,7 @@ class SpotifyWorker:
resets_in = budget['resets_in_seconds'] resets_in = budget['resets_in_seconds']
logger.info(f"Daily enrichment budget exhausted ({budget['used']}/{budget['limit']}), " logger.info(f"Daily enrichment budget exhausted ({budget['used']}/{budget['limit']}), "
f"resets in {resets_in // 3600}h {(resets_in % 3600) // 60}m") f"resets in {resets_in // 3600}h {(resets_in % 3600) // 60}m")
time.sleep(min(resets_in, 300)) # Check every 5 min max interruptible_sleep(self._stop_event, min(resets_in, 300)) # Check every 5 min max
continue continue
# Post-ban cooldown guard — after ban expires, wait before resuming # Post-ban cooldown guard — after ban expires, wait before resuming
@ -192,7 +196,7 @@ class SpotifyWorker:
cooldown = self.client.get_post_ban_cooldown_remaining() cooldown = self.client.get_post_ban_cooldown_remaining()
if cooldown > 0: if cooldown > 0:
logger.debug(f"Post-ban cooldown active ({cooldown}s left), sleeping...") logger.debug(f"Post-ban cooldown active ({cooldown}s left), sleeping...")
time.sleep(min(cooldown, 60)) interruptible_sleep(self._stop_event, min(cooldown, 60))
continue continue
# Auth guard — check if Spotify client is configured (no API call). # Auth guard — check if Spotify client is configured (no API call).
@ -203,7 +207,7 @@ class SpotifyWorker:
self.client.reload_config() self.client.reload_config()
if not self.client.is_spotify_authenticated(): if not self.client.is_spotify_authenticated():
logger.debug("Spotify not authenticated, sleeping 30s...") logger.debug("Spotify not authenticated, sleeping 30s...")
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
self.current_item = None self.current_item = None
@ -211,7 +215,7 @@ class SpotifyWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -229,14 +233,14 @@ class SpotifyWorker:
self._process_item(item) self._process_item(item)
self._increment_daily_budget() self._increment_daily_budget()
time.sleep(self.inter_item_sleep) interruptible_sleep(self._stop_event, self.inter_item_sleep)
except SpotifyRateLimitError: except SpotifyRateLimitError:
logger.debug("Spotify rate limit hit in worker loop, will retry after ban expires") logger.debug("Spotify rate limit hit in worker loop, will retry after ban expires")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
self.current_item = None self.current_item = None
logger.info("Spotify worker thread finished") logger.info("Spotify worker thread finished")
@ -541,7 +545,7 @@ class SpotifyWorker:
self._mark_status('album', db_id, 'not_found') self._mark_status('album', db_id, 'not_found')
self.stats['not_found'] += 1 self.stats['not_found'] += 1
time.sleep(self.batch_inter_item_sleep) interruptible_sleep(self._stop_event, self.batch_inter_item_sleep)
logger.info(f"Album batch for '{artist_name}': {matched_count}/{len(db_albums)} matched") logger.info(f"Album batch for '{artist_name}': {matched_count}/{len(db_albums)} matched")
@ -614,7 +618,7 @@ class SpotifyWorker:
self._mark_status('track', db_id, 'not_found') self._mark_status('track', db_id, 'not_found')
self.stats['not_found'] += 1 self.stats['not_found'] += 1
time.sleep(self.batch_inter_item_sleep) interruptible_sleep(self._stop_event, self.batch_inter_item_sleep)
logger.info(f"Track batch for '{album_name}': {matched_count}/{len(db_tracks)} matched") logger.info(f"Track batch for '{album_name}': {matched_count}/{len(db_tracks)} matched")

20
core/tidal_worker.py
View file

@ -7,6 +7,7 @@ from datetime import datetime, timedelta
from utils.logging_config import get_logger from utils.logging_config import get_logger
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
from core.tidal_client import TidalClient from core.tidal_client import TidalClient
from core.worker_utils import interruptible_sleep
logger = get_logger("tidal_worker") logger = get_logger("tidal_worker")
@ -45,6 +46,7 @@ class TidalWorker:
self.paused = False self.paused = False
self.should_stop = False self.should_stop = False
self.thread = None self.thread = None
self._stop_event = threading.Event()
# Current item being processed (for UI tooltip) # Current item being processed (for UI tooltip)
self.current_item = None self.current_item = None
@ -73,6 +75,7 @@ class TidalWorker:
self.running = True self.running = True
self.should_stop = False self.should_stop = False
self._stop_event.clear()
self.thread = threading.Thread(target=self._run, daemon=True) self.thread = threading.Thread(target=self._run, daemon=True)
self.thread.start() self.thread.start()
logger.info("Tidal background worker started") logger.info("Tidal background worker started")
@ -85,9 +88,10 @@ class TidalWorker:
logger.info("Stopping Tidal worker...") logger.info("Stopping Tidal worker...")
self.should_stop = True self.should_stop = True
self.running = False self.running = False
self._stop_event.set()
if self.thread: if self.thread:
self.thread.join(timeout=5) self.thread.join(timeout=1)
logger.info("Tidal worker stopped") logger.info("Tidal worker stopped")
@ -140,17 +144,17 @@ class TidalWorker:
while not self.should_stop: while not self.should_stop:
try: try:
if self.paused: if self.paused:
time.sleep(1) interruptible_sleep(self._stop_event, 1)
continue continue
# Auth guard: sleep if not authenticated # Auth guard: sleep if not authenticated
try: try:
if not self.client.is_authenticated(): if not self.client.is_authenticated():
self.current_item = None self.current_item = None
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
except Exception: except Exception:
time.sleep(30) interruptible_sleep(self._stop_event, 30)
continue continue
self.current_item = None self.current_item = None
@ -159,7 +163,7 @@ class TidalWorker:
if not item: if not item:
logger.debug("No pending items, sleeping...") logger.debug("No pending items, sleeping...")
time.sleep(10) interruptible_sleep(self._stop_event, 10)
continue continue
self.current_item = item self.current_item = item
@ -178,11 +182,11 @@ class TidalWorker:
self._process_item(item) self._process_item(item)
time.sleep(2) interruptible_sleep(self._stop_event, 2)
except Exception as e: except Exception as e:
logger.error(f"Error in worker loop: {e}") logger.error(f"Error in worker loop: {e}")
time.sleep(5) interruptible_sleep(self._stop_event, 5)
logger.info("Tidal worker thread finished") logger.info("Tidal worker thread finished")
@ -364,7 +368,7 @@ class TidalWorker:
if '429' in error_str or 'rate limit' in error_str: if '429' in error_str or 'rate limit' in error_str:
# Rate limit — don't mark as error, back off then retry # Rate limit — don't mark as error, back off then retry
logger.warning(f"Rate limited while processing {item['type']} #{item['id']}, backing off 30s") logger.warning(f"Rate limited while processing {item['type']} #{item['id']}, backing off 30s")
time.sleep(30) interruptible_sleep(self._stop_event, 30)
return return
logger.error(f"Error processing {item['type']} #{item['id']}: {e}") logger.error(f"Error processing {item['type']} #{item['id']}: {e}")
self.stats['errors'] += 1 self.stats['errors'] += 1

73
core/web_scan_manager.py
View file

@ -38,6 +38,8 @@ class WebScanManager:
self._max_scan_time = 1800 # 30 minutes maximum self._max_scan_time = 1800 # 30 minutes maximum
self._current_server_type = None self._current_server_type = None
self._scan_progress = {} self._scan_progress = {}
self._completion_check_timer = None
self._shutting_down = False
logger.info(f"WebScanManager initialized with {delay_seconds}s debounce delay") logger.info(f"WebScanManager initialized with {delay_seconds}s debounce delay")
@ -84,6 +86,14 @@ class WebScanManager:
logger.info(f"Web scan requested - reason: {reason}") logger.info(f"Web scan requested - reason: {reason}")
with self._lock: with self._lock:
if self._shutting_down:
logger.debug("Web scan request ignored during shutdown")
return {
"status": "ignored",
"message": "Server is shutting down",
"delay_seconds": 0,
"reason": reason,
}
# Add callback if provided # Add callback if provided
if callback and callback not in self._scan_completion_callbacks: if callback and callback not in self._scan_completion_callbacks:
self._scan_completion_callbacks.append(callback) self._scan_completion_callbacks.append(callback)
@ -107,6 +117,7 @@ class WebScanManager:
# Start the debounce timer # Start the debounce timer
self._timer = threading.Timer(self.delay, self._execute_scan) self._timer = threading.Timer(self.delay, self._execute_scan)
self._timer.daemon = True
self._timer.start() self._timer.start()
return { return {
@ -169,6 +180,9 @@ class WebScanManager:
def _execute_scan(self): def _execute_scan(self):
"""Execute the actual media library scan""" """Execute the actual media library scan"""
with self._lock: with self._lock:
if self._shutting_down:
logger.debug("Web scan execution skipped during shutdown")
return
if self._scan_in_progress: if self._scan_in_progress:
logger.warning("Web scan already in progress - skipping duplicate execution") logger.warning("Web scan already in progress - skipping duplicate execution")
return return
@ -231,6 +245,9 @@ class WebScanManager:
"""Start periodic checking for scan completion""" """Start periodic checking for scan completion"""
def check_completion(): def check_completion():
try: try:
if self._shutting_down:
logger.debug("Web scan completion check aborted during shutdown")
return
# Check for timeout # Check for timeout
if self._scan_start_time and (time.time() - self._scan_start_time) > self._max_scan_time: if self._scan_start_time and (time.time() - self._scan_start_time) > self._max_scan_time:
logger.warning(f"Web scan timed out after {self._max_scan_time} seconds") logger.warning(f"Web scan timed out after {self._max_scan_time} seconds")
@ -254,23 +271,38 @@ class WebScanManager:
self._handle_scan_completion() self._handle_scan_completion()
else: else:
# Continue checking # Continue checking
threading.Timer(30, check_completion).start() # Check every 30 seconds if self._shutting_down:
return
timer = threading.Timer(30, check_completion) # Check every 30 seconds
timer.daemon = True
self._completion_check_timer = timer
timer.start()
except Exception as e: except Exception as e:
logger.error(f"Error during web scan completion check: {e}") logger.error(f"Error during web scan completion check: {e}")
self._reset_scan_state() self._reset_scan_state()
# Start first check after 30 seconds # Start first check after 30 seconds
threading.Timer(30, check_completion).start() timer = threading.Timer(30, check_completion)
timer.daemon = True
self._completion_check_timer = timer
timer.start()
def _handle_scan_completion(self): def _handle_scan_completion(self):
"""Handle scan completion and trigger callbacks""" """Handle scan completion and trigger callbacks"""
logger.info(f"Web {self._current_server_type.upper()} library scan completed")
# Call completion callbacks
callbacks_to_call = []
with self._lock: with self._lock:
if self._shutting_down:
return
server_type = self._current_server_type
callbacks_to_call = self._scan_completion_callbacks.copy() callbacks_to_call = self._scan_completion_callbacks.copy()
downloads_during_scan = self._downloads_during_scan
if not server_type:
logger.debug("Skipping web scan completion: no active server type")
self._reset_scan_state()
return
logger.info(f"Web {server_type.upper()} library scan completed")
for callback in callbacks_to_call: for callback in callbacks_to_call:
try: try:
@ -283,10 +315,9 @@ class WebScanManager:
self._reset_scan_state() self._reset_scan_state()
# Check if we need another scan due to downloads during this scan # Check if we need another scan due to downloads during this scan
with self._lock: if downloads_during_scan:
if self._downloads_during_scan: logger.info("Web scan follow-up needed for downloads during scan")
logger.info("Web scan follow-up needed for downloads during scan") self.request_scan("Follow-up scan for downloads during previous scan")
self.request_scan("Follow-up scan for downloads during previous scan")
def _reset_scan_state(self): def _reset_scan_state(self):
"""Reset internal scan state""" """Reset internal scan state"""
@ -295,4 +326,24 @@ class WebScanManager:
self._current_server_type = None self._current_server_type = None
self._scan_start_time = None self._scan_start_time = None
self._scan_progress = {} self._scan_progress = {}
# Don't clear callbacks - they might be reused # Don't clear callbacks - they might be reused
def shutdown(self):
"""Cancel any pending timers and stop scheduling new work."""
with self._lock:
self._shutting_down = True
self._scan_in_progress = False
self._current_server_type = None
self._scan_start_time = None
self._scan_progress = {}
if self._timer:
self._timer.cancel()
self._timer = None
if self._completion_check_timer:
self._completion_check_timer.cancel()
self._completion_check_timer = None
self._downloads_during_scan = False
logger.info("WebScanManager shutdown - cancelled all pending timers")

17
core/worker_utils.py Normal file
View file

@ -0,0 +1,17 @@
"""Shared helpers for background workers."""
import threading
def interruptible_sleep(stop_event: threading.Event, seconds: float, step: float = 0.5) -> bool:
"""Sleep in chunks so shutdown can interrupt long waits."""
if seconds <= 0:
return stop_event.is_set()
remaining = float(seconds)
while remaining > 0 and not stop_event.is_set():
wait_for = min(step, remaining)
if stop_event.wait(wait_for):
break
remaining -= wait_for
return stop_event.is_set()

48
database/music_database.py
View file

@ -16,6 +16,7 @@ from utils.logging_config import get_logger
logger = get_logger("music_database") logger = get_logger("music_database")
_database_initialized_paths = set() _database_initialized_paths = set()
_database_sidecar_warnings = set()
_database_initialization_lock = threading.Lock() _database_initialization_lock = threading.Lock()
# Import matching engine for enhanced similarity logic # Import matching engine for enhanced similarity logic
@ -170,10 +171,57 @@ class MusicDatabase:
database_path = os.environ.get('DATABASE_PATH', 'database/music_library.db') database_path = os.environ.get('DATABASE_PATH', 'database/music_library.db')
self.database_path = Path(database_path) self.database_path = Path(database_path)
self.database_path.parent.mkdir(parents=True, exist_ok=True) self.database_path.parent.mkdir(parents=True, exist_ok=True)
self._warn_about_stale_sqlite_sidecars()
# Initialize database once per process for this path # Initialize database once per process for this path
self._initialize_database_once() self._initialize_database_once()
def _warn_about_stale_sqlite_sidecars(self):
"""Warn if SQLite sidecars are present and the database looks unhealthy."""
db_key = str(self.database_path.resolve())
with _database_initialization_lock:
if db_key in _database_sidecar_warnings:
return
_database_sidecar_warnings.add(db_key)
wal_path = Path(f"{self.database_path}-wal")
shm_path = Path(f"{self.database_path}-shm")
existing = [p.name for p in (wal_path, shm_path) if p.exists()]
if existing:
check_result = None
try:
conn = sqlite3.connect(f"file:{self.database_path}?mode=ro", uri=True, timeout=5.0)
try:
row = conn.execute("PRAGMA quick_check").fetchone()
check_result = row[0] if row else None
finally:
conn.close()
except Exception as e:
logger.warning(
"SQLite sidecar files detected for %s: %s, and database health check could not be run (%s). "
"This usually means the previous shutdown was not clean.",
self.database_path,
", ".join(existing),
e,
)
return
if check_result != "ok":
logger.warning(
"SQLite sidecar files detected for %s: %s, and quick_check returned %r. "
"This usually means the previous shutdown was not clean.",
self.database_path,
", ".join(existing),
check_result,
)
else:
logger.debug(
"SQLite sidecar files present for %s (%s) but quick_check returned ok.",
self.database_path,
", ".join(existing),
)
def _initialize_database_once(self): def _initialize_database_once(self):
"""Run schema setup and migrations once per database path per process.""" """Run schema setup and migrations once per database path per process."""
db_key = str(self.database_path.resolve()) db_key = str(self.database_path.resolve())

256
web_server.py
View file

@ -2519,27 +2519,41 @@ class WebUIDownloadMonitor:
self.monitoring = False self.monitoring = False
self.monitor_thread = None self.monitor_thread = None
self.monitored_batches = set() self.monitored_batches = set()
self._lock = threading.Lock()
def start_monitoring(self, batch_id): def start_monitoring(self, batch_id):
"""Start monitoring a download batch""" """Start monitoring a download batch"""
self.monitored_batches.add(batch_id) with self._lock:
if not self.monitoring: self.monitored_batches.add(batch_id)
self.monitoring = True if not self.monitoring:
self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True) self.monitoring = True
self.monitor_thread.start() self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
print(f"Started download monitor for batch {batch_id}") self.monitor_thread.start()
print(f"Started download monitor for batch {batch_id}")
def stop_monitoring(self, batch_id): def stop_monitoring(self, batch_id):
"""Stop monitoring a specific batch""" """Stop monitoring a specific batch"""
self.monitored_batches.discard(batch_id) with self._lock:
if not self.monitored_batches: self.monitored_batches.discard(batch_id)
if not self.monitored_batches:
self.monitoring = False
print(f"Stopped download monitor (no active batches)")
def shutdown(self):
"""Stop the monitor loop and clear active batch tracking."""
with self._lock:
self.monitoring = False self.monitoring = False
print(f"Stopped download monitor (no active batches)") self.monitored_batches.clear()
self.monitor_thread = None
print("Download monitor shutdown requested")
def _monitor_loop(self): def _monitor_loop(self):
"""Main monitoring loop - checks downloads every 1 second for responsive web UX""" """Main monitoring loop - checks downloads every 1 second for responsive web UX"""
while self.monitoring and self.monitored_batches: while self.monitoring and self.monitored_batches:
try: try:
if globals().get('IS_SHUTTING_DOWN', False):
self.monitoring = False
break
self._check_all_downloads() self._check_all_downloads()
time.sleep(1) # 1-second polling for fast web UI updates time.sleep(1) # 1-second polling for fast web UI updates
except Exception as e: except Exception as e:
@ -2626,6 +2640,8 @@ class WebUIDownloadMonitor:
completed_tasks.append((batch_id, task_id)) completed_tasks.append((batch_id, task_id))
# ---- All work below runs WITHOUT tasks_lock held ---- # ---- All work below runs WITHOUT tasks_lock held ----
if globals().get('IS_SHUTTING_DOWN', False) or not self.monitoring:
return
# Execute deferred operations from _should_retry_task (network calls, nested locks) # Execute deferred operations from _should_retry_task (network calls, nested locks)
for op in deferred_ops: for op in deferred_ops:
@ -3208,6 +3224,8 @@ def validate_and_heal_batch_states():
This is the server-side equivalent of the frontend's worker count validation. This is the server-side equivalent of the frontend's worker count validation.
""" """
try: try:
if globals().get('IS_SHUTTING_DOWN', False):
return
import time import time
current_time = time.time() current_time = time.time()
@ -3312,15 +3330,41 @@ def validate_and_heal_batch_states():
# Start periodic batch healing (every 30 seconds) # Start periodic batch healing (every 30 seconds)
import threading import threading
_batch_healing_timer = None
_batch_healing_timer_lock = threading.Lock()
def _schedule_batch_healing_timer(delay_seconds=30.0):
"""Schedule the next batch healing cycle."""
global _batch_healing_timer
if globals().get('IS_SHUTTING_DOWN', False):
return
timer = threading.Timer(delay_seconds, start_batch_healing_timer)
timer.daemon = True
with _batch_healing_timer_lock:
_batch_healing_timer = timer
timer.start()
def _cancel_batch_healing_timer():
"""Cancel the current batch healing timer if one exists."""
global _batch_healing_timer
with _batch_healing_timer_lock:
timer = _batch_healing_timer
_batch_healing_timer = None
if timer:
timer.cancel()
def start_batch_healing_timer(): def start_batch_healing_timer():
"""Start periodic batch state validation and healing""" """Start periodic batch state validation and healing"""
try: try:
if globals().get('IS_SHUTTING_DOWN', False):
return
validate_and_heal_batch_states() validate_and_heal_batch_states()
except Exception as e: except Exception as e:
print(f"[Batch Healing Timer] Error: {e}") print(f"[Batch Healing Timer] Error: {e}")
finally: finally:
# Schedule next healing cycle # Schedule next healing cycle
threading.Timer(30.0, start_batch_healing_timer).start() _schedule_batch_healing_timer(30.0)
# Start the healing timer when the server starts # Start the healing timer when the server starts
start_batch_healing_timer() start_batch_healing_timer()
@ -3334,35 +3378,85 @@ def cleanup_monitor():
"""Clean up background monitor on shutdown""" """Clean up background monitor on shutdown"""
if download_monitor.monitoring: if download_monitor.monitoring:
print("Flask shutdown detected, stopping download monitor...") print("Flask shutdown detected, stopping download monitor...")
download_monitor.monitoring = False download_monitor.shutdown()
download_monitor.monitored_batches.clear()
# Give the thread a moment to exit cleanly # Give the thread a moment to exit cleanly
time.sleep(0.5) time.sleep(0.5)
# Clean up batch locks to prevent memory leaks # Clean up batch locks to prevent memory leaks
with tasks_lock: try:
batch_locks.clear() acquired = tasks_lock.acquire(timeout=1.0)
print("Cleaned up batch locks") if acquired:
try:
batch_locks.clear()
print("Cleaned up batch locks")
finally:
tasks_lock.release()
else:
print("Skipped batch lock cleanup - tasks_lock busy")
except Exception as e:
print(f"Error cleaning up batch locks: {e}")
# Global shutdown flag # Global shutdown flag
IS_SHUTTING_DOWN = False IS_SHUTTING_DOWN = False
def signal_handler(signum, frame): def _shutdown_executor(executor, name):
"""Handle SIGINT (Ctrl+C) and SIGTERM""" """Shut down a ThreadPoolExecutor without waiting for long-running tasks."""
global IS_SHUTTING_DOWN if executor is None:
print(f"Signal {signum} received, cleaning up...") return
IS_SHUTTING_DOWN = True
cleanup_monitor()
# Stop automation engine
try: try:
if automation_engine: print(f"Shutting down {name}...")
print("Stopping automation engine...") executor.shutdown(wait=False, cancel_futures=True)
automation_engine.stop()
except Exception as e: except Exception as e:
print(f"Error stopping automation engine: {e}") print(f"Error shutting down {name}: {e}")
# Persist API call history def _stop_component(component, name, method_names=("stop", "shutdown")):
"""Call a best-effort stop method on a component if it has one."""
if component is None:
return
for method_name in method_names:
method = getattr(component, method_name, None)
if callable(method):
try:
print(f"Stopping {name}...")
method()
except Exception as e:
print(f"Error stopping {name}: {e}")
return
def _stop_components_parallel(components):
"""Stop multiple components concurrently and wait for all stop calls to finish."""
stop_threads = []
for component, name in components:
if component is None:
continue
thread = threading.Thread(
target=_stop_component,
args=(component, name),
name=f"shutdown-{name.replace(' ', '-')}",
)
thread.start()
stop_threads.append((name, thread))
for name, thread in stop_threads:
thread.join()
def _shutdown_runtime_components():
"""Best-effort shutdown for timers, monitors, workers, and executors."""
global IS_SHUTTING_DOWN
if IS_SHUTTING_DOWN:
return
IS_SHUTTING_DOWN = True
_cancel_batch_healing_timer()
cleanup_monitor()
_stop_component(web_scan_manager, "web scan manager")
_stop_component(automation_engine, "automation engine")
# Persist API call history before shutting down worker pools.
try: try:
from core.api_call_tracker import api_call_tracker from core.api_call_tracker import api_call_tracker
api_call_tracker.save() api_call_tracker.save()
@ -3370,13 +3464,57 @@ def signal_handler(signum, frame):
except Exception as e: except Exception as e:
print(f"Error saving API call history: {e}") print(f"Error saving API call history: {e}")
# Shutdown executor to prevent new tasks # Stop the active DB update worker before tearing down the executor it runs on.
try: # This lets an in-flight update observe should_stop and exit cleanly.
print("Shutting down missing_download_executor...") _stop_component(db_update_worker, "db update worker")
missing_download_executor.shutdown(wait=False, cancel_futures=True) _stop_component(metadata_update_runtime_worker, "metadata update worker")
except Exception as e:
print(f"Error shutting down executor: {e}")
# Stop long-lived worker components in parallel so shutdown waits for the
# slowest worker instead of serially burning the timeout for each one.
_stop_components_parallel([
(mb_worker, "musicbrainz worker"),
(audiodb_worker, "audiodb worker"),
(discogs_worker, "discogs worker"),
(deezer_worker, "deezer worker"),
(spotify_enrichment_worker, "spotify enrichment worker"),
(itunes_enrichment_worker, "itunes enrichment worker"),
(lastfm_worker, "lastfm worker"),
(genius_worker, "genius worker"),
(tidal_enrichment_worker, "tidal enrichment worker"),
(qobuz_enrichment_worker, "qobuz enrichment worker"),
(hydrabase_worker, "hydrabase worker"),
(soulid_worker, "soulid worker"),
(listening_stats_worker, "listening stats worker"),
(repair_worker, "repair worker"),
])
# Shut down executor pools so their worker threads stop keeping the process alive.
for executor, name in [
(stream_executor, "stream executor"),
(db_update_executor, "db update executor"),
(quality_scanner_executor, "quality scanner executor"),
(duplicate_cleaner_executor, "duplicate cleaner executor"),
(retag_executor, "retag executor"),
(sync_executor, "sync executor"),
(missing_download_executor, "missing download executor"),
(tidal_discovery_executor, "tidal discovery executor"),
(deezer_discovery_executor, "deezer discovery executor"),
(spotify_public_discovery_executor, "spotify public discovery executor"),
(youtube_discovery_executor, "youtube discovery executor"),
(beatport_discovery_executor, "beatport discovery executor"),
(listenbrainz_discovery_executor, "listenbrainz discovery executor"),
(similar_artists_executor, "similar artists executor"),
(metadata_update_executor, "metadata update executor"),
]:
_shutdown_executor(executor, name)
# Give daemon cleanup threads a moment to observe the shutdown flag.
time.sleep(0.2)
def signal_handler(signum, frame):
"""Handle SIGINT (Ctrl+C) and SIGTERM"""
print(f"Signal {signum} received, cleaning up...")
_shutdown_runtime_components()
sys.exit(0) sys.exit(0)
# Register cleanup handlers # Register cleanup handlers
@ -3387,7 +3525,14 @@ def _atexit_save_history():
except Exception: except Exception:
pass pass
def _atexit_shutdown():
try:
_shutdown_runtime_components()
except Exception:
pass
atexit.register(_atexit_save_history) atexit.register(_atexit_save_history)
atexit.register(_atexit_shutdown)
atexit.register(cleanup_monitor) atexit.register(cleanup_monitor)
signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler) signal.signal(signal.SIGTERM, signal_handler)
@ -22742,7 +22887,7 @@ def _simple_monitor_task():
Search cleanup and download cleanup are now handled by system automations.""" Search cleanup and download cleanup are now handled by system automations."""
print("Simple background monitor started") print("Simple background monitor started")
while True: while not globals().get('IS_SHUTTING_DOWN', False):
try: try:
with matched_context_lock: with matched_context_lock:
pending_count = len(matched_downloads_context) pending_count = len(matched_downloads_context)
@ -22769,6 +22914,8 @@ def _simple_monitor_task():
print(f"Simple monitor error: {e}") print(f"Simple monitor error: {e}")
time.sleep(10) time.sleep(10)
print("Simple background monitor stopped")
def start_simple_background_monitor(): def start_simple_background_monitor():
"""Starts the simple background monitor thread.""" """Starts the simple background monitor thread."""
monitor_thread = threading.Thread(target=_simple_monitor_task) monitor_thread = threading.Thread(target=_simple_monitor_task)
@ -40708,6 +40855,7 @@ metadata_update_state = {
} }
metadata_update_worker = None metadata_update_worker = None
metadata_update_runtime_worker = None
metadata_update_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="metadata_update") metadata_update_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="metadata_update")
# =============================== # ===============================
@ -44822,7 +44970,7 @@ def _old_get_listenbrainz_playlist_tracks_DEPRECATED(playlist_mbid):
@app.route('/api/metadata/start', methods=['POST']) @app.route('/api/metadata/start', methods=['POST'])
def start_metadata_update(): def start_metadata_update():
"""Start the metadata update process - EXACT copy of dashboard.py logic""" """Start the metadata update process - EXACT copy of dashboard.py logic"""
global metadata_update_worker, metadata_update_state global metadata_update_worker, metadata_update_runtime_worker, metadata_update_state
try: try:
# Check if already running # Check if already running
@ -44890,6 +45038,7 @@ def start_metadata_update():
# Start the metadata update worker - EXACTLY like dashboard.py # Start the metadata update worker - EXACTLY like dashboard.py
def run_metadata_update(): def run_metadata_update():
global metadata_update_runtime_worker
try: try:
metadata_worker = WebMetadataUpdateWorker( metadata_worker = WebMetadataUpdateWorker(
None, # Artists will be loaded in the worker thread - EXACTLY like dashboard.py None, # Artists will be loaded in the worker thread - EXACTLY like dashboard.py
@ -44898,12 +45047,15 @@ def start_metadata_update():
active_server, active_server,
refresh_interval_days refresh_interval_days
) )
metadata_update_runtime_worker = metadata_worker
metadata_worker.run() metadata_worker.run()
except Exception as e: except Exception as e:
print(f"Error in metadata update worker: {e}") print(f"Error in metadata update worker: {e}")
metadata_update_state['status'] = 'error' metadata_update_state['status'] = 'error'
metadata_update_state['error'] = str(e) metadata_update_state['error'] = str(e)
add_activity_item("", "Metadata Error", str(e), "Now") add_activity_item("", "Metadata Error", str(e), "Now")
finally:
metadata_update_runtime_worker = None
metadata_update_worker = metadata_update_executor.submit(run_metadata_update) metadata_update_worker = metadata_update_executor.submit(run_metadata_update)
@ -52037,7 +52189,7 @@ def _hydrabase_reconnect_loop():
global _hydrabase_ws global _hydrabase_ws
_consecutive_failures = 0 _consecutive_failures = 0
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(30) socketio.sleep(30)
try: try:
# Only attempt reconnect if auto_connect is enabled # Only attempt reconnect if auto_connect is enabled
@ -52087,7 +52239,7 @@ def _hydrabase_reconnect_loop():
def _emit_service_status_loop(): def _emit_service_status_loop():
"""Background thread that pushes service status every 5 seconds.""" """Background thread that pushes service status every 5 seconds."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(5) socketio.sleep(5)
try: try:
socketio.emit('status:update', _build_status_payload()) socketio.emit('status:update', _build_status_payload())
@ -52096,7 +52248,7 @@ def _emit_service_status_loop():
def _emit_watchlist_count_loop(): def _emit_watchlist_count_loop():
"""Background thread that pushes watchlist count every 10 seconds to each profile room.""" """Background thread that pushes watchlist count every 10 seconds to each profile room."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(10) socketio.sleep(10)
try: try:
database = get_database() database = get_database()
@ -52109,7 +52261,7 @@ def _emit_watchlist_count_loop():
def _emit_download_status_loop(): def _emit_download_status_loop():
"""Background thread that pushes download batch status every 2 seconds to subscribed rooms.""" """Background thread that pushes download batch status every 2 seconds to subscribed rooms."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(2) socketio.sleep(2)
try: try:
live_transfers_lookup = get_cached_transfer_data() live_transfers_lookup = get_cached_transfer_data()
@ -52170,7 +52322,7 @@ def handle_profile_join(data):
def _emit_system_stats_loop(): def _emit_system_stats_loop():
"""Background thread that pushes system stats every 10 seconds.""" """Background thread that pushes system stats every 10 seconds."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(10) socketio.sleep(10)
try: try:
socketio.emit('dashboard:stats', _build_system_stats()) socketio.emit('dashboard:stats', _build_system_stats())
@ -52179,7 +52331,7 @@ def _emit_system_stats_loop():
def _emit_activity_feed_loop(): def _emit_activity_feed_loop():
"""Background thread that pushes activity feed every 2 seconds.""" """Background thread that pushes activity feed every 2 seconds."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(2) socketio.sleep(2)
try: try:
with activity_feed_lock: with activity_feed_lock:
@ -52190,7 +52342,7 @@ def _emit_activity_feed_loop():
def _emit_db_stats_loop(): def _emit_db_stats_loop():
"""Background thread that pushes database stats every 10 seconds.""" """Background thread that pushes database stats every 10 seconds."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(10) socketio.sleep(10)
try: try:
db = get_database() db = get_database()
@ -52201,7 +52353,7 @@ def _emit_db_stats_loop():
def _emit_wishlist_count_loop(): def _emit_wishlist_count_loop():
"""Background thread that pushes wishlist count every 10 seconds to each profile room.""" """Background thread that pushes wishlist count every 10 seconds to each profile room."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(10) socketio.sleep(10)
try: try:
from core.wishlist_service import get_wishlist_service from core.wishlist_service import get_wishlist_service
@ -52247,7 +52399,7 @@ def _emit_rate_monitor_loop():
'tidal': 'tidal_enrichment', 'qobuz': 'qobuz_enrichment', 'tidal': 'tidal_enrichment', 'qobuz': 'qobuz_enrichment',
} }
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(1) socketio.sleep(1)
try: try:
from core.api_call_tracker import api_call_tracker from core.api_call_tracker import api_call_tracker
@ -52317,7 +52469,7 @@ def _emit_enrichment_status_loop():
'genius-enrichment': lambda: genius_worker, 'genius-enrichment': lambda: genius_worker,
} }
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(2) socketio.sleep(2)
# Auto-pause/resume rate-limited workers during downloads # Auto-pause/resume rate-limited workers during downloads
@ -52355,7 +52507,7 @@ def _emit_enrichment_status_loop():
def _emit_tool_progress_loop(): def _emit_tool_progress_loop():
"""Background thread that pushes all tool progress statuses every 1 second.""" """Background thread that pushes all tool progress statuses every 1 second."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(1) socketio.sleep(1)
# Stream status # Stream status
try: try:
@ -52443,7 +52595,7 @@ def handle_discovery_unsubscribe(data):
def _emit_sync_progress_loop(): def _emit_sync_progress_loop():
"""Push sync progress to subscribed rooms every 1 second.""" """Push sync progress to subscribed rooms every 1 second."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(1) socketio.sleep(1)
try: try:
with sync_lock: with sync_lock:
@ -52467,7 +52619,7 @@ def _emit_discovery_progress_loop():
'listenbrainz': lambda: listenbrainz_playlist_states, 'listenbrainz': lambda: listenbrainz_playlist_states,
'spotify_public': lambda: spotify_public_discovery_states, 'spotify_public': lambda: spotify_public_discovery_states,
} }
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(1) socketio.sleep(1)
for platform, get_states in platform_states.items(): for platform, get_states in platform_states.items():
try: try:
@ -52497,7 +52649,7 @@ def _emit_discovery_progress_loop():
def _emit_scan_status_loop(): def _emit_scan_status_loop():
"""Push watchlist and media scan status every 2 seconds.""" """Push watchlist and media scan status every 2 seconds."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(2) socketio.sleep(2)
# Watchlist scan # Watchlist scan
try: try:
@ -52529,7 +52681,7 @@ def _emit_scan_status_loop():
def _emit_automation_progress_loop(): def _emit_automation_progress_loop():
"""Push automation:progress events every 1 second for running automations.""" """Push automation:progress events every 1 second for running automations."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(1) socketio.sleep(1)
try: try:
with automation_progress_lock: with automation_progress_lock:
@ -52573,7 +52725,7 @@ def _emit_automation_progress_loop():
def _emit_repair_progress_loop(): def _emit_repair_progress_loop():
"""Push repair:progress events every 1 second for running repair jobs.""" """Push repair:progress events every 1 second for running repair jobs."""
while True: while not globals().get('IS_SHUTTING_DOWN', False):
socketio.sleep(1) socketio.sleep(1)
try: try:
if repair_worker is None: if repair_worker is None: