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

Improve graceful shutdown and rollback safety

Browse source 
- 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.
This commit is contained in:
Antti Kettunen 2026-04-12 15:15:53 +03:00
parent fff76a4be0
commit aec3047216
26 changed files with 520 additions and 142 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 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")

14
core/deezer_worker.py
View file

@ -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")

14
core/discogs_worker.py
View file

@ -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")

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 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")

12
core/hydrabase_worker.py
View file

@ -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()

20
core/itunes_worker.py
View file

@ -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")

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 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")

19
core/listening_stats_worker.py
View file

@ -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")

15
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_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")
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 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")

19
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_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")
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 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

3
core/repair_jobs/acoustid_scanner.py
View file

@ -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)

28
core/repair_jobs/base.py
View file

@ -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()

3
core/repair_jobs/library_reorganize.py
View file

@ -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)

3
core/repair_jobs/mbid_mismatch_detector.py
View file

@ -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:

3
core/repair_jobs/metadata_gap_filler.py
View file

@ -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)

3
core/repair_jobs/unknown_artist_fixer.py
View file

@ -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

34
core/repair_worker.py
View file

@ -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.

20
core/soulid_worker.py
View file

@ -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}")

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 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")

20
core/tidal_worker.py
View file

@ -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

51
core/web_scan_manager.py
View file

@ -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
# 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")
_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())

206
web_server.py
View file

@ -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)