Provider migration: survive its own restart, carry fingerprints and the unsignable marker, and stop tie-breaks skewing the dashboard pies

Author: NeptuneHub

app_dashboard.py

@@ -597,6 +597,7 @@ def _collect_charts_metrics(cur):
     # never buffers all rows at once (an unnamed cursor would).
     mood_dominant_counts = {}
     other_feature_dominant_counts = {}
+    tied_songs = 0
     complete = True
     try:
         with cur.connection.cursor(name='dash_mood_scan') as scan:
@@ -611,26 +612,35 @@ def _collect_charts_metrics(cur):
                 parsed = _parse_keyval(mv)
                 if not parsed:
                     continue
-                dom = max(parsed.items(), key=lambda kv: kv[1])[0]
-                mood_dominant_counts[dom] = mood_dominant_counts.get(dom, 0) + 1
+                dom = _dominant_label(parsed)
+                if dom:
+                    mood_dominant_counts[dom[0]] = mood_dominant_counts.get(dom[0], 0) + 1
+                else:
+                    tied_songs += 1
 
                 if of:
                     emotional_scores = {
                         k: s for k, s in _parse_keyval(of).items()
                         if k not in ('tempo_normalized', 'energy_normalized')
                     }
-                    top = max(
-                        emotional_scores.items(), key=lambda kv: kv[1], default=None
-                    )
+                    top = _dominant_label(emotional_scores)
                     # analyze_track writes ZERO_OTHER_FEATURES up front and only
                     # refresh_other_features fills it in once CLAP lands, so an
                     # all-zero row means "not scored yet", not "the least
                     # danceable song in the library". Counting its argmax would
                     # hand every un-CLAPped song to whichever label parses first.
-                    if top and top[1] > 0:
+                    if top is None:
+                        tied_songs += 1
+                    elif top[1] > 0:
                         other_feature_dominant_counts[top[0]] = (
                             other_feature_dominant_counts.get(top[0], 0) + 1
                         )
+        if tied_songs:
+            logger.info(
+                "dashboard: %d song(s) had no single strongest label and were left "
+                "out of the dominance counts rather than handed to whichever label "
+                "parses first", tied_songs,
+            )
     except Exception as e:
         logger.debug(f"dashboard: mood aggregation failed: {e}")
         _safe_rollback(cur)
@@ -687,6 +697,16 @@ def _collect_charts_metrics(cur):
     return metrics
 
 
+def _dominant_label(scores):
+    if not scores:
+        return None
+    best = max(scores.values())
+    winners = [label for label, score in scores.items() if score == best]
+    if len(winners) != 1:
+        return None
+    return winners[0], best
+
+
 def _parse_keyval(s):
     """Parse a ``key:value,key:value`` string (as stored in the ``score``
     table's ``mood_vector`` / ``other_features`` columns) into a dict of

app_provider_migration.py

@@ -184,6 +184,20 @@ def _current_provider_creds():
     return None, {}
 
 
+def _overrides_by_catalogue_id(by_provider_id):
+    if not by_provider_id:
+        return {}
+    from tasks.mediaserver import registry
+
+    canonical_of = registry.canonical_input_ids(list(by_provider_id.keys()))
+    overrides = {}
+    for provider_id in sorted(by_provider_id):
+        catalogue_id = canonical_of.get(provider_id, provider_id)
+        if catalogue_id not in overrides:
+            overrides[catalogue_id] = by_provider_id[provider_id]
+    return overrides
+
+
 def _apply_source_path_overrides(old_rows, overrides):
     """Patch ``old_rows[i]['file_path']`` from the overrides dict in place.
 
@@ -831,7 +845,9 @@ def run_source_refresh_core(session_id):
     tracks = provider_probe.fetch_all_tracks(source_type, creds)
 
     path_format = _detect_path_format(tracks)
-    overrides = {t['id']: t['path'] for t in tracks if t.get('id') and t.get('path')}
+    overrides = _overrides_by_catalogue_id(
+        {t['id']: t['path'] for t in tracks if t.get('id') and t.get('path')}
+    )
 
     warnings = []
     if path_format != 'absolute':

tasks/fingerprint_canonicalize.py

@@ -109,9 +109,15 @@
 # 'analysis' match tier. Such a row can never be relabelled, so counting it as
 # legacy work made this "one-time" migration re-hash the whole catalogue on EVERY
 # boot and relabel nothing.
+# The fp_0 head IS the marker for a row minted as unsignable, and unlike the map
+# row it cannot be taken away: a provider migration unbinds an unmatched song from
+# its server, which used to strip the only evidence and hand the row straight back
+# to this migration as legacy work it can never relabel.
 _UNSIGNABLE_SQL = (
-    "EXISTS (SELECT 1 FROM track_server_map t "
-    "WHERE t.item_id = s.item_id AND t.match_tier = 'analysis')"
+    "((s.item_id LIKE 'fp\\_0%%' AND length(s.item_id) = "
+    + str(simhash.CANONICAL_ID_LEN)
+    + ") OR EXISTS (SELECT 1 FROM track_server_map t "
+    "WHERE t.item_id = s.item_id AND t.match_tier = 'analysis'))"
 )
 _LEGACY_ROW_SQL = "NOT " + _CURRENT_SCHEME_SQL + " AND NOT " + _UNSIGNABLE_SQL
 _RELABEL_ADVISORY_LOCK = 726354822

tasks/multiserver_sync.py

@@ -104,6 +104,12 @@ def _sweep_job_state(task_id):
 
 _recovery_state = {'last': None}
 
+# A row is written BEFORE its RQ job is enqueued (and, for a provider migration,
+# committed with the migration itself), so a brand-new row legitimately has no job
+# yet. Without this, recovery could revoke a perfectly good alignment microseconds
+# before its enqueue landed and run a second one alongside it.
+_ENQUEUE_GRACE_SECONDS = 120
+
 _ABANDONED_FIRST_SEEN = {}
 
 _ABANDONED_CONFIRM_SECONDS = 60
@@ -135,23 +141,28 @@ def _details_full_refresh(details):
     return bool(details.get('full_refresh')) if isinstance(details, dict) else False
 
 
-def enqueue_server_alignment(server_id=None, message=None):
-    """Queue a full-refresh alignment of ONE server (the default when unnamed).
-
-    Raw connection and a direct enqueue, like ``recover_abandoned_sweeps``, so a
-    caller with no Flask app context - the provider migration runs in an RQ job -
-    can ask for the alignment that rebuilds what a provider swap cannot carry:
-    the server's artist ids and the file paths of the tracks it repointed.
-    Returns the task id, or None when there is no server to align.
-    """
+def insert_pending_sweep_row(cur, task_id, message, full_refresh=True):
     import config
+
+    details = json.dumps({
+        'message': message,
+        'status_message': message,
+        'full_refresh': bool(full_refresh),
+    })
+    cur.execute(
+        "INSERT INTO task_status "
+        "(task_id, task_type, status, progress, details, timestamp, start_time) "
+        "VALUES (%s, %s, %s, 0, %s, NOW(), %s) "
+        "ON CONFLICT (task_id) DO NOTHING",
+        (task_id, SWEEP_TASK_TYPE, config.TASK_STATUS_PENDING, details, time.time()),
+    )
+
+
+def enqueue_server_alignment(server_id=None, message=None, task_id=None):
     from app_helper import rq_queue_high
 
-    task_id = str(uuid.uuid4())
+    task_id = task_id or str(uuid.uuid4())
     text = message or 'Server alignment queued.'
-    details = json.dumps(
-        {'message': text, 'status_message': text, 'full_refresh': True}
-    )
     db = connect_raw()
     db.autocommit = True
     try:
@@ -160,13 +171,7 @@ def enqueue_server_alignment(server_id=None, message=None):
             return None
         cur = db.cursor()
         try:
-            cur.execute(
-                "INSERT INTO task_status "
-                "(task_id, task_type, status, progress, details, timestamp, start_time) "
-                "VALUES (%s, %s, %s, 0, %s, NOW(), %s) "
-                "ON CONFLICT (task_id) DO NOTHING",
-                (task_id, SWEEP_TASK_TYPE, config.TASK_STATUS_PENDING, details, time.time()),
-            )
+            insert_pending_sweep_row(cur, task_id, text)
         finally:
             cur.close()
     finally:
@@ -220,9 +225,11 @@ def recover_abandoned_sweeps():
         try:
             cur.execute(
                 "SELECT task_id, details FROM task_status WHERE task_type = %s "
-                "AND status NOT IN (%s, %s, %s)",
+                "AND status NOT IN (%s, %s, %s) "
+                "AND timestamp < NOW() - make_interval(secs => %s)",
                 (SWEEP_TASK_TYPE, config.TASK_STATUS_SUCCESS,
-                 config.TASK_STATUS_FAILURE, config.TASK_STATUS_REVOKED),
+                 config.TASK_STATUS_FAILURE, config.TASK_STATUS_REVOKED,
+                 _ENQUEUE_GRACE_SECONDS),
             )
             rows = cur.fetchall()
             candidates = [r[0] for r in rows]
@@ -264,16 +271,10 @@ def recover_abandoned_sweeps():
             if not revoked_count:
                 return None
             new_task_id = str(uuid.uuid4())
-            queued = json.dumps({
-                'message': 'Server alignment queued for all servers.',
-                'full_refresh': full_refresh,
-            })
-            cur.execute(
-                "INSERT INTO task_status "
-                "(task_id, task_type, status, progress, details, timestamp, start_time) "
-                "VALUES (%s, %s, %s, 0, %s, NOW(), %s) "
-                "ON CONFLICT (task_id) DO NOTHING",
-                (new_task_id, SWEEP_TASK_TYPE, config.TASK_STATUS_PENDING, queued, now),
+            insert_pending_sweep_row(
+                cur, new_task_id,
+                'Server alignment queued for all servers.',
+                full_refresh=full_refresh,
             )
         finally:
             cur.close()