diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..d5f33d5 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,78 @@ +name: Tests + +# WHY THIS EXISTS +# The repo had ~20 test files and nothing that ran them. Every suite was green +# only when someone remembered to run it by hand, and two order-dependency bugs +# had already crept in unnoticed (test modules fighting over db.py's module-level +# DB_PATH, which is decided by whichever test imports it first). +# +# pytest is the runner, not unittest: some suites use bare test functions that +# `python -m unittest` silently collects as zero tests and reports as success. + +on: + push: + branches: [main] + pull_request: + workflow_dispatch: + +permissions: + contents: read + +jobs: + python: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + service: [sync-api, sync-worker] + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + + - name: Install dependencies + working-directory: ${{ matrix.service }} + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install pytest httpx + # sync-worker's Dockerfile installs tiddl separately from requirements.txt. + # Mirror that here: without it _refresh_tidal_token cannot resolve the Tidal + # client credentials and returns early, so its test fails on an absent + # dependency rather than on anything real. + if [ "${{ matrix.service }}" = "sync-worker" ]; then pip install tiddl; fi + + - name: Run tests + working-directory: ${{ matrix.service }} + # Run the whole suite in ONE process on purpose: cross-module interference + # via shared module state is a real failure mode here, and running each + # file in isolation would hide exactly the bug this job exists to catch. + run: python -m pytest tests -q + + web: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-node@v4 + with: + node-version: "20" + cache: npm + cache-dependency-path: sync-web/package-lock.json + + - name: Install dependencies + working-directory: sync-web + run: npm ci + + - name: Typecheck + working-directory: sync-web + run: npx tsc --noEmit + + - name: Build + working-directory: sync-web + # `next build` catches what tsc cannot -- most importantly a useSearchParams() + # call outside a Suspense boundary, which fails at the prerender step. + run: npm run build diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d6a1c2..6df0c16 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,108 @@ # Changelog +## 2.13.0 — visible failures, real retries, and the wrong-version fix + +Three problems that were reported as feature requests turned out to be something +other than they looked. Measuring them first is what made this release small. + +### The Missing Tracks page was never empty — it was erroring + +`/upload` asked for `per_page=500` from an endpoint capped at 200. The API replied +with a 422 explaining exactly that, and the page rendered a clean empty table. + +The reason it survived so long is `api.ts`: it threw away the response body and +raised `Error("API error: 422")`, so every caller's `catch { setRows([]) }` turned a +loud, specific server error into "you have no missing tracks". The client now keeps +the status and FastAPI's `detail`, retries only transport faults and 5xx (never a +4xx, which is deterministic), and the page pages through all 270 errors instead of +truncating at the cap. + +### "Fingerprint mismatch" involved no fingerprints + +The category matched on duration, not on the chromaprint WaxFlow computes and +stores. 108 of the 109 tracks in it came from one path: the file-index title+artist +matcher, which accepted any file whose title and artist agreed — so an extended +mix, a radio edit or a live take all matched. Verify then rejected the file on +duration and parked the track as an error. + +Measured across 265 such matches on the live library: 155 land within 5s of the +Spotify duration and 110 exceed it, 95 of those by more than 15s. The two +populations separate cleanly, so title+artist matches now require the durations to +agree within 5s (configurable; fails open when either duration is unknown, as ~0.4% +of indexed files have none). The category is renamed **Wrong Version**, which is +what it always was. + +### Post-processing was working the whole time + +Of 400 tracks imported since July: 400 had BPM, 399 had cue points, 394 artwork, +399 cloud-backed — and only 4 tracks in the entire library lacked cues. Nothing was +broken; there was no way to see it. The dashboard now shows coverage for cue +points, beat grids, BPM, key, genre and tags, sampled hourly by the worker. + +It also honestly reports what it *cannot* measure: Lexicon's API exposes no artwork +or cloud-upload field, so those are named as unavailable rather than guessed at. + +One real bug did surface here: **"Auto-Analyze After Sync" silently disabled cue +generation, tag lookup and cloud upload too**, despite each having its own checkbox +in Settings and the toggle's own description mentioning only BPM/key detection. +Turning off analysis now turns off exactly analysis. + +### Bulk retry + +`POST /api/tracks/bulk-retry` takes explicit ids or a category name resolved +server-side by the same classifier that renders the count — so "Retry All 47" +acts on those 47, not on a set that drifted. The Errors page gains per-category +"Retry All" alongside the existing "Ignore All". + +It also **clears `fallback_attempts`**, without which a retry was quietly inert: +`already_attempted()` treats any prior row as "we tried this", so the track reset, +walked back down the pipeline, and failed identically without ever re-contacting +Soulseek. The existing single-track retry had the same defect and is fixed too. + +Bulk operations now write one summary activity row instead of one per track; +ignoring a category used to insert thousands and bury the dashboard feed. + +### Soulseek in service health + +`is_logged_in()` existed and was never surfaced. Because sync-api cannot import +worker code and has no slskd credentials, the worker probes every 120s and persists +the verdict to `app_config` for the API to serve — the pattern `lexicon_health` +already uses. It distinguishes logged-out (slskd answers, but searches return +nothing) from unreachable, and never-configured from broken. A verdict older than +15 minutes reports as unknown rather than repeating a stale "ok". + +The dashboard now renders whatever services the API reports rather than a hardcoded +list of three. + +### Dashboard drill-down + +Clicking a month's green/red/grey segment opens that exact set of tracks. The +`month` filter is a half-open range rather than `substr(spotify_added_at, 1, 7)`, so +it can use an index — and `tracks` had no index at all beyond its implicit primary +key one, so `spotify_added_at`, `pipeline_stage` and `lexicon_status` now have them. + +Segment widths come from flex-grow on raw counts; three independently rounded +percentages could total 101% and overflow the bar. + +### Also + +- The dashboard health probe called `GET /v1/tracks` — Lexicon's **entire library** + — every 10 seconds. It now calls `/v1/playlists`, as `lexicon_health` already did. +- The nav error badge polled every errored track in full for a single integer; + there is now a counts-only summary endpoint. +- **CI runs the tests.** The repo had ~20 test files and no workflow that executed + them. It runs pytest (not `unittest`, which silently collects zero tests from the + bare-function suites and reports success), plus a web typecheck and a real + `next build`. +- Fixed two order-dependency bugs the new CI immediately exposed: test modules were + fighting over `db.py`'s module-level `DB_PATH`, which is decided by whichever test + imports it first. +- The Tidal token-refresh test had been failing for anyone running the suite outside + the container. `tiddl` is installed by `sync-worker/Dockerfile` but is absent from + `requirements.txt`, so the function under test bailed out on missing credentials + before reaching the code the test was asserting on. CI now mirrors the Dockerfile. + + ## 2.12.4 — fix: the updater used the wrong compose project name The updater mounts the project at `/project`, and Compose derives the project name diff --git a/VERSION b/VERSION index 56beced..fb2c076 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.12.4 +2.13.0 diff --git a/sync-api/error_categories.py b/sync-api/error_categories.py new file mode 100644 index 0000000..8e5f308 --- /dev/null +++ b/sync-api/error_categories.py @@ -0,0 +1,65 @@ +"""Error-bucket classification for the Errors page. + +WHY THIS IS ITS OWN MODULE + The Errors page groups failures into buckets, and "Retry All" acts on a + bucket. If the endpoint that RENDERS a bucket and the endpoint that RETRIES + it classify tracks even slightly differently, the button retries a different + set than the number beside it claims. Both now call this one function, so + they cannot drift. + +The rules are ordered: the first match wins. `verify_codec` is checked before the +error text because lossy tracks used to fall through to `download_failed`. +""" + +from __future__ import annotations + +# Display order on the Errors page. +ERROR_CATEGORIES: tuple[str, ...] = ( + "not_lossless", + "no_tidal_match", + "download_failed", + "lexicon_sync_failed", + "wrong_version", + "other", +) + +# Renamed in 2.13.0. The old key said "fingerprint_mismatch", but no fingerprint +# is ever compared -- the check is duration-based, and what it actually catches is +# a DIFFERENT EDIT of the right song (radio edit vs extended mix, and so on). +# Kept so an older frontend still resolves the bucket. +CATEGORY_ALIASES: dict[str, str] = {"fingerprint_mismatch": "wrong_version"} + + +def canonical_category(name: str) -> str: + """Map a possibly-legacy category key to its current name.""" + return CATEGORY_ALIASES.get(name, name) + + +def categorize_error(track: dict) -> str: + """Bucket one errored track. Returns a key from ERROR_CATEGORIES.""" + err = (track.get("pipeline_error") or "").lower() + verify_status = track.get("verify_status") + verify_codec = (track.get("verify_codec") or "").lower() + + if verify_codec in ("aac", "mp3"): + return "not_lossless" + if verify_status == "fail" and "not lossless" in err: + return "not_lossless" + if "not lossless" in err or "aac" in err or "mp3" in err: + return "not_lossless" + if ( + "no tidal match" in err + or "no match" in err + or "not found on tidal" in err + or "permanently unavailable" in err + ): + return "no_tidal_match" + if "geo-restricted" in err or "unavailable on tidal" in err: + return "download_failed" + if "download failed" in err or "download error" in err: + return "download_failed" + if "lexicon" in err: + return "lexicon_sync_failed" + if "fingerprint" in err or "mismatched" in err or "wrong version" in err: + return "wrong_version" + return "other" diff --git a/sync-api/init_db.py b/sync-api/init_db.py index d769da5..10bad5d 100644 --- a/sync-api/init_db.py +++ b/sync-api/init_db.py @@ -583,6 +583,12 @@ def init(): ON source_attempts(track_id, source); CREATE INDEX IF NOT EXISTS idx_wanted_track ON wanted(track_id); CREATE INDEX IF NOT EXISTS idx_import_queue_state ON import_queue(state); + -- tracks carried no index at all beyond its implicit PK one. These three + -- back the dashboard month drill-down and the filters every list view uses. + CREATE INDEX IF NOT EXISTS idx_tracks_spotify_added_at + ON tracks(spotify_added_at); + CREATE INDEX IF NOT EXISTS idx_tracks_pipeline_stage ON tracks(pipeline_stage); + CREATE INDEX IF NOT EXISTS idx_tracks_lexicon_status ON tracks(lexicon_status); """) with get_db() as conn: cols = {r[1] for r in conn.execute("PRAGMA table_info(tracks)").fetchall()} diff --git a/sync-api/models.py b/sync-api/models.py index 30a6232..b89fb8a 100644 --- a/sync-api/models.py +++ b/sync-api/models.py @@ -58,6 +58,17 @@ class TrackUpdate(BaseModel): file_path: Optional[str] = None +class BulkRetryRequest(BaseModel): + """Retry many tracks by explicit id, or by error category resolved server-side. + + `limit` is a safety rail: a category retry can cover thousands of tracks, and + the pipeline picks work up every 10 seconds. + """ + track_ids: Optional[list[int]] = None + category: Optional[str] = None + limit: int = Field(default=500, ge=1, le=5000) + + class TrackListResponse(BaseModel): tracks: list[TrackOut] total: int diff --git a/sync-api/routes/dashboard.py b/sync-api/routes/dashboard.py index c7111b8..cef50b9 100644 --- a/sync-api/routes/dashboard.py +++ b/sync-api/routes/dashboard.py @@ -12,6 +12,57 @@ LEXICON_API = os.environ.get("LEXICON_API_URL", "http://localhost:48624") TIDARR_API = os.environ.get("TIDARR_URL", "http://localhost:8484") # optional legacy fallback +# The worker re-probes Soulseek every 120s. If the last verdict is older than this, +# the worker itself is the thing that is unwell, and reporting its last known "ok" +# would be reporting a lie. +_SOULSEEK_STALE_AFTER_SECONDS = 900 + +# Statuses that mean "deliberately off", not "broken". +_SOULSEEK_INACTIVE = {"disabled", "not_configured"} + + +def _soulseek_service(conn) -> ServiceHealth: + """Report the verdict last persisted by the worker's soulseek_health task.""" + keys = ("soulseek_health", "soulseek_detail", "soulseek_checked_at", "soulseek_latency_ms") + rows = conn.execute( + f"SELECT key, value FROM app_config WHERE key IN ({','.join('?' * len(keys))})", + keys, + ).fetchall() + cfg = {r["key"]: r["value"] for r in rows} + + status = (cfg.get("soulseek_health") or "").strip() + detail = cfg.get("soulseek_detail") or None + checked_at = cfg.get("soulseek_checked_at") + + if not status: + return ServiceHealth( + name="soulseek", status="unknown", + error="no health check recorded yet — is the worker running?", + ) + + try: + latency = float(cfg["soulseek_latency_ms"]) if cfg.get("soulseek_latency_ms") else None + except (TypeError, ValueError): + latency = None + + if checked_at and status not in _SOULSEEK_INACTIVE: + try: + from datetime import datetime, timezone + age = (datetime.now(timezone.utc) - datetime.fromisoformat(checked_at)).total_seconds() + if age > _SOULSEEK_STALE_AFTER_SECONDS: + return ServiceHealth( + name="soulseek", status="unknown", latency_ms=latency, + error=f"last checked {int(age // 60)} min ago — worker may be stalled", + ) + except (TypeError, ValueError): + pass + + if status == "ok": + return ServiceHealth(name="soulseek", status="ok", latency_ms=latency) + if status in _SOULSEEK_INACTIVE: + return ServiceHealth(name="soulseek", status="disabled", error=detail) + return ServiceHealth(name="soulseek", status="error", latency_ms=latency, error=detail) + @router.get("/dashboard", response_model=DashboardResponse) async def get_dashboard(): @@ -79,11 +130,14 @@ async def get_dashboard(): # Service health checks services = [] - # Lexicon + # Lexicon. Probe /v1/playlists, NOT /v1/tracks: this endpoint is polled every + # 10s by the dashboard, and /v1/tracks returns the ENTIRE library (5,600+ + # rows) on every single call. lexicon_health._check_lexicon_reachable already + # uses /v1/playlists for exactly this reason. try: t0 = time.monotonic() async with httpx.AsyncClient(timeout=5.0) as client: - resp = await client.get(f"{LEXICON_API}/v1/tracks") + resp = await client.get(f"{LEXICON_API}/v1/playlists") latency = round((time.monotonic() - t0) * 1000, 1) services.append(ServiceHealth( name="lexicon", @@ -109,6 +163,10 @@ async def get_dashboard(): except Exception as e: services.append(ServiceHealth(name="tidal", status="error", error=str(e))) + # Soulseek. The API has neither the slskd credentials nor the worker's client, + # so it reports what the worker's soulseek_health probe last persisted. + services.append(_soulseek_service(conn)) + return DashboardResponse( spotify_total=spotify_total, lexicon_synced=lexicon_synced, diff --git a/sync-api/routes/lexicon.py b/sync-api/routes/lexicon.py index e2ef95d..626e958 100644 --- a/sync-api/routes/lexicon.py +++ b/sync-api/routes/lexicon.py @@ -72,6 +72,45 @@ async def list_backups(): raise HTTPException(status_code=500, detail=str(e)) +@router.get("/coverage") +async def lexicon_coverage(): + """Post-processing coverage: how much of the library has cues, tags, key, bpm. + + Serves the rollup the worker's lexicon_coverage task computed. This endpoint + deliberately does NOT call Lexicon: /v1/tracks returns the whole library, which + is far too expensive for a request handler. + """ + try: + with get_db() as conn: + rows = conn.execute( + "SELECT key, value FROM app_config WHERE key IN (?, ?)", + ("lexicon_coverage", "lexicon_coverage_checked_at"), + ).fetchall() + cfg = {r["key"]: r["value"] for r in rows} + + raw = cfg.get("lexicon_coverage") + if not raw: + return { + "available": False, + "reason": "no coverage sample recorded yet — the worker collects " + "this hourly and skips while the Mac is asleep", + "checked_at": None, + } + try: + summary = json.loads(raw) + except (json.JSONDecodeError, TypeError): + return {"available": False, "reason": "stored coverage sample is unreadable", + "checked_at": cfg.get("lexicon_coverage_checked_at")} + + return { + "available": True, + "checked_at": cfg.get("lexicon_coverage_checked_at"), + **summary, + } + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + @router.get("/protected") async def protected_tracks(): try: diff --git a/sync-api/routes/tracks.py b/sync-api/routes/tracks.py index 3122edd..b13cf80 100644 --- a/sync-api/routes/tracks.py +++ b/sync-api/routes/tracks.py @@ -4,10 +4,54 @@ from typing import Optional from db import get_db -from models import TrackOut, TrackUpdate, TrackListResponse, ParityResponse +from error_categories import ERROR_CATEGORIES, canonical_category, categorize_error +from models import ( + BulkRetryRequest, + TrackOut, + TrackUpdate, + TrackListResponse, + ParityResponse, +) router = APIRouter(prefix="/api", tags=["tracks"]) +# Putting a track back at the head of the pipeline. Shared by the single-track and +# bulk retry paths so they can never reset different sets of columns. +_RETRY_RESET_SQL = """ + UPDATE tracks SET + pipeline_stage = 'new', + pipeline_error = NULL, + match_status = 'pending', + download_status = 'pending', + download_error = NULL, + download_attempts = 0, + verify_status = 'pending', + lexicon_status = 'pending', + updated_at = datetime('now') + WHERE id = ? +""" + + +def _clear_retry_blockers(conn, track_ids: list[int]) -> None: + """Remove the rows that would otherwise make a retry a no-op. + + soulseek_fallback.already_attempted() treats ANY fallback_attempts row -- + including a finalised, failed one -- as "we tried this already" and refuses to + re-queue the track. So resetting the pipeline columns alone produces a track + that marches straight back to the same error without ever re-contacting + Soulseek. Clearing the attempt history is what makes a retry mean anything. + """ + if not track_ids: + return + conn.executemany( + "DELETE FROM fallback_attempts WHERE track_id = ?", + [(tid,) for tid in track_ids], + ) + conn.executemany( + "DELETE FROM source_attempts WHERE track_id = ?", + [(tid,) for tid in track_ids], + ) + def row_to_track(row) -> dict: d = dict(row) @@ -34,6 +78,10 @@ async def list_tracks( verify_status: Optional[str] = Query(None, description="Filter by verify_status"), lexicon_status: Optional[str] = Query(None, description="Filter by lexicon_status"), search: Optional[str] = Query(None, description="Search title/artist/album"), + month: Optional[str] = Query( + None, pattern=r"^\d{4}-\d{2}$", + description="Filter by Spotify added month, YYYY-MM (dashboard drill-down)", + ), playlist_id: Optional[int] = Query(None, description="Filter by playlist"), sort_by: Optional[str] = Query(None, description="Column to sort by"), sort_dir: Optional[str] = Query("desc", description="Sort direction: asc or desc"), @@ -64,6 +112,14 @@ async def list_tracks( conditions.append("(t.title LIKE ? OR t.artist LIKE ? OR t.album LIKE ?)") like = f"%{search}%" params.extend([like, like, like]) + if month: + # Half-open range rather than substr(spotify_added_at, 1, 7) = ?, so + # the index on spotify_added_at can actually be used. The regex on the + # query param is what makes this string arithmetic safe. + year, mon = int(month[:4]), int(month[5:7]) + next_month = f"{year + 1:04d}-01" if mon == 12 else f"{year:04d}-{mon + 1:02d}" + conditions.append("t.spotify_added_at >= ? AND t.spotify_added_at < ?") + params.extend([f"{month}-01", f"{next_month}-01"]) join_clause = "" if playlist_id is not None: @@ -150,40 +206,10 @@ async def get_error_tracks(): ORDER BY title""" ).fetchall() - categories = { - "not_lossless": [], - "no_tidal_match": [], - "download_failed": [], - "lexicon_sync_failed": [], - "fingerprint_mismatch": [], - "other": [], - } + categories: dict[str, list] = {key: [] for key in ERROR_CATEGORIES} for r in errors: t = row_to_track(r) - err = (t.get("pipeline_error") or "").lower() - verify_status = t.get("verify_status") - verify_codec = (t.get("verify_codec") or "").lower() - - # Check verify_status and codec first for accurate not_lossless detection - # (Bug #31: lossy tracks were misclassified as download_failed) - if verify_codec in ("aac", "mp3"): - categories["not_lossless"].append(t) - elif verify_status == "fail" and "not lossless" in err: - categories["not_lossless"].append(t) - elif "not lossless" in err or "aac" in err or "mp3" in err: - categories["not_lossless"].append(t) - elif "no tidal match" in err or "no match" in err or "not found on tidal" in err or "permanently unavailable" in err: - categories["no_tidal_match"].append(t) - elif "geo-restricted" in err or "unavailable on tidal" in err: - categories["download_failed"].append(t) - elif "download failed" in err or "download error" in err: - categories["download_failed"].append(t) - elif "lexicon" in err: - categories["lexicon_sync_failed"].append(t) - elif "fingerprint" in err or "mismatched" in err: - categories["fingerprint_mismatch"].append(t) - else: - categories["other"].append(t) + categories[categorize_error(t)].append(t) return { "categories": categories, @@ -195,22 +221,131 @@ async def get_error_tracks(): raise HTTPException(status_code=500, detail=str(e)) +@router.get("/tracks/errors/summary") +async def get_error_summary(): + """Counts only, for the nav badge. + + The layout polls for a badge number every 30s; it used to call + /tracks/errors, which returns every errored track in full. + """ + try: + with get_db() as conn: + rows = conn.execute( + """SELECT pipeline_error, verify_status, verify_codec + FROM tracks WHERE pipeline_stage = 'error'""" + ).fetchall() + ignored = conn.execute( + "SELECT COUNT(*) FROM tracks WHERE pipeline_stage = 'ignored'" + ).fetchone()[0] + + by_category = {key: 0 for key in ERROR_CATEGORIES} + for r in rows: + by_category[categorize_error(dict(r))] += 1 + + return { + "total_errors": len(rows), + "total_ignored": ignored, + "by_category": by_category, + } + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + @router.post("/tracks/bulk-ignore") async def bulk_ignore_tracks(track_ids: list[int]): """Ignore multiple tracks at once.""" try: with get_db() as conn: - for track_id in track_ids: - conn.execute( - """UPDATE tracks SET pipeline_stage = 'ignored', is_protected = 1, - updated_at = datetime('now') WHERE id = ?""", - (track_id,), + conn.executemany( + """UPDATE tracks SET pipeline_stage = 'ignored', is_protected = 1, + updated_at = datetime('now') WHERE id = ?""", + [(tid,) for tid in track_ids], + ) + # One summary row, not one per track: bulk-ignoring a category used to + # insert thousands of activity rows and bury the dashboard feed. + conn.execute( + "INSERT INTO activity_log (event_type, message, details) VALUES (?, ?, ?)", + ( + "pipeline_bulk_ignore", + f"{len(track_ids)} track(s) bulk-ignored by user", + json.dumps({"track_ids": track_ids[:500], "count": len(track_ids)}), + ), + ) + return {"status": "ok", "count": len(track_ids)} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +@router.post("/tracks/bulk-retry") +async def bulk_retry_tracks(payload: BulkRetryRequest): + """Re-enter the pipeline for many tracks at once. + + Accepts either explicit ids or a category name. Category mode resolves the + membership SERVER-SIDE using the same classifier the Errors page renders with, + so "Retry All 47" retries exactly those 47 -- and the client never has to POST + thousands of ids. + + Protected/ignored tracks are skipped: ignoring something is a deliberate user + decision and a bulk retry should not silently undo it. + """ + try: + with get_db() as conn: + if payload.track_ids: + rows = conn.execute( + f"""SELECT * FROM tracks + WHERE id IN ({','.join('?' * len(payload.track_ids))})""", + payload.track_ids, + ).fetchall() + elif payload.category: + rows = conn.execute( + "SELECT * FROM tracks WHERE pipeline_stage = 'error'" + ).fetchall() + else: + raise HTTPException( + status_code=400, detail="Provide either track_ids or category" ) + + wanted = canonical_category(payload.category) if payload.category else None + if wanted and wanted not in ERROR_CATEGORIES: + raise HTTPException( + status_code=400, + detail=f"Unknown category '{payload.category}'. " + f"Expected one of: {', '.join(ERROR_CATEGORIES)}", + ) + + eligible, skipped = [], 0 + for r in rows: + t = row_to_track(r) + if t.get("is_protected") or t.get("pipeline_stage") == "ignored": + skipped += 1 + continue + if wanted and categorize_error(t) != wanted: + continue + eligible.append(t["id"]) + + eligible = eligible[: payload.limit] + if eligible: + conn.executemany(_RETRY_RESET_SQL, [(tid,) for tid in eligible]) + _clear_retry_blockers(conn, eligible) conn.execute( - "INSERT INTO activity_log (event_type, track_id, message) VALUES (?, ?, ?)", - ("track_ignored", track_id, f"Track {track_id} bulk-ignored by user"), + "INSERT INTO activity_log (event_type, message, details) VALUES (?, ?, ?)", + ( + "pipeline_bulk_retry", + f"{len(eligible)} track(s) re-entered the pipeline" + + (f" (category: {wanted})" if wanted else ""), + json.dumps( + { + "track_ids": eligible[:500], + "count": len(eligible), + "category": wanted, + } + ), + ), ) - return {"status": "ok", "count": len(track_ids)} + + return {"status": "ok", "count": len(eligible), "skipped": skipped} + except HTTPException: + raise except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @@ -355,20 +490,8 @@ async def retry_track(track_id: int): if not row: raise HTTPException(status_code=404, detail="Track not found") - conn.execute( - """UPDATE tracks SET - pipeline_stage = 'new', - pipeline_error = NULL, - match_status = 'pending', - download_status = 'pending', - download_error = NULL, - download_attempts = 0, - verify_status = 'pending', - lexicon_status = 'pending', - updated_at = datetime('now') - WHERE id = ?""", - (track_id,), - ) + conn.execute(_RETRY_RESET_SQL, (track_id,)) + _clear_retry_blockers(conn, [track_id]) conn.execute( "INSERT INTO activity_log (event_type, track_id, message) VALUES (?, ?, ?)", ("pipeline_retry", track_id, f"Track {track_id} re-entered pipeline"), diff --git a/sync-api/tests/test_bulk_retry.py b/sync-api/tests/test_bulk_retry.py new file mode 100644 index 0000000..59fcd60 --- /dev/null +++ b/sync-api/tests/test_bulk_retry.py @@ -0,0 +1,259 @@ +"""Tests for bulk retry and the shared error classifier. + +The properties worth pinning here are the ones that fail SILENTLY: + + * A retry that leaves fallback_attempts in place is a no-op. The track resets, + marches back down the pipeline, and soulseek_fallback.already_attempted() + refuses to re-queue it -- so it lands on the identical error and the user sees + a "retry" button that demonstrably does nothing. + * If the endpoint that COUNTS a category and the endpoint that RETRIES it + classify differently, "Retry All 47" acts on some other number of tracks. + * Bulk operations that log per-track bury the dashboard activity feed. + +Self-contained: temp SQLite file, no network. +""" + +import asyncio +import json +import os +import sqlite3 +import sys +import tempfile +import unittest + +SYNC_API_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if SYNC_API_DIR not in sys.path: + sys.path.insert(0, SYNC_API_DIR) + +# mkdtemp + join, not mktemp: mktemp returns a name without creating anything, so +# the file it names can be created by someone else between the call and our open +# (CodeQL py/insecure-temporary-file). The directory here is created atomically with +# owner-only permissions, and we own every path inside it. +_DB = os.path.join(tempfile.mkdtemp(prefix="waxflow-test-"), "sync.db") +os.environ.setdefault("SLS_DB_PATH", _DB) + +import db as db_mod # noqa: E402 +from error_categories import ERROR_CATEGORIES, canonical_category, categorize_error # noqa: E402 +from models import BulkRetryRequest # noqa: E402 +from routes import tracks as tracks_mod # noqa: E402 + +# db.py captures SLS_DB_PATH into a module global at import time, so whichever test +# module imports it FIRST decides the path for the whole pytest run. Setting the env +# var is therefore not enough -- point db.DB_PATH at our file per-test and put it +# back afterwards, so this module neither depends on collection order nor breaks the +# other test modules that make the same assumption. + + +def _seed(path: str) -> None: + conn = sqlite3.connect(path) + conn.executescript( + """ + CREATE TABLE tracks ( + id INTEGER PRIMARY KEY, + spotify_id TEXT, + title TEXT, + artist TEXT, + album TEXT, + duration_ms INTEGER, + spotify_added_at TEXT, + pipeline_stage TEXT, + pipeline_error TEXT, + match_status TEXT, + download_status TEXT, + download_error TEXT, + download_attempts INTEGER DEFAULT 0, + verify_status TEXT, + verify_codec TEXT, + lexicon_status TEXT, + is_protected INTEGER DEFAULT 0, + updated_at TEXT + ); + CREATE TABLE fallback_attempts ( + id INTEGER PRIMARY KEY, + track_id INTEGER, + source TEXT, + status TEXT, + attempted_at TEXT DEFAULT (datetime('now')) + ); + CREATE TABLE source_attempts ( + id INTEGER PRIMARY KEY, + track_id INTEGER, + source TEXT, + status TEXT, + attempted_at TEXT DEFAULT (datetime('now')) + ); + CREATE TABLE activity_log ( + id INTEGER PRIMARY KEY, + event_type TEXT, + track_id INTEGER, + message TEXT, + details TEXT, + created_at TEXT DEFAULT (datetime('now')) + ); + """ + ) + rows = [ + # id, title, error, verify_status, verify_codec, stage, protected + (1, "Lossy One", "verify failed: not lossless", "fail", "aac", "error", 0), + (2, "Lossy Two", "download ok", None, "mp3", "error", 0), + (3, "Nowhere", "No Tidal match found", None, None, "error", 0), + (4, "Nowhere Two", "permanently unavailable", None, None, "error", 0), + (5, "Broke", "download failed after 3 attempts", None, None, "error", 0), + (6, "Geo", "geo-restricted in your region", None, None, "error", 0), + (7, "Lex", "lexicon import returned 0 tracks", None, None, "error", 0), + (8, "Edit", "fingerprint mismatch: duration differs", None, None, "error", 0), + (9, "Weird", "something nobody predicted", None, None, "error", 0), + (10, "Protected", "No Tidal match found", None, None, "error", 1), + (11, "Ignored", "No Tidal match found", None, None, "ignored", 1), + ] + for tid, title, err, vs, vc, stage, prot in rows: + conn.execute( + """INSERT INTO tracks (id, title, artist, pipeline_stage, pipeline_error, + verify_status, verify_codec, match_status, + download_status, download_attempts, is_protected) + VALUES (?, ?, 'A', ?, ?, ?, ?, 'failed', 'failed', 3, ?)""", + (tid, title, stage, err, vs, vc, prot), + ) + # Every track carries a spent Soulseek attempt -- the thing that makes a + # naive retry a no-op. + conn.execute( + "INSERT INTO fallback_attempts (track_id, source, status) VALUES (?, 'soulseek', 'failed')", + (tid,), + ) + conn.commit() + conn.close() + + +class ClassifierTest(unittest.TestCase): + def test_every_bucket_is_reachable_and_declared(self): + cases = { + "not_lossless": {"verify_codec": "aac", "pipeline_error": ""}, + "no_tidal_match": {"pipeline_error": "No Tidal match found"}, + "download_failed": {"pipeline_error": "download failed after 3 attempts"}, + "lexicon_sync_failed": {"pipeline_error": "lexicon import returned 0"}, + "wrong_version": {"pipeline_error": "fingerprint mismatch"}, + "other": {"pipeline_error": "something nobody predicted"}, + } + for expected, track in cases.items(): + self.assertEqual(categorize_error(track), expected) + self.assertIn(expected, ERROR_CATEGORIES) + # Nothing declared is unreachable. + self.assertEqual(set(cases), set(ERROR_CATEGORIES)) + + def test_codec_beats_error_text(self): + # A lossy file whose error text mentions downloading must still land in + # not_lossless, or "Retry All" on the wrong bucket re-downloads it forever. + self.assertEqual( + categorize_error({"verify_codec": "mp3", "pipeline_error": "download failed"}), + "not_lossless", + ) + + def test_legacy_category_name_still_resolves(self): + self.assertEqual(canonical_category("fingerprint_mismatch"), "wrong_version") + self.assertEqual(canonical_category("no_tidal_match"), "no_tidal_match") + + def test_missing_fields_do_not_raise(self): + self.assertEqual(categorize_error({}), "other") + self.assertEqual(categorize_error({"pipeline_error": None}), "other") + + +class BulkRetryTest(unittest.TestCase): + def setUp(self): + self._saved_db_path = db_mod.DB_PATH + db_mod.DB_PATH = _DB + self.addCleanup(lambda: setattr(db_mod, "DB_PATH", self._saved_db_path)) + if os.path.exists(_DB): + os.remove(_DB) + _seed(_DB) + + def _rows(self, sql, args=()): + conn = sqlite3.connect(_DB) + conn.row_factory = sqlite3.Row + try: + return [dict(r) for r in conn.execute(sql, args).fetchall()] + finally: + conn.close() + + def test_category_retry_touches_only_that_category(self): + result = asyncio.run( + tracks_mod.bulk_retry_tracks(BulkRetryRequest(category="no_tidal_match")) + ) + # Tracks 3 and 4 qualify. Track 10 is in the error set but protected, so it + # is actively skipped; track 11 is already 'ignored' and so never enters the + # candidate set at all -- which is why skipped is 1 rather than 2. + self.assertEqual(result["count"], 2) + self.assertEqual(result["skipped"], 1) + + reset = {r["id"] for r in self._rows( + "SELECT id FROM tracks WHERE pipeline_stage = 'new'")} + self.assertEqual(reset, {3, 4}) + + def test_retry_clears_the_blocker_that_made_it_a_noop(self): + asyncio.run(tracks_mod.bulk_retry_tracks(BulkRetryRequest(track_ids=[3]))) + left = self._rows("SELECT * FROM fallback_attempts WHERE track_id = 3") + self.assertEqual(left, [], "stale fallback_attempts row makes the retry inert") + # Untouched tracks keep their history. + self.assertEqual(len(self._rows("SELECT * FROM fallback_attempts WHERE track_id = 5")), 1) + + def test_retry_resets_the_full_pipeline_state(self): + asyncio.run(tracks_mod.bulk_retry_tracks(BulkRetryRequest(track_ids=[5]))) + t = self._rows("SELECT * FROM tracks WHERE id = 5")[0] + self.assertEqual(t["pipeline_stage"], "new") + self.assertIsNone(t["pipeline_error"]) + self.assertEqual(t["download_attempts"], 0) + self.assertEqual(t["match_status"], "pending") + self.assertEqual(t["verify_status"], "pending") + + def test_protected_and_ignored_are_never_resurrected(self): + asyncio.run(tracks_mod.bulk_retry_tracks(BulkRetryRequest(track_ids=[10, 11]))) + stages = {r["id"]: r["pipeline_stage"] for r in + self._rows("SELECT id, pipeline_stage FROM tracks WHERE id IN (10, 11)")} + self.assertEqual(stages[10], "error") + self.assertEqual(stages[11], "ignored") + + def test_logs_one_summary_row_not_one_per_track(self): + asyncio.run(tracks_mod.bulk_retry_tracks(BulkRetryRequest(category="not_lossless"))) + rows = self._rows("SELECT * FROM activity_log WHERE event_type = 'pipeline_bulk_retry'") + self.assertEqual(len(rows), 1) + details = json.loads(rows[0]["details"]) + self.assertEqual(details["count"], 2) + self.assertEqual(details["category"], "not_lossless") + + def test_legacy_category_name_is_accepted(self): + result = asyncio.run( + tracks_mod.bulk_retry_tracks(BulkRetryRequest(category="fingerprint_mismatch")) + ) + self.assertEqual(result["count"], 1) # track 8 + + def test_unknown_category_is_rejected_not_silently_empty(self): + with self.assertRaises(Exception) as ctx: + asyncio.run(tracks_mod.bulk_retry_tracks(BulkRetryRequest(category="nonsense"))) + self.assertIn("Unknown category", str(ctx.exception)) + + def test_neither_ids_nor_category_is_rejected(self): + with self.assertRaises(Exception) as ctx: + asyncio.run(tracks_mod.bulk_retry_tracks(BulkRetryRequest())) + self.assertIn("track_ids or category", str(ctx.exception)) + + def test_limit_caps_the_blast_radius(self): + result = asyncio.run( + tracks_mod.bulk_retry_tracks(BulkRetryRequest(category="no_tidal_match", limit=1)) + ) + self.assertEqual(result["count"], 1) + + def test_summary_counts_match_what_retry_would_act_on(self): + summary = asyncio.run(tracks_mod.get_error_summary()) + for category, expected in summary["by_category"].items(): + if expected == 0: + continue + asyncio.run(tracks_mod.bulk_retry_tracks(BulkRetryRequest(category=category))) + acted = len(self._rows( + "SELECT id FROM tracks WHERE pipeline_stage = 'new'")) + # Protected/ignored are counted by the summary but skipped by retry; + # everything else must line up exactly. + self.assertLessEqual(acted, expected) + self.setUp() + + +if __name__ == "__main__": + unittest.main() diff --git a/sync-api/tests/test_status.py b/sync-api/tests/test_status.py index c425d4b..b4e43f0 100644 --- a/sync-api/tests/test_status.py +++ b/sync-api/tests/test_status.py @@ -298,7 +298,16 @@ def test_renders_when_signals_missing(self): class TestEndpoints(unittest.TestCase): @classmethod def setUpClass(cls): - _seed_full(_ENDPOINT_DB) # the file SLS_DB_PATH points at + _seed_full(_ENDPOINT_DB) + # Setting SLS_DB_PATH above only works if THIS module is the first to import + # db.py, which captures the env var into a module global. That made the test + # silently dependent on pytest's alphabetical collection order -- adding any + # test module sorting earlier pointed the handlers at someone else's DB. + # Pin it explicitly instead. + import db as db_mod + cls._saved_db_path = db_mod.DB_PATH + db_mod.DB_PATH = _ENDPOINT_DB + from fastapi import FastAPI from fastapi.testclient import TestClient # Mount ONLY the status router (proves it stands alone and keeps the test @@ -307,6 +316,11 @@ def setUpClass(cls): app.include_router(status_mod.router) cls.client = TestClient(app) + @classmethod + def tearDownClass(cls): + import db as db_mod + db_mod.DB_PATH = cls._saved_db_path + def test_status_json_ok(self): r = self.client.get("/api/status.json") self.assertEqual(r.status_code, 200) diff --git a/sync-web/app/api.ts b/sync-web/app/api.ts index 7d62d6f..7cd8224 100644 --- a/sync-web/app/api.ts +++ b/sync-web/app/api.ts @@ -1,19 +1,149 @@ +/** + * API client. + * + * WHY THIS IS MORE THAN A fetch() WRAPPER + * The previous version threw `new Error("API error: " + res.status)` and + * discarded the response body. FastAPI puts the actual reason in `detail`, so + * every validation failure arrived as an opaque number and callers uniformly + * did `catch { setRows([]) }` — rendering an empty page that looked like a + * design decision rather than a fault. + * + * That is exactly how the Missing Tracks page stayed broken: it asked for + * `per_page=500` against an endpoint capped at 200, got a 422 explaining + * precisely that, and showed a clean empty table instead. + * + * So: keep the status, keep the detail, and make failures inspectable. + */ + const API_BASE = '/api' -export async function apiFetch(path: string, options?: RequestInit): Promise { - const res = await fetch(`${API_BASE}${path}`, { - headers: { 'Content-Type': 'application/json', ...options?.headers }, - ...options, - }) - if (!res.ok) throw new Error(`API error: ${res.status}`) - return res.json() +/** An HTTP-level failure, carrying enough context to actually diagnose it. */ +export class ApiError extends Error { + readonly status: number + readonly detail?: string + readonly path: string + + constructor(status: number, path: string, detail?: string) { + super(detail ? `${status} ${path}: ${detail}` : `${status} ${path}`) + this.name = 'ApiError' + this.status = status + this.detail = detail + this.path = path + } + + /** 4xx means we sent something wrong; retrying sends the same wrong thing. */ + get isClientError(): boolean { + return this.status >= 400 && this.status < 500 + } +} + +/** Pull FastAPI's `detail` out of an error body, whatever shape it arrived in. */ +async function extractDetail(res: Response): Promise { + try { + const body = await res.json() + const d = body?.detail + if (typeof d === 'string') return d + // 422 from pydantic is a list of {loc, msg, type}. + if (Array.isArray(d)) { + return d + .map((e) => { + const loc = Array.isArray(e?.loc) ? e.loc.filter((p: unknown) => p !== 'query').join('.') : '' + return loc ? `${loc}: ${e?.msg}` : e?.msg + }) + .filter(Boolean) + .join('; ') + } + if (d != null) return JSON.stringify(d) + } catch { + /* non-JSON body; fall through */ + } + return undefined +} + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)) + +export interface ApiOptions extends RequestInit { + /** Retries for transport faults and 5xx only. Never applied to 4xx. */ + retries?: number +} + +export async function apiFetch(path: string, options?: ApiOptions): Promise { + const { retries = 2, ...init } = options ?? {} + let lastError: unknown + + for (let attempt = 0; attempt <= retries; attempt++) { + let res: Response + try { + res = await fetch(`${API_BASE}${path}`, { + ...init, + headers: { 'Content-Type': 'application/json', ...init.headers }, + }) + } catch (err) { + // Transport-level failure (offline, DNS, connection reset). + lastError = err + if (attempt < retries) { + await sleep(250 * 2 ** attempt) + continue + } + throw err + } + + if (res.ok) return res.json() as Promise + + const error = new ApiError(res.status, path, await extractDetail(res)) + + // A 4xx is our fault and is deterministic — surface it immediately. + if (error.isClientError) throw error + + lastError = error + if (attempt < retries) { + await sleep(250 * 2 ** attempt) + continue + } + throw error + } + + throw lastError } export async function apiUpload(path: string, formData: FormData): Promise { - const res = await fetch(`${API_BASE}${path}`, { - method: 'POST', - body: formData, - }) - if (!res.ok) throw new Error(`Upload error: ${res.status}`) - return res.json() + // No Content-Type header: the browser must set the multipart boundary itself. + const res = await fetch(`${API_BASE}${path}`, { method: 'POST', body: formData }) + if (!res.ok) throw new ApiError(res.status, path, await extractDetail(res)) + return res.json() as Promise +} + +/** Shape shared by every paginated list endpoint. */ +interface Paged { + tracks?: T[] + total?: number +} + +/** + * Fetch every page of a list endpoint. + * + * `/tracks` caps `per_page` at 200, but several pages want the whole result set + * so they can sort and filter client-side. Asking for more than the cap is a 422, + * and asking for exactly the cap silently truncates — both were live bugs. Page + * through instead, with a hard stop so a misreported `total` cannot spin forever. + */ +export async function apiFetchAll( + path: string, + { perPage = 200, maxPages = 25 }: { perPage?: number; maxPages?: number } = {}, +): Promise<{ items: T[]; total: number; truncated: boolean }> { + const sep = path.includes('?') ? '&' : '?' + const items: T[] = [] + let total = 0 + + for (let page = 1; page <= maxPages; page++) { + const result = await apiFetch>(`${path}${sep}page=${page}&per_page=${perPage}`) + const batch = result.tracks ?? [] + total = result.total ?? items.length + batch.length + items.push(...batch) + if (batch.length < perPage || items.length >= total) { + return { items, total, truncated: false } + } + } + + return { items, total, truncated: true } } diff --git a/sync-web/app/errors/page.tsx b/sync-web/app/errors/page.tsx index 68fecf0..3b6d872 100644 --- a/sync-web/app/errors/page.tsx +++ b/sync-web/app/errors/page.tsx @@ -65,13 +65,13 @@ const CATEGORIES: CategoryInfo[] = [ description: 'Tracks with files that could not be added to Lexicon', }, { - key: 'fingerprint_mismatch', - label: 'Fingerprint Mismatch', + key: 'wrong_version', + label: 'Wrong Version', color: 'orange', borderColor: 'border-orange-500/30', bgColor: 'bg-orange-500/10', textColor: 'text-orange-400', - description: 'Tracks where the downloaded file might be the wrong version', + description: 'The file is a different edit of the right song (radio vs extended, live, remix)', }, { key: 'other', @@ -205,7 +205,7 @@ export default function ErrorsPage() { const toIgnore = selected.length > 0 ? selected : tracks if (toIgnore.length === 0) return - setBulkLoading(categoryKey) + setBulkLoading(`ignore:${categoryKey}`) try { await apiFetch('/tracks/bulk-ignore', { method: 'POST', @@ -214,8 +214,46 @@ export default function ErrorsPage() { setToast(`${toIgnore.length} track${toIgnore.length !== 1 ? 's' : ''} ignored`) setSelectedTracks(new Set()) fetchData() - } catch { - setToast('Bulk ignore failed') + } catch (err) { + setToast(err instanceof Error ? `Bulk ignore failed: ${err.message}` : 'Bulk ignore failed') + } finally { + setBulkLoading(null) + } + } + + // Retrying a whole category sends every one of those tracks back through a + // pipeline that picks up work every 10 seconds, so it is worth a confirmation + // once the count gets large. When nothing is explicitly selected we send the + // CATEGORY rather than thousands of ids and let the server resolve membership + // with the same classifier that produced the number on the button. + const handleBulkRetry = async (categoryKey: string) => { + const tracks = filterTracks(data?.categories[categoryKey] || []) + const selected = tracks.filter(t => selectedTracks.has(t.id)) + const count = selected.length > 0 ? selected.length : tracks.length + if (count === 0) return + + if (count > 100 && !confirm( + `Retry ${count} tracks? They will all re-enter the pipeline, which processes ` + + `continuously in the background.` + )) return + + setBulkLoading(`retry:${categoryKey}`) + try { + const body = selected.length > 0 + ? { track_ids: selected.map(t => t.id) } + : { category: categoryKey } + const result = await apiFetch<{ count: number; skipped: number }>('/tracks/bulk-retry', { + method: 'POST', + body: JSON.stringify(body), + }) + setToast( + `${result.count} track${result.count !== 1 ? 's' : ''} queued for retry` + + (result.skipped ? ` (${result.skipped} protected, skipped)` : '') + ) + setSelectedTracks(new Set()) + fetchData() + } catch (err) { + setToast(err instanceof Error ? `Bulk retry failed: ${err.message}` : 'Bulk retry failed') } finally { setBulkLoading(null) } @@ -386,12 +424,23 @@ export default function ErrorsPage() { <> {/* Bulk actions bar */}
+ + )}