From aec3047216753a598d40c6849598c54655405f2f Mon Sep 17 00:00:00 2001 From: Antti Kettunen Date: Sun, 12 Apr 2026 15:15:53 +0300 Subject: [PATCH] Improve graceful shutdown and rollback safety - Add interruptible stop events to background workers so shutdown wakes out of long sleeps instead of waiting on fixed delays. - Stop scan managers, repair worker, executors, and cleanup helpers deterministically so process exit does not leave background threads alive. - Add startup warnings for stale SQLite WAL/SHM sidecars so unclean shutdowns are easier to spot before init/migration errors cascade. - Prevent forced kills from leaving SQLite sidecars behind, which made rollbacks to older branches fail with malformed database errors. --- core/audiodb_worker.py | 14 +- core/deezer_worker.py | 14 +- core/discogs_worker.py | 14 +- core/genius_worker.py | 16 +- core/hydrabase_worker.py | 12 +- core/itunes_worker.py | 20 +- core/lastfm_worker.py | 16 +- core/listening_stats_worker.py | 19 +- core/media_scan_manager.py | 15 +- core/musicbrainz_worker.py | 14 +- core/plex_scan_manager.py | 19 +- core/qobuz_worker.py | 22 ++- core/repair_jobs/acoustid_scanner.py | 3 +- core/repair_jobs/base.py | 28 ++- core/repair_jobs/library_reorganize.py | 3 +- core/repair_jobs/mbid_mismatch_detector.py | 3 +- core/repair_jobs/metadata_gap_filler.py | 3 +- core/repair_jobs/unknown_artist_fixer.py | 3 +- core/repair_worker.py | 34 +++- core/soulid_worker.py | 20 +- core/spotify_worker.py | 28 +-- core/tidal_worker.py | 20 +- core/web_scan_manager.py | 51 ++++- core/worker_utils.py | 17 ++ database/music_database.py | 48 +++++ web_server.py | 206 +++++++++++++++++---- 26 files changed, 520 insertions(+), 142 deletions(-) create mode 100644 core/worker_utils.py diff --git a/core/audiodb_worker.py b/core/audiodb_worker.py index bd204c3f..16d665ed 100644 --- a/core/audiodb_worker.py +++ b/core/audiodb_worker.py @@ -8,6 +8,7 @@ from datetime import datetime, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.audiodb_client import AudioDBClient +from core.worker_utils import interruptible_sleep logger = get_logger("audiodb_worker") @@ -24,6 +25,7 @@ class AudioDBWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -52,6 +54,7 @@ class AudioDBWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("AudioDB background worker started") @@ -64,9 +67,10 @@ class AudioDBWorker: logger.info("Stopping AudioDB worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("AudioDB worker stopped") @@ -115,7 +119,7 @@ class AudioDBWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue self.current_item = None @@ -124,7 +128,7 @@ class AudioDBWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -143,11 +147,11 @@ class AudioDBWorker: self._process_item(item) - time.sleep(2) + interruptible_sleep(self._stop_event, 2) except Exception as e: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) logger.info("AudioDB worker thread finished") diff --git a/core/deezer_worker.py b/core/deezer_worker.py index 9e216f06..b70ea57a 100644 --- a/core/deezer_worker.py +++ b/core/deezer_worker.py @@ -8,6 +8,7 @@ from datetime import datetime, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.deezer_client import DeezerClient +from core.worker_utils import interruptible_sleep logger = get_logger("deezer_worker") @@ -24,6 +25,7 @@ class DeezerWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -52,6 +54,7 @@ class DeezerWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Deezer background worker started") @@ -64,9 +67,10 @@ class DeezerWorker: logger.info("Stopping Deezer worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Deezer worker stopped") @@ -115,7 +119,7 @@ class DeezerWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue self.current_item = None @@ -124,7 +128,7 @@ class DeezerWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -143,11 +147,11 @@ class DeezerWorker: self._process_item(item) - time.sleep(2) + interruptible_sleep(self._stop_event, 2) except Exception as e: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) logger.info("Deezer worker thread finished") diff --git a/core/discogs_worker.py b/core/discogs_worker.py index 6bcdda31..98112c94 100644 --- a/core/discogs_worker.py +++ b/core/discogs_worker.py @@ -18,6 +18,7 @@ from datetime import datetime, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.discogs_client import DiscogsClient +from core.worker_utils import interruptible_sleep logger = get_logger("discogs_worker") @@ -34,6 +35,7 @@ class DiscogsWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -62,6 +64,7 @@ class DiscogsWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Discogs background worker started") @@ -73,8 +76,9 @@ class DiscogsWorker: logger.info("Stopping Discogs worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Discogs worker stopped") def pause(self): @@ -113,14 +117,14 @@ class DiscogsWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue self.current_item = None item = self._get_next_item() if not item: - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item.get('name', '') @@ -132,11 +136,11 @@ class DiscogsWorker: continue self._process_item(item) - time.sleep(2) + interruptible_sleep(self._stop_event, 2) except Exception as 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") diff --git a/core/genius_worker.py b/core/genius_worker.py index f3468b71..63de47a6 100644 --- a/core/genius_worker.py +++ b/core/genius_worker.py @@ -9,6 +9,7 @@ from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.genius_client import GeniusClient from config.settings import config_manager +from core.worker_utils import interruptible_sleep logger = get_logger("genius_worker") @@ -31,6 +32,7 @@ class GeniusWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -64,6 +66,7 @@ class GeniusWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Genius background worker started") @@ -76,9 +79,10 @@ class GeniusWorker: logger.info("Stopping Genius worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Genius worker stopped") @@ -123,14 +127,14 @@ class GeniusWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # Check if access token is configured if not self.client.access_token: self._init_client() if not self.client.access_token: - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue self.current_item = None @@ -138,7 +142,7 @@ class GeniusWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -157,11 +161,11 @@ class GeniusWorker: self._process_item(item) # Genius rate limiting is conservative (500ms per call) + lyrics scraping - time.sleep(1) + interruptible_sleep(self._stop_event, 1) except Exception as e: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) logger.info("Genius worker thread finished") diff --git a/core/hydrabase_worker.py b/core/hydrabase_worker.py index f6e44c6b..99b30a1b 100644 --- a/core/hydrabase_worker.py +++ b/core/hydrabase_worker.py @@ -15,6 +15,7 @@ import time import uuid logger = logging.getLogger(__name__) +from core.worker_utils import interruptible_sleep class HydrabaseWorker: @@ -31,6 +32,7 @@ class HydrabaseWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Queue with cap self.queue = queue.Queue(maxsize=1000) @@ -47,6 +49,7 @@ class HydrabaseWorker: return self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Hydrabase P2P mirror worker started") @@ -56,8 +59,9 @@ class HydrabaseWorker: return self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Hydrabase P2P mirror worker stopped") def pause(self): @@ -102,7 +106,7 @@ class HydrabaseWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # Non-blocking dequeue with timeout @@ -112,12 +116,12 @@ class HydrabaseWorker: continue self._process_item(item) - time.sleep(0.5) # Rate limit + interruptible_sleep(self._stop_event, 0.5) # Rate limit except Exception as e: logger.error(f"Error in Hydrabase worker loop: {e}") self.stats['errors'] += 1 - time.sleep(2) + interruptible_sleep(self._stop_event, 2) def _process_item(self, item): ws, lock = self.get_ws_and_lock() diff --git a/core/itunes_worker.py b/core/itunes_worker.py index 723ce5b6..9a8bd085 100644 --- a/core/itunes_worker.py +++ b/core/itunes_worker.py @@ -8,6 +8,7 @@ from datetime import datetime, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.itunes_client import iTunesClient +from core.worker_utils import interruptible_sleep logger = get_logger("itunes_worker") @@ -34,6 +35,7 @@ class iTunesWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -66,6 +68,7 @@ class iTunesWorker: return self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("iTunes background worker started") @@ -76,8 +79,9 @@ class iTunesWorker: logger.info("Stopping iTunes worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("iTunes worker stopped") def pause(self): @@ -116,7 +120,7 @@ class iTunesWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # No auth check needed — iTunes API requires no authentication @@ -126,7 +130,7 @@ class iTunesWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -147,13 +151,13 @@ class iTunesWorker: # Sleep depends on item type — search items need more delay item_type = item.get('type', '') 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: - time.sleep(self.inter_item_sleep) + interruptible_sleep(self._stop_event, self.inter_item_sleep) except Exception as e: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) self.current_item = None logger.info("iTunes worker thread finished") @@ -444,7 +448,7 @@ class iTunesWorker: self._mark_status('album', db_id, 'not_found') 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") @@ -516,7 +520,7 @@ class iTunesWorker: self._mark_status('track', db_id, 'not_found') 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") diff --git a/core/lastfm_worker.py b/core/lastfm_worker.py index 4005e593..6c0f4de3 100644 --- a/core/lastfm_worker.py +++ b/core/lastfm_worker.py @@ -9,6 +9,7 @@ from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.lastfm_client import LastFMClient from config.settings import config_manager +from core.worker_utils import interruptible_sleep logger = get_logger("lastfm_worker") @@ -31,6 +32,7 @@ class LastFMWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -65,6 +67,7 @@ class LastFMWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Last.fm background worker started") @@ -77,9 +80,10 @@ class LastFMWorker: logger.info("Stopping Last.fm worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Last.fm worker stopped") @@ -124,14 +128,14 @@ class LastFMWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # Check if API key is configured if not self.client.api_key: self._init_client() if not self.client.api_key: - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue self.current_item = None @@ -139,7 +143,7 @@ class LastFMWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -158,11 +162,11 @@ class LastFMWorker: self._process_item(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: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) logger.info("Last.fm worker thread finished") diff --git a/core/listening_stats_worker.py b/core/listening_stats_worker.py index 5710db18..42648a51 100644 --- a/core/listening_stats_worker.py +++ b/core/listening_stats_worker.py @@ -11,6 +11,7 @@ import time from typing import Dict, Any from utils.logging_config import get_logger +from core.worker_utils import interruptible_sleep logger = get_logger("listening_stats_worker") @@ -32,6 +33,7 @@ class ListeningStatsWorker: self.should_stop = False self.thread = None self.current_item = None + self._stop_event = threading.Event() # Stats self.stats = { @@ -52,6 +54,7 @@ class ListeningStatsWorker: return self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Listening stats worker started") @@ -61,8 +64,9 @@ class ListeningStatsWorker: return self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Listening stats worker stopped") def pause(self): @@ -86,7 +90,7 @@ class ListeningStatsWorker: logger.info("Listening stats worker thread started") # Build cache from existing data immediately (before first poll) - time.sleep(5) + interruptible_sleep(self._stop_event, 5) try: self._build_stats_cache() logger.info("Initial stats cache built from existing data") @@ -94,17 +98,17 @@ class ListeningStatsWorker: logger.debug(f"Initial cache build skipped: {e}") # Wait before first poll - time.sleep(10) + interruptible_sleep(self._stop_event, 10) while not self.should_stop: try: if self.paused: - time.sleep(5) + interruptible_sleep(self._stop_event, 5) continue # Check if enabled if not self.config_manager.get('listening_stats.enabled', True): - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue # Update poll interval from config @@ -119,12 +123,13 @@ class ListeningStatsWorker: for _ in range(int(self.poll_interval)): if self.should_stop: break - time.sleep(1) + if interruptible_sleep(self._stop_event, 1): + break except Exception as e: logger.error(f"Error in listening stats worker: {e}", exc_info=True) self.stats['errors'] += 1 - time.sleep(60) + interruptible_sleep(self._stop_event, 60) self.current_item = None logger.info("Listening stats worker thread finished") diff --git a/core/media_scan_manager.py b/core/media_scan_manager.py index 28f083b3..451d5fed 100644 --- a/core/media_scan_manager.py +++ b/core/media_scan_manager.py @@ -39,6 +39,7 @@ class MediaScanManager: self._periodic_update_timer = None # Timer for 5-minute periodic updates self._periodic_update_interval = 300 # 5 minutes in seconds 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") @@ -119,6 +120,9 @@ class MediaScanManager: """ logger.info(f"DEBUG: Media scan requested - reason: {reason}") with self._lock: + if self._shutting_down: + logger.debug("Media scan request ignored during shutdown") + return if self._scan_in_progress: # Server is currently scanning - mark that we need another scan later self._downloads_during_scan = True @@ -134,6 +138,7 @@ class MediaScanManager: # Start the debounce timer self._timer = threading.Timer(self.delay, self._execute_scan) + self._timer.daemon = True self._timer.start() def add_scan_completion_callback(self, callback): @@ -164,6 +169,9 @@ class MediaScanManager: def _execute_scan(self): """Execute the actual media library scan""" with self._lock: + if self._shutting_down: + logger.debug("Media scan execution skipped during shutdown") + return if self._scan_in_progress: logger.warning("Scan already in progress - skipping duplicate execution") return @@ -211,6 +219,7 @@ class MediaScanManager: # 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.daemon = True self._periodic_update_timer.start() except Exception as e: @@ -250,8 +259,11 @@ class MediaScanManager: self._call_completion_callbacks() # Schedule next periodic update + if self._shutting_down: + return 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.daemon = True self._periodic_update_timer.start() else: # Scanning stopped - final update and cleanup @@ -360,6 +372,7 @@ class MediaScanManager: def shutdown(self): """Clean shutdown - cancel any pending timers""" with self._lock: + self._shutting_down = True if self._timer: self._timer.cancel() self._timer = None @@ -369,4 +382,4 @@ class MediaScanManager: self._periodic_update_timer = None self._is_doing_periodic_updates = False - logger.info("MediaScanManager shutdown - cancelled all pending timers") \ No newline at end of file + logger.info("MediaScanManager shutdown - cancelled all pending timers") diff --git a/core/musicbrainz_worker.py b/core/musicbrainz_worker.py index ccdd7108..79496d2b 100644 --- a/core/musicbrainz_worker.py +++ b/core/musicbrainz_worker.py @@ -5,6 +5,7 @@ from datetime import datetime, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.musicbrainz_service import MusicBrainzService +from core.worker_utils import interruptible_sleep logger = get_logger("musicbrainz_worker") @@ -20,6 +21,7 @@ class MusicBrainzWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -45,6 +47,7 @@ class MusicBrainzWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("MusicBrainz background worker started") @@ -57,9 +60,10 @@ class MusicBrainzWorker: logger.info("Stopping MusicBrainz worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Music Brainz worker stopped") @@ -112,7 +116,7 @@ class MusicBrainzWorker: try: # Check if paused if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # Clear previous item before getting next @@ -124,7 +128,7 @@ class MusicBrainzWorker: if not item: # No more items - sleep for a bit logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue # 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 # Rate limit: 1 request per second - time.sleep(1) + interruptible_sleep(self._stop_event, 1) except Exception as 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") diff --git a/core/plex_scan_manager.py b/core/plex_scan_manager.py index 8c694284..76df7dd5 100644 --- a/core/plex_scan_manager.py +++ b/core/plex_scan_manager.py @@ -39,6 +39,7 @@ class PlexScanManager: self._periodic_update_timer = None # Timer for 5-minute periodic updates self._periodic_update_interval = 300 # 5 minutes in seconds 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") @@ -51,6 +52,9 @@ class PlexScanManager: """ logger.info(f"DEBUG: Plex scan requested - reason: {reason}") with self._lock: + if self._shutting_down: + logger.debug("Plex scan request ignored during shutdown") + return if self._scan_in_progress: # Plex is currently scanning - mark that we need another scan later self._downloads_during_scan = True @@ -66,6 +70,7 @@ class PlexScanManager: # Start the debounce timer self._timer = threading.Timer(self.delay, self._execute_scan) + self._timer.daemon = True self._timer.start() def add_scan_completion_callback(self, callback): @@ -96,6 +101,9 @@ class PlexScanManager: def _execute_scan(self): """Execute the actual Plex library scan""" with self._lock: + if self._shutting_down: + logger.debug("Plex scan execution skipped during shutdown") + return if self._scan_in_progress: logger.warning("Scan already in progress - skipping duplicate execution") return @@ -136,6 +144,7 @@ class PlexScanManager: # 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.daemon = True self._periodic_update_timer.start() except Exception as e: @@ -168,8 +177,11 @@ class PlexScanManager: self._call_completion_callbacks() # Schedule next periodic update + if self._shutting_down: + return 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.daemon = True self._periodic_update_timer.start() else: # Scanning stopped - final update and cleanup @@ -218,7 +230,9 @@ class PlexScanManager: if is_scanning: # Still scanning, poll 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: # Scan completed! elapsed_time = time.time() - self._scan_start_time if self._scan_start_time else 0 @@ -311,6 +325,7 @@ class PlexScanManager: def shutdown(self): """Clean shutdown - cancel any pending timers""" with self._lock: + self._shutting_down = True if self._timer: self._timer.cancel() self._timer = None @@ -320,4 +335,4 @@ class PlexScanManager: self._periodic_update_timer = None self._is_doing_periodic_updates = False - logger.info("PlexScanManager shutdown - cancelled all pending timers") \ No newline at end of file + logger.info("PlexScanManager shutdown - cancelled all pending timers") diff --git a/core/qobuz_worker.py b/core/qobuz_worker.py index bf54c4e8..d47b58b4 100644 --- a/core/qobuz_worker.py +++ b/core/qobuz_worker.py @@ -8,6 +8,7 @@ from datetime import datetime, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.qobuz_client import _qobuz_is_rate_limited +from core.worker_utils import interruptible_sleep logger = get_logger("qobuz_worker") @@ -24,6 +25,7 @@ class QobuzWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -52,6 +54,7 @@ class QobuzWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Qobuz background worker started") @@ -64,9 +67,10 @@ class QobuzWorker: logger.info("Stopping Qobuz worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Qobuz worker stopped") @@ -120,24 +124,24 @@ class QobuzWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # Auth guard: sleep if not authenticated try: if not self.client or not self.client.is_authenticated(): self.current_item = None - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue except Exception: - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue # Rate limit guard: back off if globally rate limited if _qobuz_is_rate_limited(): self.current_item = None logger.debug("Qobuz rate limited, backing off...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = None @@ -146,7 +150,7 @@ class QobuzWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -166,11 +170,11 @@ class QobuzWorker: self._process_item(item) # Throttle between API calls - time.sleep(2) + interruptible_sleep(self._stop_event, 2) except Exception as e: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) logger.info("Qobuz worker thread finished") @@ -351,7 +355,7 @@ class QobuzWorker: error_str = str(e).lower() 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") - time.sleep(30) + interruptible_sleep(self._stop_event, 30) return logger.error(f"Error processing {item['type']} #{item['id']}: {e}") self.stats['errors'] += 1 diff --git a/core/repair_jobs/acoustid_scanner.py b/core/repair_jobs/acoustid_scanner.py index dad6ae60..560cdd64 100644 --- a/core/repair_jobs/acoustid_scanner.py +++ b/core/repair_jobs/acoustid_scanner.py @@ -141,7 +141,8 @@ class AcoustIDScannerJob(RepairJob): if batch_count >= batch_size: batch_count = 0 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: context.update_progress(i + 1, total) diff --git a/core/repair_jobs/base.py b/core/repair_jobs/base.py index 41178780..f07b6019 100644 --- a/core/repair_jobs/base.py +++ b/core/repair_jobs/base.py @@ -2,6 +2,7 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field +import threading from typing import Any, Callable, Dict, List, Optional @@ -29,6 +30,7 @@ class JobContext: mb_client: Any = None acoustid_client: Any = None metadata_cache: Any = None + stop_event: Optional[threading.Event] = None # Callbacks create_finding: Optional[Callable] = None @@ -39,6 +41,8 @@ class JobContext: def check_stop(self) -> bool: """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 def is_spotify_rate_limited(self) -> bool: @@ -55,11 +59,31 @@ class JobContext: def wait_if_paused(self): """Block until unpaused or stopped. Returns True if should stop.""" - import time while self.is_paused and self.is_paused(): if self.check_stop(): 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() diff --git a/core/repair_jobs/library_reorganize.py b/core/repair_jobs/library_reorganize.py index 8c6debe0..68afd5bd 100644 --- a/core/repair_jobs/library_reorganize.py +++ b/core/repair_jobs/library_reorganize.py @@ -821,7 +821,8 @@ class LibraryReorganizeJob(RepairJob): years[key] = year_str break import time - time.sleep(0.1) # Rate limit courtesy + if context.sleep_or_stop(0.1): # Rate limit courtesy + break except Exception as e: logger.debug("API year lookup failed for %s - %s: %s", artist, album, e) diff --git a/core/repair_jobs/mbid_mismatch_detector.py b/core/repair_jobs/mbid_mismatch_detector.py index 465de1a7..51d89939 100644 --- a/core/repair_jobs/mbid_mismatch_detector.py +++ b/core/repair_jobs/mbid_mismatch_detector.py @@ -294,7 +294,8 @@ class MbidMismatchDetectorJob(RepairJob): try: # 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']) if not recording: diff --git a/core/repair_jobs/metadata_gap_filler.py b/core/repair_jobs/metadata_gap_filler.py index b2186ded..323f7522 100644 --- a/core/repair_jobs/metadata_gap_filler.py +++ b/core/repair_jobs/metadata_gap_filler.py @@ -173,7 +173,8 @@ class MetadataGapFillerJob(RepairJob): # Rate limit API calls 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: context.update_progress(i + 1, total) diff --git a/core/repair_jobs/unknown_artist_fixer.py b/core/repair_jobs/unknown_artist_fixer.py index 353039fa..56de6284 100644 --- a/core/repair_jobs/unknown_artist_fixer.py +++ b/core/repair_jobs/unknown_artist_fixer.py @@ -337,7 +337,8 @@ class UnknownArtistFixerJob(RepairJob): except Exception as e: logger.debug(f"Title search failed for '{title}': {e}") # Rate limit courtesy - time.sleep(0.2) + if context.sleep_or_stop(0.2): + return None return None diff --git a/core/repair_worker.py b/core/repair_worker.py index ff3bd531..305c1fb9 100644 --- a/core/repair_worker.py +++ b/core/repair_worker.py @@ -67,6 +67,7 @@ class RepairWorker: self.running = False self.enabled = False # Master toggle (replaces 'paused') self.should_stop = False + self._stop_event = threading.Event() self.thread = None # Current job being executed @@ -293,6 +294,7 @@ class RepairWorker: return self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Repair worker started") @@ -303,8 +305,9 @@ class RepairWorker: logger.info("Stopping repair worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=2) logger.info("Repair worker stopped") def toggle(self) -> bool: @@ -418,7 +421,7 @@ class RepairWorker: logger.info("Repair worker thread started") self._ensure_jobs_loaded() - while not self.should_stop: + while not self._stop_event.is_set(): try: # Check force-run queue even when disabled (user explicitly requested) forced_job = None @@ -428,13 +431,15 @@ class RepairWorker: if forced_job: self._run_job(forced_job) - time.sleep(2) + if self._sleep_or_stop(2): + break continue if not self.enabled: self._current_job_id = None self._current_job_name = None - time.sleep(2) + if self._sleep_or_stop(2): + break continue # Find the next job to run based on staleness @@ -444,20 +449,23 @@ class RepairWorker: # Nothing due — sleep and re-check self._current_job_id = None self._current_job_name = None - time.sleep(10) + if self._sleep_or_stop(10): + break continue # Run the selected job self._run_job(next_job) # Brief pause between jobs - time.sleep(5) + if self._sleep_or_stop(5): + break except Exception as e: logger.error("Error in repair worker loop: %s", e, exc_info=True) self._current_job_id = None self._current_job_name = None - time.sleep(30) + if self._sleep_or_stop(30): + break logger.info("Repair worker thread finished") @@ -552,6 +560,7 @@ class RepairWorker: metadata_cache=self.metadata_cache, create_finding=self._create_finding, should_stop=lambda: self.should_stop, + stop_event=self._stop_event, is_paused=lambda: not self.enabled, update_progress=self._update_progress, report_progress=_report_progress, @@ -595,6 +604,17 @@ class RepairWorker: self._current_job_name = None 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): """Queue a job for immediate execution by the main worker loop. diff --git a/core/soulid_worker.py b/core/soulid_worker.py index aeebd4a9..cc800460 100644 --- a/core/soulid_worker.py +++ b/core/soulid_worker.py @@ -24,6 +24,7 @@ import unicodedata from typing import Dict, Any, List, Optional from utils.logging_config import get_logger +from core.worker_utils import interruptible_sleep logger = get_logger("soulid_worker") @@ -79,6 +80,7 @@ class SoulIDWorker: self.should_stop = False self.thread = None self.current_item = None + self._stop_event = threading.Event() # API clients (lazy-initialized) self._itunes_client = None @@ -135,6 +137,7 @@ class SoulIDWorker: return self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("SoulID worker started") @@ -144,8 +147,9 @@ class SoulIDWorker: return self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("SoulID worker stopped") def pause(self): @@ -178,7 +182,7 @@ class SoulIDWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue processed = 0 @@ -188,16 +192,16 @@ class SoulIDWorker: if processed == 0: self.current_item = None - time.sleep(self.idle_sleep) + interruptible_sleep(self._stop_event, self.idle_sleep) else: # Albums/tracks get inter_batch_sleep, artists get their # 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: logger.error(f"Error in SoulID worker loop: {e}", exc_info=True) self.stats['errors'] += 1 - time.sleep(5) + interruptible_sleep(self._stop_event, 5) self.current_item = None 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 "")) # Rate limit courtesy for API calls - time.sleep(self.artist_sleep) + interruptible_sleep(self._stop_event, self.artist_sleep) return 1 except Exception as e: @@ -324,7 +328,7 @@ class SoulIDWorker: deezer_artist_id = int(raw_id) logger.debug(f"Deezer artist ID for '{artist_name}': {deezer_artist_id}") break - time.sleep(0.3) + interruptible_sleep(self._stop_event, 0.3) except Exception as e: logger.debug(f"Deezer track search failed for '{artist_name}': {e}") @@ -344,7 +348,7 @@ class SoulIDWorker: itunes_artist_id = int(raw_id) logger.debug(f"iTunes artist ID for '{artist_name}': {itunes_artist_id}") break - time.sleep(0.3) + interruptible_sleep(self._stop_event, 0.3) except Exception as e: logger.debug(f"iTunes track search failed for '{artist_name}': {e}") diff --git a/core/spotify_worker.py b/core/spotify_worker.py index 17342528..28beaf5f 100644 --- a/core/spotify_worker.py +++ b/core/spotify_worker.py @@ -8,6 +8,7 @@ from datetime import datetime, date, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.spotify_client import SpotifyClient, SpotifyRateLimitError +from core.worker_utils import interruptible_sleep logger = get_logger("spotify_worker") @@ -31,6 +32,7 @@ class SpotifyWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -66,6 +68,7 @@ class SpotifyWorker: return self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Spotify background worker started") @@ -76,8 +79,9 @@ class SpotifyWorker: logger.info("Stopping Spotify worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Spotify worker stopped") def pause(self): @@ -167,7 +171,7 @@ class SpotifyWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # Rate limit guard — if globally rate limited, sleep until ban expires @@ -175,7 +179,7 @@ class SpotifyWorker: info = self.client.get_rate_limit_info() remaining = info['remaining_seconds'] if info else 60 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 # Daily budget guard — worker-only cap to avoid saturating Spotify rate limits @@ -184,7 +188,7 @@ class SpotifyWorker: resets_in = budget['resets_in_seconds'] logger.info(f"Daily enrichment budget exhausted ({budget['used']}/{budget['limit']}), " 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 # Post-ban cooldown guard — after ban expires, wait before resuming @@ -192,7 +196,7 @@ class SpotifyWorker: cooldown = self.client.get_post_ban_cooldown_remaining() if cooldown > 0: 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 # Auth guard — check if Spotify client is configured (no API call). @@ -203,7 +207,7 @@ class SpotifyWorker: self.client.reload_config() if not self.client.is_spotify_authenticated(): logger.debug("Spotify not authenticated, sleeping 30s...") - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue self.current_item = None @@ -211,7 +215,7 @@ class SpotifyWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -229,14 +233,14 @@ class SpotifyWorker: self._process_item(item) self._increment_daily_budget() - time.sleep(self.inter_item_sleep) + interruptible_sleep(self._stop_event, self.inter_item_sleep) except SpotifyRateLimitError: 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: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) self.current_item = None logger.info("Spotify worker thread finished") @@ -541,7 +545,7 @@ class SpotifyWorker: self._mark_status('album', db_id, 'not_found') 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") @@ -614,7 +618,7 @@ class SpotifyWorker: self._mark_status('track', db_id, 'not_found') 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") diff --git a/core/tidal_worker.py b/core/tidal_worker.py index c02e398a..097227fc 100644 --- a/core/tidal_worker.py +++ b/core/tidal_worker.py @@ -7,6 +7,7 @@ from datetime import datetime, timedelta from utils.logging_config import get_logger from database.music_database import MusicDatabase from core.tidal_client import TidalClient +from core.worker_utils import interruptible_sleep logger = get_logger("tidal_worker") @@ -45,6 +46,7 @@ class TidalWorker: self.paused = False self.should_stop = False self.thread = None + self._stop_event = threading.Event() # Current item being processed (for UI tooltip) self.current_item = None @@ -73,6 +75,7 @@ class TidalWorker: self.running = True self.should_stop = False + self._stop_event.clear() self.thread = threading.Thread(target=self._run, daemon=True) self.thread.start() logger.info("Tidal background worker started") @@ -85,9 +88,10 @@ class TidalWorker: logger.info("Stopping Tidal worker...") self.should_stop = True self.running = False + self._stop_event.set() if self.thread: - self.thread.join(timeout=5) + self.thread.join(timeout=1) logger.info("Tidal worker stopped") @@ -140,17 +144,17 @@ class TidalWorker: while not self.should_stop: try: if self.paused: - time.sleep(1) + interruptible_sleep(self._stop_event, 1) continue # Auth guard: sleep if not authenticated try: if not self.client.is_authenticated(): self.current_item = None - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue except Exception: - time.sleep(30) + interruptible_sleep(self._stop_event, 30) continue self.current_item = None @@ -159,7 +163,7 @@ class TidalWorker: if not item: logger.debug("No pending items, sleeping...") - time.sleep(10) + interruptible_sleep(self._stop_event, 10) continue self.current_item = item @@ -178,11 +182,11 @@ class TidalWorker: self._process_item(item) - time.sleep(2) + interruptible_sleep(self._stop_event, 2) except Exception as e: logger.error(f"Error in worker loop: {e}") - time.sleep(5) + interruptible_sleep(self._stop_event, 5) logger.info("Tidal worker thread finished") @@ -364,7 +368,7 @@ class TidalWorker: if '429' in error_str or 'rate limit' in error_str: # 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") - time.sleep(30) + interruptible_sleep(self._stop_event, 30) return logger.error(f"Error processing {item['type']} #{item['id']}: {e}") self.stats['errors'] += 1 diff --git a/core/web_scan_manager.py b/core/web_scan_manager.py index 343d36b4..e13d2d72 100644 --- a/core/web_scan_manager.py +++ b/core/web_scan_manager.py @@ -38,6 +38,8 @@ class WebScanManager: self._max_scan_time = 1800 # 30 minutes maximum self._current_server_type = None self._scan_progress = {} + self._completion_check_timer = None + self._shutting_down = False 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}") 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 if callback and callback not in self._scan_completion_callbacks: self._scan_completion_callbacks.append(callback) @@ -107,6 +117,7 @@ class WebScanManager: # Start the debounce timer self._timer = threading.Timer(self.delay, self._execute_scan) + self._timer.daemon = True self._timer.start() return { @@ -169,6 +180,9 @@ class WebScanManager: def _execute_scan(self): """Execute the actual media library scan""" with self._lock: + if self._shutting_down: + logger.debug("Web scan execution skipped during shutdown") + return if self._scan_in_progress: logger.warning("Web scan already in progress - skipping duplicate execution") return @@ -231,6 +245,9 @@ class WebScanManager: """Start periodic checking for scan completion""" def check_completion(): try: + if self._shutting_down: + logger.debug("Web scan completion check aborted during shutdown") + return # Check for timeout 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") @@ -254,14 +271,22 @@ class WebScanManager: self._handle_scan_completion() else: # 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: logger.error(f"Error during web scan completion check: {e}") self._reset_scan_state() # 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): """Handle scan completion and trigger callbacks""" @@ -295,4 +320,24 @@ class WebScanManager: self._current_server_type = None self._scan_start_time = None self._scan_progress = {} - # Don't clear callbacks - they might be reused \ No newline at end of file + # 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") diff --git a/core/worker_utils.py b/core/worker_utils.py new file mode 100644 index 00000000..0c61fb0a --- /dev/null +++ b/core/worker_utils.py @@ -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() diff --git a/database/music_database.py b/database/music_database.py index 95b54646..285ac243 100644 --- a/database/music_database.py +++ b/database/music_database.py @@ -16,6 +16,7 @@ from utils.logging_config import get_logger logger = get_logger("music_database") _database_initialized_paths = set() +_database_sidecar_warnings = set() _database_initialization_lock = threading.Lock() # Import matching engine for enhanced similarity logic @@ -170,10 +171,57 @@ class MusicDatabase: database_path = os.environ.get('DATABASE_PATH', 'database/music_library.db') self.database_path = Path(database_path) self.database_path.parent.mkdir(parents=True, exist_ok=True) + self._warn_about_stale_sqlite_sidecars() # Initialize database once per process for this path 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): """Run schema setup and migrations once per database path per process.""" db_key = str(self.database_path.resolve()) diff --git a/web_server.py b/web_server.py index 08bdcf8e..fabfeb88 100644 --- a/web_server.py +++ b/web_server.py @@ -2517,27 +2517,41 @@ class WebUIDownloadMonitor: self.monitoring = False self.monitor_thread = None self.monitored_batches = set() + self._lock = threading.Lock() def start_monitoring(self, batch_id): """Start monitoring a download batch""" - self.monitored_batches.add(batch_id) - if not self.monitoring: - self.monitoring = True - self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True) - self.monitor_thread.start() - print(f"Started download monitor for batch {batch_id}") + with self._lock: + self.monitored_batches.add(batch_id) + if not self.monitoring: + self.monitoring = True + self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True) + self.monitor_thread.start() + print(f"Started download monitor for batch {batch_id}") def stop_monitoring(self, batch_id): """Stop monitoring a specific batch""" - self.monitored_batches.discard(batch_id) - if not self.monitored_batches: + with self._lock: + 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 - 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): """Main monitoring loop - checks downloads every 1 second for responsive web UX""" while self.monitoring and self.monitored_batches: try: + if globals().get('IS_SHUTTING_DOWN', False): + self.monitoring = False + break self._check_all_downloads() time.sleep(1) # 1-second polling for fast web UI updates except Exception as e: @@ -3206,6 +3220,8 @@ def validate_and_heal_batch_states(): This is the server-side equivalent of the frontend's worker count validation. """ try: + if globals().get('IS_SHUTTING_DOWN', False): + return import time current_time = time.time() @@ -3310,15 +3326,41 @@ def validate_and_heal_batch_states(): # Start periodic batch healing (every 30 seconds) 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(): """Start periodic batch state validation and healing""" try: + if globals().get('IS_SHUTTING_DOWN', False): + return validate_and_heal_batch_states() except Exception as e: print(f"[Batch Healing Timer] Error: {e}") finally: # 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_batch_healing_timer() @@ -3332,35 +3374,85 @@ def cleanup_monitor(): """Clean up background monitor on shutdown""" if download_monitor.monitoring: print("Flask shutdown detected, stopping download monitor...") - download_monitor.monitoring = False - download_monitor.monitored_batches.clear() + download_monitor.shutdown() # Give the thread a moment to exit cleanly time.sleep(0.5) # Clean up batch locks to prevent memory leaks - with tasks_lock: - batch_locks.clear() - print("Cleaned up batch locks") + try: + acquired = tasks_lock.acquire(timeout=1.0) + 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 IS_SHUTTING_DOWN = False -def signal_handler(signum, frame): - """Handle SIGINT (Ctrl+C) and SIGTERM""" - global IS_SHUTTING_DOWN - print(f"Signal {signum} received, cleaning up...") - IS_SHUTTING_DOWN = True - cleanup_monitor() - - # Stop automation engine +def _shutdown_executor(executor, name): + """Shut down a ThreadPoolExecutor without waiting for long-running tasks.""" + if executor is None: + return try: - if automation_engine: - print("Stopping automation engine...") - automation_engine.stop() + print(f"Shutting down {name}...") + executor.shutdown(wait=False, cancel_futures=True) 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: from core.api_call_tracker import api_call_tracker api_call_tracker.save() @@ -3368,13 +3460,52 @@ def signal_handler(signum, frame): except Exception as e: print(f"Error saving API call history: {e}") - # Shutdown executor to prevent new tasks - try: - print("Shutting down missing_download_executor...") - missing_download_executor.shutdown(wait=False, cancel_futures=True) - 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) # Register cleanup handlers @@ -3385,7 +3516,14 @@ def _atexit_save_history(): except Exception: pass +def _atexit_shutdown(): + try: + _shutdown_runtime_components() + except Exception: + pass + atexit.register(_atexit_save_history) +atexit.register(_atexit_shutdown) atexit.register(cleanup_monitor) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler)