Persist API call history, record rate limit events, fix Spotify re-auth issues
API Call Tracker: - Save/load 24h minute-bucketed history + events to database/api_call_history.json - Persists across server restarts via atexit + signal handler hooks - New record_event() for rate limit bans (called from _set_global_rate_limit) - New get_debug_summary() for Copy Debug Info — 24h totals, peak cpm with timestamp, per-endpoint breakdown, and last 20 rate limit events - Fixed race condition: events iteration now inside lock during save Spotify Rate Limit Mitigation: - Enrichment worker: max_pages=5 on get_artist_albums (was unlimited — artist with 217 albums caused 22 paginated API calls, now capped at 5) - Enrichment worker: inter_item_sleep raised from 0.5s to 1.5s Spotify Re-Auth Fix: - Both OAuth callbacks (port 8008 + 8888) now clear rate limit ban AND post-ban cooldown after successful re-auth — Spotify usable immediately instead of stuck on Deezer fallback for 5 minutes - Auth cache invalidated on both global client and enrichment worker client
This commit is contained in:
Broque Thomas
parent
9f7fe27e7c
commit
e674a79c88
4 changed files with 203 additions and 6 deletions
|
|
@ -3,9 +3,11 @@ Centralized API call tracker for all enrichment services.
|
|||
|
||||
Tracks actual API calls (not items processed) with rolling timestamps
|
||||
for real-time rate monitoring and minute-bucketed history for 24-hour graphs.
|
||||
Thread-safe, no persistence (resets on restart).
|
||||
Thread-safe, persists 24h history to disk on shutdown and restores on startup.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
from collections import deque, defaultdict
|
||||
|
|
@ -46,6 +48,9 @@ SERVICE_ORDER = [
|
|||
]
|
||||
|
||||
|
||||
_PERSIST_PATH = os.path.join('database', 'api_call_history.json')
|
||||
|
||||
|
||||
class ApiCallTracker:
|
||||
"""Centralized tracker for actual API calls across all enrichment services."""
|
||||
|
||||
|
|
@ -61,6 +66,13 @@ class ApiCallTracker:
|
|||
self._current_minute_counts = defaultdict(int)
|
||||
self._current_minute_ts = {}
|
||||
|
||||
# Rate limit event log — records bans, peaks, escalations
|
||||
# Each entry: {ts, event, service, endpoint, duration, detail}
|
||||
self._events = deque(maxlen=200)
|
||||
|
||||
# Restore persisted history from disk
|
||||
self._load()
|
||||
|
||||
def record_call(self, service_key, endpoint=None):
|
||||
"""Record an API call. Called from rate_limited decorators.
|
||||
|
||||
|
|
@ -104,6 +116,94 @@ class ApiCallTracker:
|
|||
else:
|
||||
self._current_minute_counts[key] += 1
|
||||
|
||||
def record_event(self, service_key, event_type, detail='', endpoint='', duration=0):
|
||||
"""Record a rate limit event (ban, escalation, cooldown, etc.).
|
||||
Called from spotify_client.py when rate limits are detected."""
|
||||
with self._lock:
|
||||
self._events.append({
|
||||
'ts': time.time(),
|
||||
'event': event_type,
|
||||
'service': service_key,
|
||||
'endpoint': endpoint,
|
||||
'duration': duration,
|
||||
'detail': detail,
|
||||
})
|
||||
|
||||
def get_events(self, since=None):
|
||||
"""Get rate limit events, optionally filtered by timestamp."""
|
||||
cutoff = since or (time.time() - 86400)
|
||||
with self._lock:
|
||||
return [e for e in self._events if e['ts'] >= cutoff]
|
||||
|
||||
def get_debug_summary(self):
|
||||
"""Build a comprehensive debug summary for Copy Debug Info.
|
||||
Includes 24h totals, peaks, rate limit events, and per-endpoint breakdown."""
|
||||
now = time.time()
|
||||
cutoff_24h = now - 86400
|
||||
summary = {}
|
||||
|
||||
with self._lock:
|
||||
for svc in SERVICE_ORDER:
|
||||
# 24h total calls
|
||||
total = 0
|
||||
peak_cpm = 0
|
||||
peak_ts = 0
|
||||
for ts, count in self._minute_history.get(svc, []):
|
||||
if ts >= cutoff_24h:
|
||||
total += count
|
||||
if count > peak_cpm:
|
||||
peak_cpm = count
|
||||
peak_ts = ts
|
||||
# Include current minute
|
||||
cur_ts = self._current_minute_ts.get(svc)
|
||||
cur_count = self._current_minute_counts.get(svc, 0)
|
||||
if cur_ts and cur_ts >= cutoff_24h:
|
||||
total += cur_count
|
||||
if cur_count > peak_cpm:
|
||||
peak_cpm = cur_count
|
||||
peak_ts = cur_ts
|
||||
|
||||
if total == 0:
|
||||
continue
|
||||
|
||||
entry = {
|
||||
'total_24h': total,
|
||||
'peak_cpm': peak_cpm,
|
||||
'limit_cpm': RATE_LIMITS.get(svc, 60),
|
||||
}
|
||||
if peak_ts:
|
||||
entry['peak_at'] = time.strftime('%Y-%m-%d %H:%M', time.localtime(peak_ts))
|
||||
summary[svc] = entry
|
||||
|
||||
# Spotify per-endpoint breakdown
|
||||
if svc == 'spotify':
|
||||
ep_totals = {}
|
||||
for key in list(self._minute_history.keys()):
|
||||
if key.startswith('spotify:'):
|
||||
ep_name = key[8:]
|
||||
ep_total = sum(c for ts, c in self._minute_history[key] if ts >= cutoff_24h)
|
||||
cur = self._current_minute_counts.get(key, 0)
|
||||
ep_total += cur
|
||||
if ep_total > 0:
|
||||
ep_totals[ep_name] = ep_total
|
||||
if ep_totals:
|
||||
summary[svc]['endpoints'] = ep_totals
|
||||
|
||||
# Rate limit events
|
||||
events = [e for e in self._events if e['ts'] >= cutoff_24h]
|
||||
|
||||
if events:
|
||||
summary['_rate_limit_events'] = [{
|
||||
'time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(e['ts'])),
|
||||
'event': e['event'],
|
||||
'service': e['service'],
|
||||
'endpoint': e.get('endpoint', ''),
|
||||
'duration': e.get('duration', 0),
|
||||
'detail': e.get('detail', ''),
|
||||
} for e in events[-20:]] # Last 20 events
|
||||
|
||||
return summary
|
||||
|
||||
def get_calls_per_minute(self, service_key):
|
||||
"""Get current calls/minute rate from last 60 seconds."""
|
||||
now = time.time()
|
||||
|
|
@ -161,5 +261,54 @@ class ApiCallTracker:
|
|||
return result
|
||||
|
||||
|
||||
def save(self):
|
||||
"""Persist 24h minute history to disk. Call on shutdown."""
|
||||
try:
|
||||
now = time.time()
|
||||
cutoff = now - 86400
|
||||
data = {}
|
||||
with self._lock:
|
||||
for key, hist in self._minute_history.items():
|
||||
entries = [[ts, count] for ts, count in hist if ts >= cutoff]
|
||||
# Include current in-progress minute
|
||||
cur_ts = self._current_minute_ts.get(key)
|
||||
cur_count = self._current_minute_counts.get(key, 0)
|
||||
if cur_ts is not None and cur_count > 0 and cur_ts >= cutoff:
|
||||
entries.append([cur_ts, cur_count])
|
||||
if entries:
|
||||
data[key] = entries
|
||||
events = [dict(e) for e in self._events if e['ts'] >= cutoff]
|
||||
with open(_PERSIST_PATH, 'w') as f:
|
||||
json.dump({'ts': now, 'history': data, 'events': events}, f)
|
||||
except Exception as e:
|
||||
print(f"[ApiCallTracker] Failed to save history: {e}")
|
||||
|
||||
def _load(self):
|
||||
"""Restore 24h minute history from disk. Called on init."""
|
||||
try:
|
||||
if not os.path.exists(_PERSIST_PATH):
|
||||
return
|
||||
with open(_PERSIST_PATH, 'r') as f:
|
||||
raw = json.load(f)
|
||||
saved_ts = raw.get('ts', 0)
|
||||
# Only restore if saved within last 24h
|
||||
if time.time() - saved_ts > 86400:
|
||||
return
|
||||
history = raw.get('history', {})
|
||||
events = raw.get('events', [])
|
||||
cutoff = time.time() - 86400
|
||||
with self._lock:
|
||||
for key, entries in history.items():
|
||||
for ts, count in entries:
|
||||
if ts >= cutoff:
|
||||
self._minute_history[key].append((ts, count))
|
||||
for e in events:
|
||||
if e.get('ts', 0) >= cutoff:
|
||||
self._events.append(e)
|
||||
print(f"[ApiCallTracker] Restored history for {len(history)} services, {len(events)} events")
|
||||
except Exception as e:
|
||||
print(f"[ApiCallTracker] Failed to load history: {e}")
|
||||
|
||||
|
||||
# Singleton instance
|
||||
api_call_tracker = ApiCallTracker()
|
||||
|
|
|
|||
|
|
@ -103,6 +103,19 @@ def _set_global_rate_limit(retry_after_seconds, endpoint_name, has_real_header=F
|
|||
f"(expires {time.strftime('%H:%M:%S', time.localtime(new_until))}) "
|
||||
f"triggered by {endpoint_name}"
|
||||
)
|
||||
# Record event for debug diagnostics
|
||||
try:
|
||||
from core.api_call_tracker import api_call_tracker
|
||||
escalated = _rate_limit_hit_count > 1
|
||||
api_call_tracker.record_event(
|
||||
'spotify', 'rate_limit_ban',
|
||||
endpoint=endpoint_name,
|
||||
duration=retry_after_seconds,
|
||||
detail=f'{"escalation #" + str(_rate_limit_hit_count) if escalated else "initial"}'
|
||||
f'{", real Retry-After" if has_real_header else ", estimated"}'
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _is_globally_rate_limited():
|
||||
|
|
|
|||
|
|
@ -49,8 +49,8 @@ class SpotifyWorker:
|
|||
# Name matching threshold
|
||||
self.name_similarity_threshold = 0.80
|
||||
|
||||
# Rate limiting (SpotifyClient already rate-limits at 200ms between API calls)
|
||||
self.inter_item_sleep = 0.5 # Between top-level items
|
||||
# Rate limiting (SpotifyClient already rate-limits at 350ms between API calls)
|
||||
self.inter_item_sleep = 1.5 # Between top-level items (each can trigger 5+ paginated calls)
|
||||
self.batch_inter_item_sleep = 0.1 # Between local matches within a batch (no API calls)
|
||||
|
||||
# Daily budget — caps how many items this worker processes per calendar day
|
||||
|
|
@ -491,10 +491,13 @@ class SpotifyWorker:
|
|||
spotify_artist_id = item['spotify_artist_id']
|
||||
artist_name = item['artist_name']
|
||||
|
||||
# 1 API call: get all albums for this artist from Spotify
|
||||
# Fetch albums with pagination cap — Spotify returns 10/page, so max_pages=5
|
||||
# gives 50 albums (newest first). Avoids 20+ paginated calls for prolific artists
|
||||
# (e.g., 217 albums = 22 API calls without cap, vs 5 with cap)
|
||||
try:
|
||||
spotify_albums = self.client.get_artist_albums(
|
||||
spotify_artist_id, album_type='album,single,compilation', limit=50
|
||||
spotify_artist_id, album_type='album,single,compilation', limit=50,
|
||||
max_pages=5
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to get Spotify albums for artist '{artist_name}': {e}")
|
||||
|
|
|
|||
|
|
@ -3243,6 +3243,14 @@ def signal_handler(signum, frame):
|
|||
except Exception as e:
|
||||
print(f"⚠️ Error stopping automation engine: {e}")
|
||||
|
||||
# Persist API call history
|
||||
try:
|
||||
from core.api_call_tracker import api_call_tracker
|
||||
api_call_tracker.save()
|
||||
print("💾 API call history saved")
|
||||
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...")
|
||||
|
|
@ -3253,6 +3261,14 @@ def signal_handler(signum, frame):
|
|||
sys.exit(0)
|
||||
|
||||
# Register cleanup handlers
|
||||
def _atexit_save_history():
|
||||
try:
|
||||
from core.api_call_tracker import api_call_tracker
|
||||
api_call_tracker.save()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
atexit.register(_atexit_save_history)
|
||||
atexit.register(cleanup_monitor)
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
|
|
@ -4801,11 +4817,13 @@ def get_debug_info():
|
|||
elif not soulseek_client:
|
||||
info['download_client_failures'] = ['ALL (orchestrator failed to initialize)']
|
||||
|
||||
# API rate monitor — current calls/min per service + Spotify rate limit state
|
||||
# API rate monitor — current calls/min, 24h totals, peaks, rate limit events
|
||||
try:
|
||||
from core.api_call_tracker import api_call_tracker
|
||||
rates = api_call_tracker.get_all_rates()
|
||||
info['api_rates'] = rates
|
||||
# Rich 24h debug summary with peaks, totals, per-endpoint breakdown, events
|
||||
info['api_debug_summary'] = api_call_tracker.get_debug_summary()
|
||||
# Spotify rate limit details
|
||||
if spotify_client:
|
||||
rl_info = spotify_client.get_rate_limit_info()
|
||||
|
|
@ -4821,6 +4839,7 @@ def get_debug_info():
|
|||
info['spotify_rate_limit'] = {'active': False}
|
||||
except Exception:
|
||||
info['api_rates'] = {}
|
||||
info['api_debug_summary'] = {}
|
||||
info['spotify_rate_limit'] = {'active': False}
|
||||
|
||||
# Database size
|
||||
|
|
@ -6723,13 +6742,21 @@ def spotify_callback():
|
|||
token_info = auth_manager.get_access_token(auth_code, as_dict=True)
|
||||
|
||||
if token_info:
|
||||
# CRITICAL: update the GLOBAL spotify_client, not a local variable
|
||||
global spotify_client
|
||||
spotify_client = SpotifyClient()
|
||||
if spotify_client.is_authenticated():
|
||||
# Clear any active rate limit ban and post-ban cooldown
|
||||
# so Spotify is immediately usable after re-auth
|
||||
from core.spotify_client import _clear_rate_limit
|
||||
_clear_rate_limit()
|
||||
spotify_client._invalidate_auth_cache()
|
||||
# Invalidate status cache so next poll picks up the new connection
|
||||
_status_cache_timestamps['spotify'] = 0
|
||||
# Refresh enrichment worker's client so it picks up new auth
|
||||
if spotify_enrichment_worker and hasattr(spotify_enrichment_worker, 'client'):
|
||||
spotify_enrichment_worker.client.reload_config()
|
||||
spotify_enrichment_worker.client._invalidate_auth_cache()
|
||||
add_activity_item("✅", "Spotify Auth Complete", "Successfully authenticated with Spotify", "Now")
|
||||
return "<h1>Spotify Authentication Successful!</h1><p>You can close this window.</p>"
|
||||
else:
|
||||
|
|
@ -47633,11 +47660,16 @@ def start_oauth_callback_servers():
|
|||
spotify_client = SpotifyClient()
|
||||
|
||||
if spotify_client.is_authenticated():
|
||||
# Clear rate limit ban + post-ban cooldown so Spotify is usable immediately
|
||||
from core.spotify_client import _clear_rate_limit
|
||||
_clear_rate_limit()
|
||||
spotify_client._invalidate_auth_cache()
|
||||
# Invalidate status cache so next poll picks up the new connection
|
||||
_status_cache_timestamps['spotify'] = 0
|
||||
# Refresh enrichment worker's client so it picks up new auth
|
||||
if spotify_enrichment_worker and hasattr(spotify_enrichment_worker, 'client'):
|
||||
spotify_enrichment_worker.client.reload_config()
|
||||
spotify_enrichment_worker.client._invalidate_auth_cache()
|
||||
add_activity_item("✅", "Spotify Auth Complete", "Successfully authenticated with Spotify", "Now")
|
||||
self.send_response(200)
|
||||
self.send_header('Content-type', 'text/html')
|
||||
|
|
|
|||
Loading…
Reference in a new issue