diff --git a/.env.example b/.env.example index 7d7aaf7..3c853e0 100644 --- a/.env.example +++ b/.env.example @@ -17,5 +17,23 @@ NEXT_PUBLIC_API_URL=http://localhost:8402 # CORS origins for API CORS_ORIGINS=http://localhost:8400 -# Music library path (inside container) -MUSIC_LIBRARY_PATH=/music + +# --- Lexicon host paths ------------------------------------------------------- +# Where the machine running Lexicon sees the music tree. The worker rewrites its +# own container path to this before telling Lexicon where a file is. +# +# Default (seeded in the DB) is the SMB mount, which needs no replication: +# LEXICON_LIBRARY_PATH=/Volumes/music/Database +# LEXICON_INPUT_PATH=/Volumes/music/Input +# +# If you ALSO export to Engine DJ, prefer LOCAL paths on the Lexicon host and run +# scripts/sync-nas-to-mac.sh to replicate. See "Local paths and Engine DJ" in the +# README. Contains a username, so it must be set per install: +# LEXICON_LIBRARY_PATH=/Users/you/Music/Database +# LEXICON_INPUT_PATH=/Users/you/Music/Input + +# --- Library root inside the container ---------------------------------------- +# The worker writes finished audio here and index_library scans it. Point it at a +# SUBDIRECTORY of the bind mount, not the mount root, so the share root stays +# clean and Plex path translation keeps working. +MUSIC_LIBRARY_PATH=/music/Database diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c5a923..b13ec90 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,88 @@ # Changelog +## 2.11.0 — Path contract, one-way replication, and a pile of real bugs + +Everything here was found by *running* the system during a live incident, not by +reading it. Several are bugs any deployment would hit. + +### Fixed — bugs that affect every install +- **`MUSIC_LIBRARY_PATH` was hardcoded in `docker-compose.yml`.** Both services + pinned `/music`, so setting it in `.env` silently did nothing. Now + `${MUSIC_LIBRARY_PATH:-/music/Database}`. Pointing the library root at a + SUBDIRECTORY of the bind mount keeps the share root clean and leaves Plex path + translation intact — remapping the mount instead would break every existing + `file_path` row. +- **The update banner was inverted.** `admin.py` compared versions as strings, so + `"2.9.0" > "2.10.1"` was `True` and no update was ever offered across a + major.minor boundary. Replaced with a numeric-tuple compare. +- **The API reported the wrong version.** `main.py` hardcoded `2.1.0` while + `VERSION` said `2.10.1`. It now reads `/app/VERSION`, baked in at build time. +- **`busy_timeout` contradicted the connect timeout.** 5 s against `timeout=30`, + and `busy_timeout` is what governs — so `PATCH /api/settings` still returned + `{"detail":"database is locked"}` whenever the worker was mid-index. Both 30 s. +- **`deep-repair.sh` could never alert anyone.** It computed a repair verdict and + then dispatched to a commented-out example. Wired to `WAXFLOW_ALERT_WEBHOOK`. +- **CodeQL scanned Python only**; `sync-web`'s TypeScript went unanalysed. (Held + back from this PR — needs a `workflow`-scoped token.) + +### Added — one-way NAS -> Mac replication (optional) +`scripts/sync-nas-to-mac.sh` + LaunchAgent. Pull-only, so conflict copies are +structurally impossible. Hybrid transport: change detection over SSH (~0.9 s, +the NAS walks its own disk) with an SMB transfer, versus 5 m 46 s for a full SMB +scan; 6 h reconcile as the safety net. `tasks/sync_gate.py` holds each import +until the file has landed locally, and **fails open** on every degenerate case — +a gate that can deadlock the pipeline is worse than the lag it prevents. + +### Added — tools +- `scripts/merge-duplicate-lexicon-rows.py` — Engine DJ's `Track.path` is UNIQUE, + so Lexicon rows beyond the distinct-file count can NEVER sync and can leave a + part-applied sync failing with `FOREIGN KEY constraint failed`. Migrates + playlist memberships onto the surviving row *before* deleting; a plain delete + would have destroyed 849 memberships on the library this was written against. +- `scripts/consolidate-share-root.py` — moves stray artist folders from a share + root into the library root. Non-destructive on collisions. +- `scripts/dedupe-report.py` — read-only duplicate/quality analysis. Reads remix + descriptors from the parent folder as well as the filename, so different mixes + are not reported as duplicates. + +### Fixed — macOS agent robustness +- `ensure-music-mount.sh` treated **any** failed `ls` as a stale handle and + unmounted. From a launchd agent the share cannot be remounted (`mount volume` + has no keychain access; `mount_smbfs` returns `Authentication error`), so it + destroyed a working mount that only a human in Finder could restore. It now + never unmounts by default, and distinguishes macOS TCC's `Operation not + permitted` — which means the mount is fine and *this process* is denied — from + a real stale handle. +- The share is addressed by IP/hostname, **not** a Bonjour service-instance name. + `NAME._smb._tcp.local` resolves only via service discovery; when that goes + stale, mounts hang forever instead of failing. +- `osascript mount volume` is now watchdog-wrapped. It blocks indefinitely + waiting on a credential dialog that never appears under launchd, which wedged + both agents and stopped replication entirely. + +### Changed +- Host-specific values moved out of the scripts into `~/.waxflow/waxflow.conf`. +- `lexicon_library_path` / `lexicon_input_path` seed to the SMB default and can be + set via `LEXICON_LIBRARY_PATH` / `LEXICON_INPUT_PATH` — no username in defaults. +- `bump-version.sh` updates every version source, not just `VERSION`. +- README gains "Local paths and Engine DJ", including the two things that cost the + most time here: never put an Engine library inside a two-way sync, and Engine + holds at most one row per file. + +### Removed +- `sync-api/services/{matcher,downloader,verifier}.py` — 463 lines referenced + nowhere; that logic lives in `sync-worker/tasks/`. +- `scripts/backup-lexicon.sh` — self-documented no-op; `backup-lexicon-db.sh` is + the real one. + +### Corrected +An earlier diagnosis held that Engine DJ refuses `/Volumes/*` locations. **It does +not.** Tested against a real Engine library: all 40 rows carrying a +`/Volumes/Macintosh HD/` prefix were present. That prefix is a symlink to `/` and +resolves fine. The missing-tracks symptom was caused by two-way sync destroying +the Engine database, not by path format. + + ## 2.10.0 — Sleep-tolerance catch-up: rescue downloaded-but-not-imported tracks Closes the last sleep/wake gap that stranded freshly-downloaded tracks with a real diff --git a/CLAUDE.md b/CLAUDE.md index 534ed18..6873f9a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -35,10 +35,15 @@ docker compose up -d ``` ## Testing -No test suite exists yet. This is a priority backlog item. When tests are added: +22 test modules exist across `sync-worker/tests/` and `sync-api/tests/`. ```bash -python3 -m pytest tests/ -v +cd sync-worker && python3 -m pytest tests/ -v # or: python3 -m unittest discover tests +cd sync-api && python3 -m pytest tests/ -v ``` +Most worker tests import `tasks.helpers`, which pulls in `spotipy` — run them inside +the worker image (`docker exec waxflow-worker python3 -m pytest tests/`) or in a venv +with `sync-worker/requirements.txt` installed. Leaf modules with no heavy imports +(e.g. `tasks/sync_gate.py`) run anywhere. ## Coding Standards - Python 3.12, type hints encouraged @@ -66,3 +71,43 @@ Key variables (see `.env.example` or README for full list): - **Pipeline stages**: new -> matching -> downloading -> verifying -> organizing -> complete - **5-layer dedup**: ISRC file index, Lexicon DB lookup, on-disk scan, Tidal ISRC, Tidal metadata - **Scan mode vs Full mode**: scan mode only matches existing library; full mode downloads new tracks + +## The path contract (READ THIS BEFORE TOUCHING PATHS) +Rewritten 2026-08-08 after the library ended up split across two roots and Engine DJ +lost its entire database. Three rules, and they are load-bearing: + +1. **The library root is `Database/`, not the share root.** + `MUSIC_LIBRARY_PATH=/music/Database` — the worker writes there and `index_library` + scans there. The bind mount deliberately still points at the share ROOT + (`MUSIC_HOST_PATH=/volume1/music`) so Plex path translation + (`plex_music_container_prefix=/music` -> `/volume1/music`) and the ~4,300 existing + `/music/Database/...` `tracks.file_path` rows keep resolving. Do NOT "simplify" + this by remapping the mount — it would break every one of those rows. + +2. **Prefer LOCAL Lexicon-host paths over `/Volumes/*`.** + `lexicon_library_path=` (e.g. `~/Music/Database`), set via + `LEXICON_LIBRARY_PATH`. NOT because Engine DJ rejects `/Volumes/*` — it does + not; that was tested on 2026-08-09 and disproved, all 40 rows carrying a + `/Volumes/Macintosh HD/` prefix were present in Engine. Prefer local paths + because an SMB path breaks the moment the share unmounts and the file then + exists only on the NAS. Engine stores paths relative to its own folder + (`../Database//...`), so `Engine Library/` must remain a sibling of + `Database/`. + + Engine also enforces `UNIQUE (path)`: if Lexicon holds more rows than distinct + files, the surplus can never sync. See `scripts/merge-duplicate-lexicon-rows.py`. + +3. **Replication is one-way NAS -> Mac, and imports wait for it.** + `scripts/sync-nas-to-mac.sh` (launchd, 120 s) pulls `Database/` and `Input/` down; + `tasks/sync_gate.py` holds each import until the file has landed. Synology Drive is + NOT involved — two-way syncing the whole share produced 12 conflict copies of + Engine's database and jammed permanently on SoundSwitch project files. + +`scripts/repoint-lexicon-local.sh` normalises legacy rows. Note that Lexicon +CANONICALISES imported paths through the boot-volume symlink, so new imports may +reappear as `/Volumes/Macintosh HD/Users/...` — cosmetic, not a fault. The sync +agent logs an action-needed line only for genuine SMB paths. + +**Never put an Engine library inside a two-way sync.** Two-way syncing the share +containing Engine's live `m.db` produced 12 conflict copies and left the real +database with 2 tracks in it. That, not path format, is what lost the library. diff --git a/README.md b/README.md index 5c19fd1..4b5f929 100644 --- a/README.md +++ b/README.md @@ -393,6 +393,69 @@ volumes: --- +--- + +## Local paths and Engine DJ + +By default WaxFlow tells Lexicon where a file is using the **SMB mount** +(`/Volumes/music/...`). That needs no replication and works out of the box. + +If you also export from Lexicon to **Engine DJ**, there are two things worth +knowing, both learned the hard way: + +**1. Never put your Engine library inside a two-way sync.** Engine's `m.db` is a +live SQLite file. Two-way syncing the share that contains it produced 12 conflict +copies (~9.4 GB) and left the real database with 2 tracks in it. Engine stores +track paths *relative* to its own folder (`../Database//...`), so a copy +on the NAS is meaningless as well as dangerous. Keep `Engine Library/` local and +excluded from every sync. + +**2. Engine holds at most one row per file.** Its schema declares +`CONSTRAINT C_path UNIQUE (path)`. If Lexicon has more rows than distinct files — +several rows pointing at the same audio — the surplus can never sync, and a +partially-applied sync can fail with `SqliteError: FOREIGN KEY constraint failed`. +Check with: + +```sql +SELECT COUNT(*), COUNT(DISTINCT location) FROM Track; -- Lexicon's main.db +``` + +If those differ, `scripts/merge-duplicate-lexicon-rows.py` merges them safely — it +migrates playlist memberships onto the surviving row *before* deleting, because a +plain delete silently drops memberships the duplicate held. + +### Optional: local replication + +To hand Lexicon local paths instead, set `LEXICON_LIBRARY_PATH` / +`LEXICON_INPUT_PATH` to paths on the Lexicon host and run the replication agent: + +```bash +mkdir -p ~/.waxflow +cp scripts/sync-nas-to-mac.sh scripts/ensure-music-mount.sh ~/.waxflow/ +chmod +x ~/.waxflow/*.sh +printf 'WAXFLOW_SHARE_HOST=192.168.1.50 +WAXFLOW_NAS_SSH=nas +' > ~/.waxflow/waxflow.conf +sed "s|/Users/willcurran|$HOME|g" scripts/com.waxflow.sync-database.plist \ + > ~/Library/LaunchAgents/com.waxflow.sync-database.plist +launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.waxflow.sync-database.plist +``` + +It replicates **one way, NAS to Mac** — a pull-only replica cannot create conflict +copies. `tasks/sync_gate.py` then holds each import until the file has actually +landed locally, so Lexicon never imports a path that does not exist yet. + +Two macOS gotchas the scripts now handle explicitly, both of which cost real +downtime here: + +* **Full Disk Access.** A launchd agent cannot read `/Volumes/*` until you add + `/bin/bash` under System Settings → Privacy & Security → Full Disk Access. + Without it every pass fails with `Operation not permitted`, which looks exactly + like a dead mount but is not. +* **Address the NAS by IP or plain hostname, never a Bonjour *service* name.** + `NAME._smb._tcp.local` resolves only via service discovery; when that + advertisement goes stale, mounts **hang forever** rather than failing. + ## Tech Stack | Component | Technology | diff --git a/VERSION b/VERSION index 8bbb6e4..46b81d8 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.10.1 +2.11.0 diff --git a/docker-compose.yml b/docker-compose.yml index 0e927f6..1ce86e5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -21,7 +21,11 @@ services: - SPOTIFY_REDIRECT_URI=${SPOTIFY_REDIRECT_URI:-http://localhost:8402/api/spotify/callback} - TIDARR_URL=${TIDARR_URL:-http://localhost:8484} # optional: only for legacy Tidarr fallback - LEXICON_API_URL=${LEXICON_API_URL:-http://localhost:48624} - - MUSIC_LIBRARY_PATH=/music + # The bind mount above still points at the share ROOT so Plex path + # translation (plex_music_container_prefix=/music) and existing + # /music/Database/... file_path rows keep resolving. The LIBRARY root + # is a subdirectory of it -- that is what the worker writes into. + - MUSIC_LIBRARY_PATH=${MUSIC_LIBRARY_PATH:-/music/Database} - SLS_DB_PATH=/app/data/sync.db restart: unless-stopped healthcheck: @@ -54,7 +58,11 @@ services: - SPOTIFY_REDIRECT_URI=${SPOTIFY_REDIRECT_URI:-http://localhost:8402/api/spotify/callback} - TIDARR_URL=${TIDARR_URL:-http://localhost:8484} # optional: only for legacy Tidarr fallback - LEXICON_API_URL=${LEXICON_API_URL:-http://localhost:48624} - - MUSIC_LIBRARY_PATH=/music + # The bind mount above still points at the share ROOT so Plex path + # translation (plex_music_container_prefix=/music) and existing + # /music/Database/... file_path rows keep resolving. The LIBRARY root + # is a subdirectory of it -- that is what the worker writes into. + - MUSIC_LIBRARY_PATH=${MUSIC_LIBRARY_PATH:-/music/Database} - SLS_DB_PATH=/app/data/sync.db - TIDDL_PATH=/tiddl-config depends_on: diff --git a/scripts/backup-lexicon-db.sh b/scripts/backup-lexicon-db.sh index 33aabc1..ae2231d 100755 --- a/scripts/backup-lexicon-db.sh +++ b/scripts/backup-lexicon-db.sh @@ -44,11 +44,12 @@ set -euo pipefail -LEXICON_SSH="${LEXICON_SSH:-willcurran@192.168.1.116}" +# Set LEXICON_SSH to @, or "local" when running ON that Mac. +LEXICON_SSH="${LEXICON_SSH:-local}" LEXICON_DB="${LEXICON_DB:-\$HOME/Library/Application Support/lexicon/main.db}" MAC_BACKUP_DIR="${MAC_BACKUP_DIR:-\$HOME/WaxFlow-Backups/lexicon-db}" NAS_SSH="${NAS_SSH:-nas}" -NAS_BACKUP_DIR="${NAS_BACKUP_DIR:-/volume1/homes/willcurran/WaxFlow-Backups/lexicon-db}" +NAS_BACKUP_DIR="${NAS_BACKUP_DIR:-/volume1/homes/$USER/WaxFlow-Backups/lexicon-db}" KEEP="${KEEP:-14}" LOG_DIR="${LOG_DIR:-$HOME/.waxflow/logs}" SKIP_ON_HYPERBACKUP="${SKIP_ON_HYPERBACKUP:-1}" diff --git a/scripts/backup-lexicon.sh b/scripts/backup-lexicon.sh deleted file mode 100755 index 5ab060f..0000000 --- a/scripts/backup-lexicon.sh +++ /dev/null @@ -1,44 +0,0 @@ -#!/bin/bash -# Backup Lexicon DJ database before sync operations (container-side fast-path). -# -# NOTE: The Lexicon DB lives on Will's MAC (~/Library/Application Support/ -# lexicon/main.db), which this container CANNOT reach — so in normal operation -# this script finds no local DB and is a no-op. The REAL, verified, two-location -# backup is scripts/backup-lexicon-db.sh, which runs on the ops Mac (SSHes the -# Lexicon Mac for a consistent sqlite3 online backup + pushes to the NAS). Run -# THAT before any delicate library operation. This script only helps in the -# unusual case where the DB is bind-mounted into the container via -# LEXICON_DB_PATH. - -set -euo pipefail - -BACKUP_DIR="${BACKUP_DIR:-/app/data/lexicon-backups}" -TIMESTAMP=$(date '+%Y%m%d_%H%M%S') -MAX_BACKUPS=30 - -mkdir -p "$BACKUP_DIR" - -# Lexicon stores its DB on the Mac Mini — we backup via the API -# This script is a safety net for direct file backups if mounted -LEXICON_DB="${LEXICON_DB_PATH:-}" - -if [ -n "$LEXICON_DB" ] && [ -f "$LEXICON_DB" ]; then - BACKUP_FILE="${BACKUP_DIR}/lexicon_${TIMESTAMP}.db" - cp "$LEXICON_DB" "$BACKUP_FILE" - - # Verify backup integrity - sqlite3 "$BACKUP_FILE" "PRAGMA integrity_check;" > /dev/null 2>&1 - if [ $? -eq 0 ]; then - echo "Backup created: $BACKUP_FILE ($(du -h "$BACKUP_FILE" | cut -f1))" - else - echo "ERROR: Backup integrity check failed!" >&2 - rm -f "$BACKUP_FILE" - exit 1 - fi - - # Prune old backups (keep last MAX_BACKUPS) - ls -1t "${BACKUP_DIR}"/lexicon_*.db 2>/dev/null | tail -n +$((MAX_BACKUPS + 1)) | xargs -r rm -f -else - echo "No Lexicon DB path configured or file not found. Skipping file backup." - echo "Lexicon backups are handled via the API." -fi diff --git a/scripts/bump-version.sh b/scripts/bump-version.sh index d51cd38..ba01ed1 100755 --- a/scripts/bump-version.sh +++ b/scripts/bump-version.sh @@ -1,12 +1,45 @@ #!/bin/bash # Usage: ./scripts/bump-version.sh 1.2.1 -VERSION=$1 +# +# Single source of truth is the VERSION file. Everything that needs a version +# either reads it at runtime (sync-api/main.py reads /app/VERSION, baked in by the +# Dockerfile) or is updated here. Previously only VERSION was bumped, so main.py +# reported "2.1.0" while VERSION said 2.10.1, and the MusicBrainz user agent was +# stuck at 2.9. +set -euo pipefail + +VERSION="${1:-}" if [ -z "$VERSION" ]; then - echo "Usage: $0 " + echo "Usage: $0 " >&2 exit 1 fi +if ! echo "$VERSION" | grep -qE '^[0-9]+\.[0-9]+\.[0-9]+$'; then + echo "REFUSE: '$VERSION' is not semver (X.Y.Z)" >&2 + exit 1 +fi + +REPO_DIR="$(cd "$(dirname "$0")/.." && pwd)" +cd "$REPO_DIR" + +if ! grep -q "^## $VERSION" CHANGELOG.md 2>/dev/null; then + echo "WARNING: CHANGELOG.md has no '## $VERSION' section yet." >&2 + printf "Continue anyway? [y/N] " + read -r reply + case "$reply" in [yY]*) ;; *) echo "aborted"; exit 1;; esac +fi + echo "$VERSION" > VERSION -git add VERSION + +# MusicBrainz asks that the user agent identify the app version (it is how they +# rate-limit and contact you); keep the seeded default in step with the release. +MAJOR_MINOR="$(echo "$VERSION" | cut -d. -f1,2)" +sed -i.bak -E "s|('musicbrainz_user_agent', 'WaxFlow/)[0-9]+\.[0-9]+|\1$MAJOR_MINOR|" sync-api/init_db.py +rm -f sync-api/init_db.py.bak + +echo "--- files changed ---" +git --no-pager diff --stat VERSION sync-api/init_db.py + +git add VERSION sync-api/init_db.py git commit -m "Bump version to $VERSION" git tag "v$VERSION" git push && git push --tags diff --git a/scripts/com.openclaw.waxflow-lexicon-backup.plist b/scripts/com.openclaw.waxflow-lexicon-backup.plist index ecfebbd..9656cc5 100644 --- a/scripts/com.openclaw.waxflow-lexicon-backup.plist +++ b/scripts/com.openclaw.waxflow-lexicon-backup.plist @@ -22,7 +22,7 @@ ProgramArguments /bin/bash - /Users/openclaw/spotify-lexicon-sync/scripts/backup-lexicon-db.sh + /Users/openclaw/waxflow/scripts/backup-lexicon-db.sh RunAtLoad diff --git a/scripts/com.waxflow.sync-database.plist b/scripts/com.waxflow.sync-database.plist new file mode 100644 index 0000000..d15df07 --- /dev/null +++ b/scripts/com.waxflow.sync-database.plist @@ -0,0 +1,42 @@ + + + + + + Labelcom.waxflow.sync-database + ProgramArguments + + /bin/bash + /Users/willcurran/.waxflow/sync-nas-to-mac.sh + + RunAtLoad + StartInterval120 + StandardOutPath/Users/willcurran/.waxflow/sync-database.launchd.out.log + StandardErrorPath/Users/willcurran/.waxflow/sync-database.launchd.err.log + LowPriorityIO + Nice5 + + diff --git a/scripts/consolidate-share-root.py b/scripts/consolidate-share-root.py new file mode 100755 index 0000000..2079c85 --- /dev/null +++ b/scripts/consolidate-share-root.py @@ -0,0 +1,235 @@ +#!/usr/bin/env python3 +"""WaxFlow — consolidate stray artist folders from the NAS music share ROOT into Database/. + +WHY + The worker's _move_to_library() writes to MUSIC_LIBRARY_PATH == container /music, + which was bind-mounted to the share ROOT (/volume1/music) rather than the actual + library root (/volume1/music/Database). Result: 339 audio files across 189 artist + folders live at the share root, outside the library Lexicon and Engine DJ read. + + This script moves them into Database/, preserving the relative layout so the + Lexicon repoint is a pure prefix insertion: + + // + -> /Database// + + Phase 3 then re-points the container mount at /volume1/music/Database, so + tracks.file_path values of the form /music// keep resolving to the + same bytes they resolved to before. The two changes MUST ship together. + +TRANSPORT + Runs on the Lexicon host Mac over the SMB mount. It cannot run over SSH: DSM + gives the SSH user no write access to Database/ (it is owned by PlexMediaServer), + while the SMB share grants it via ACL. Source and destination are the same share, + so each move is a server-side rename — no data crosses the network. + +COLLISIONS (never destructive) + identical size -> byte-compare; if truly identical the root copy is moved to + #waxflow-quarantine/ (NOT deleted) and logged + different size -> both kept; the incoming file gets a '__fromroot' suffix and is + flagged in the manifest for the Phase 6 dedupe pass + +KEEP AT ROOT (never touched) + Database, Input, DJ Will See, Disorganized, Engine Library, Engine Library Backup, + Friends Recordings, Processing, rekordbox, SoundSwitch, Music, #recycle, @eaDir + +USAGE + ./consolidate-share-root.py # dry run, full report (default) + ./consolidate-share-root.py --apply # perform the moves + ./consolidate-share-root.py --apply --limit 20 + ./consolidate-share-root.py --prune # after --apply: remove now-empty root + # artist folders (metadata-only leftovers) +""" +from __future__ import annotations + +import argparse +import filecmp +import os +import sys +import unicodedata +from datetime import datetime, timezone + +SHARE = os.environ.get("WAXFLOW_SHARE", "/Volumes/music") +LIBRARY = "Database" +QUARANTINE = "#waxflow-quarantine" +MANIFEST_DIR = os.path.expanduser("~/WaxFlow-Backups") +# Where the Lexicon HOST sees its library. Used only to write old->new +# columns into the manifest so a later repoint can be driven from it. +MAC_MUSIC_ROOT = os.environ.get("WAXFLOW_MAC_MUSIC_ROOT", + os.path.expanduser("~/Music")) + +KEEP_AT_ROOT = { + "Database", "Input", "DJ Will See", "Disorganized", "Engine Library", + "Engine Library Backup", "Friends Recordings", "Processing", "rekordbox", + "SoundSwitch", "Music", "#recycle", "@eaDir", QUARANTINE, +} +AUDIO_EXT = {".flac", ".m4a", ".mp3", ".wav", ".aiff", ".aif", ".ogg", ".alac", ".aac"} + + +def nfc(s: str) -> str: + return unicodedata.normalize("NFC", s) + + +def is_audio(name: str) -> bool: + return os.path.splitext(name)[1].lower() in AUDIO_EXT + + +def collect(share: str): + """Yield (artist, relpath) for every audio file in a non-keep root folder.""" + for entry in sorted(os.listdir(share)): + if entry in KEEP_AT_ROOT or entry.startswith("."): + continue + folder = os.path.join(share, entry) + if not os.path.isdir(folder): + continue + for dirpath, dirnames, filenames in os.walk(folder): + dirnames[:] = [d for d in dirnames if d != "@eaDir"] + for fn in sorted(filenames): + if fn.startswith(".") or not is_audio(fn): + continue + full = os.path.join(dirpath, fn) + yield entry, os.path.relpath(full, share) + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--apply", action="store_true", help="perform moves (default: dry run)") + ap.add_argument("--prune", action="store_true", help="remove now-empty root artist folders") + ap.add_argument("--limit", type=int, default=0, help="only process the first N files") + ap.add_argument("--share", default=SHARE) + args = ap.parse_args() + + share = args.share + if not os.path.isdir(os.path.join(share, LIBRARY)): + print(f"REFUSE: {share}/{LIBRARY} not found — is the SMB share mounted?", file=sys.stderr) + return 1 + + stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + os.makedirs(MANIFEST_DIR, exist_ok=True) + manifest_path = os.path.join(MANIFEST_DIR, f"consolidate-share-root-{stamp}.tsv") + + rows, n_clean, n_ident, n_diff, moved_bytes = [], 0, 0, 0, 0 + + for artist, rel in collect(share): + src = os.path.join(share, rel) + dst_rel = os.path.join(LIBRARY, rel) + dst = os.path.join(share, dst_rel) + try: + size = os.path.getsize(src) + except OSError as e: + print(f" SKIP unreadable {rel}: {e}", file=sys.stderr) + continue + + action = "move" + if os.path.exists(dst): + dsize = os.path.getsize(dst) + if dsize == size and filecmp.cmp(src, dst, shallow=False): + # true duplicate: quarantine the root copy rather than delete it + dst_rel = os.path.join(QUARANTINE, rel) + dst = os.path.join(share, dst_rel) + action = "quarantine-identical" + n_ident += 1 + else: + base, ext = os.path.splitext(dst_rel) + dst_rel = f"{base}__fromroot{ext}" + dst = os.path.join(share, dst_rel) + action = "keep-both-differs" + n_diff += 1 + else: + n_clean += 1 + moved_bytes += size + + rows.append((action, size, nfc(rel), nfc(dst_rel))) + if args.limit and len(rows) >= args.limit: + break + + print(f"share : {share}") + print(f"mode : {'APPLY' if args.apply else 'DRY RUN'}") + print(f"clean moves : {n_clean} ({moved_bytes / 1e9:.2f} GB)") + print(f"identical dupes : {n_ident} -> {QUARANTINE}/ (kept, not deleted)") + print(f"differing dupes : {n_diff} -> kept both, '__fromroot' suffix") + print(f"total : {len(rows)}") + print(f"manifest : {manifest_path}\n") + + if n_diff: + print("Files kept side by side (review in the Phase 6 dedupe pass):") + for a, s, r, d in rows: + if a == "keep-both-differs": + print(f" {r}\n -> {d}") + print() + + errors = 0 + with open(manifest_path, "w", encoding="utf-8") as mf: + mf.write("action\tbytes\tnas_old\tnas_new\tmac_old\tmac_new\n") + for action, size, rel, dst_rel in rows: + mac_old = f"{MAC_MUSIC_ROOT}/{rel}" + mac_new = f"{MAC_MUSIC_ROOT}/{dst_rel}" + mf.write(f"{action}\t{size}\t{rel}\t{dst_rel}\t{mac_old}\t{mac_new}\n") + + if not args.apply: + continue + src = os.path.join(share, rel) + dst = os.path.join(share, dst_rel) + try: + os.makedirs(os.path.dirname(dst), exist_ok=True) + if os.path.exists(dst) and action == "move": + print(f" SKIP (appeared since scan): {rel}", file=sys.stderr) + continue + os.rename(src, dst) # same share -> server-side rename + except OSError as e: + errors += 1 + print(f" ERROR moving {rel}: {e}", file=sys.stderr) + + if args.apply: + print(f"applied. errors={errors}") + + if args.prune: + # Loop until stable: over SMB, os.listdir can return a stale snapshot, so a + # single pass reliably leaves a tail of folders behind (observed: 189 found, + # 156 removed, 33 left). rmtree is idempotent here, so just re-scan until a + # pass removes nothing. + pruned_total = 0 + for _ in range(10): + n = prune_pass(share, args.apply) + pruned_total += n + if n == 0: + break + print(f"prune: {pruned_total} empty root artist folder(s) " + f"{'removed' if args.apply else 'would be removed'}") + + return 1 if errors else 0 + + +def prune_pass(share: str, apply: bool) -> int: + """One prune sweep. Returns how many folders were (or would be) removed.""" + import shutil + + pruned = 0 + for entry in sorted(os.listdir(share)): + if entry in KEEP_AT_ROOT or entry.startswith("."): + continue + folder = os.path.join(share, entry) + if not os.path.isdir(folder): + continue + # only prune when nothing but Synology/macOS metadata remains + leftover = [] + for dirpath, dirnames, filenames in os.walk(folder): + if "@eaDir" in dirpath: + continue + leftover += [f for f in filenames if not f.startswith(".")] + if leftover: + print(f" KEEP {entry} — still holds {len(leftover)} non-metadata file(s)") + continue + if apply: + try: + shutil.rmtree(folder) + pruned += 1 + except OSError as e: + print(f" ERROR pruning {entry}: {e}", file=sys.stderr) + else: + pruned += 1 + return pruned + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/dedupe-report.py b/scripts/dedupe-report.py new file mode 100755 index 0000000..f97bb3f --- /dev/null +++ b/scripts/dedupe-report.py @@ -0,0 +1,377 @@ +#!/usr/bin/env python3 +"""WaxFlow — duplicate analysis across Lexicon, the library, and the pre-WaxFlow archive. + +READ-ONLY. Produces a report and CSVs. Deletes nothing, moves nothing. + +Three populations are reported separately, because conflating them is what makes +"how many duplicates do I have?" unanswerable: + + A. MULTI-ROW — several Lexicon Track rows pointing at ONE file. Nothing to delete + on disk; the question is whether to merge the rows (cues/playlist + links live on the row, not the file). + B. MULTI-FILE — one recording present as several DISTINCT files. These are the real + duplicates, and the ones where quality actually differs. + C. ARCHIVE — files under the pre-WaxFlow archive (Processing/, Disorganized/) + that match a Lexicon track. This is where Beatport/Bandcamp + purchases live, so it is where an UPGRADE is most likely: a bought + WAV/AIFF beating a Tidal FLAC, or a lossless original of something + WaxFlow only found lossy. + +QUALITY RANKING (best first), applied within each group: + genuinely lossless > higher sample rate > higher bit depth > higher bitrate > + larger file > purchased-source path > earliest dateAdded +Lossy-from-lossless transcodes are NOT detected here — that needs a spectral check +(sync-worker/tasks/lossless_verify.py::spectral_cutoff) and is deliberately out of +scope for a read-only report. + +USAGE + ./dedupe-report.py # A + B (fast, metadata only) + ./dedupe-report.py --archive # also scan Processing/ + Disorganized/ (slow) + ./dedupe-report.py --probe # ffprobe every candidate for real codec info + ./dedupe-report.py --out ~/WaxFlow-Backups +""" +from __future__ import annotations + +import argparse +import collections +import csv +import os +import re +import sqlite3 +import subprocess +import sys +import unicodedata + +LEXICON_DB = os.path.expanduser("~/Library/Application Support/lexicon/main.db") +ARCHIVE_ROOTS = ["/Volumes/music/Processing", "/Volumes/music/Disorganized"] +AUDIO_EXT = {".flac", ".m4a", ".mp3", ".wav", ".aiff", ".aif", ".alac", ".aac", ".ogg"} +LOSSLESS_EXT = {".flac", ".wav", ".aiff", ".aif", ".alac"} +PURCHASE_HINTS = ("beatport", "bandcamp", "juno", "traxsource", "qobuz", "purchase", "bought") + + +def nfc(s: str) -> str: + return unicodedata.normalize("NFC", s or "") + + +def norm_key(artist: str, title: str) -> tuple[str, str]: + """Loose match key: lowercase alphanumerics only, remix/edit suffixes retained. + + Retaining remix info matters — 'Punk' and 'Punk (Vocal Extended)' are different + records and must not collapse into one another. + """ + def clean(s: str) -> str: + s = nfc(s).lower() + s = re.sub(r"\b(feat|ft|featuring|with)\b.*", "", s) + return re.sub(r"[^a-z0-9]", "", s) + return clean(artist), clean(title) + + +TRACKNO_RE = re.compile(r"^\s*\d{1,4}\s*[.\-_)]\s+") +DATE_SUFFIX_RE = re.compile(r"\s+-\s+\d{1,2}\s+\d{4}\s*$") + + +def split_archive_stem(stem: str) -> tuple[str, str]: + """Split an archive filename stem into (artist, title). + + The pre-WaxFlow folders are exports from playlist tools, so the dominant shape + is "NNN. Artist - Title" — a naive split on the first ' - ' yields an artist of + "1041. Tiesto", which matches nothing. Purchased files sometimes also carry a + trailing purchase-date suffix ("... - 02 2025"). Strip both before splitting, + or the whole archive scan silently reports ~0 matches. + """ + s = TRACKNO_RE.sub("", stem) + s = DATE_SUFFIX_RE.sub("", s) + if " - " in s: + a, t = s.split(" - ", 1) + return a.strip(), t.strip() + return "", s.strip() + + +def probe(path: str) -> dict: + """ffprobe a file for real codec facts. Returns {} on any failure.""" + try: + out = subprocess.run( + ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", + "-select_streams", "a:0", path], + capture_output=True, text=True, timeout=30, + ) + import json as _json + st = _json.loads(out.stdout).get("streams", [{}])[0] + return { + "codec": st.get("codec_name", ""), + "sample_rate": int(st.get("sample_rate") or 0), + "bit_depth": int(st.get("bits_per_raw_sample") or 0), + "bitrate": int(st.get("bit_rate") or 0), + "duration": float(st.get("duration") or 0), + } + except Exception: + return {} + + +MIX_TOKENS = ( + "remix", "vip", "v.i.p", "edit", "bootleg", "mashup", "rework", "flip", + "extended", "radio", "club", "dub", "instrumental", "acoustic", "live", + "original mix", "intro", "outro", +) + + +def mix_tokens(path: str) -> frozenset: + """Remix/edit descriptors present in a filename OR its parent folder. + + Lexicon's `title` frequently keeps the original track name even when the file + is a specific remix, so two rows can share a title while the files are + genuinely different records. Comparing path descriptors catches that, and it + is the difference between a real duplicate and a false positive. + + The parent folder matters as much as the filename: WaxFlow's layout puts the + release in the directory, so the Maduk remix of "Stay" lands at + ".../Stay (Maduk Remix)/Delta Heavy, Dirty Audio, Holly - Stay 4M88.flac" — + the remix tag is in the FOLDER, not the file. + """ + stem = os.path.splitext(os.path.basename(path))[0].lower() + parent = os.path.basename(os.path.dirname(path)).lower() + hay = f"{parent} {stem}" + return frozenset(t for t in MIX_TOKENS if t in hay) + + +def archive_verdict(cand: dict, lex: dict) -> tuple[str, str]: + """Compare an archive file against what Lexicon currently has. + + quality_key() must NOT be used here. Lexicon rows carry bitrate/sampleRate from + Lexicon's own scan, while archive files carry none unless --probe ran, so a + tuple compare makes every archive file lose on sample_rate=0 and reports ~0 + upgrades. Compare only on what is genuinely known for BOTH sides: container + losslessness and file size. + """ + a_ext = os.path.splitext(cand["path"])[1].lower() + l_ext = os.path.splitext(lex["path"])[1].lower() + a_lossless, l_lossless = a_ext in LOSSLESS_EXT, l_ext in LOSSLESS_EXT + a_sz, l_sz = cand.get("size", 0), lex.get("size", 0) or 0 + + if a_lossless and not l_lossless: + return "UPGRADE", f"archive is lossless ({a_ext}), library is lossy ({l_ext})" + if not a_lossless and l_lossless: + return "no better", f"library already lossless ({l_ext})" + # same class -> size is the only honest signal without probing + if l_sz and a_sz > l_sz * 1.05: + return "LARGER", f"{human(a_sz)} vs {human(l_sz)} (same class, {a_ext}/{l_ext})" + if l_sz and a_sz < l_sz * 0.95: + return "no better", f"smaller ({human(a_sz)} vs {human(l_sz)})" + return "equivalent", f"~same size ({human(a_sz)} vs {human(l_sz)})" + + +def quality_key(rec: dict) -> tuple: + """Sort key — higher is better. Mirrors the ranking documented above.""" + ext = os.path.splitext(rec["path"])[1].lower() + lossless = 1 if ext in LOSSLESS_EXT else 0 + purchased = 1 if any(h in rec["path"].lower() for h in PURCHASE_HINTS) else 0 + return ( + lossless, + rec.get("sample_rate", 0), + rec.get("bit_depth", 0), + rec.get("bitrate", 0), + rec.get("size", 0), + purchased, + ) + + +def human(n: float) -> str: + for unit in ("B", "KB", "MB", "GB", "TB"): + if abs(n) < 1024: + return f"{n:.1f}{unit}" + n /= 1024 + return f"{n:.1f}PB" + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--archive", action="store_true", help="also scan Processing/ and Disorganized/") + ap.add_argument("--probe", action="store_true", help="ffprobe candidates for real codec info") + ap.add_argument("--out", default=os.path.expanduser("~/WaxFlow-Backups")) + args = ap.parse_args() + + if not os.path.isfile(LEXICON_DB): + print(f"REFUSE: Lexicon DB not found at {LEXICON_DB}", file=sys.stderr) + return 1 + os.makedirs(args.out, exist_ok=True) + + conn = sqlite3.connect(f"file:{LEXICON_DB}?mode=ro", uri=True) + tracks = [ + dict(id=r[0], title=r[1] or "", artist=r[2] or "", path=nfc(r[3] or ""), + bitrate=int(r[4] or 0), sample_rate=int(r[5] or 0), + duration=float(r[6] or 0), size=int(r[7] or 0), added=r[8] or "") + for r in conn.execute( + "SELECT id,title,artist,location,bitrate,sampleRate,duration,sizeBytes,dateAdded FROM Track" + ) + ] + print(f"Lexicon tracks: {len(tracks)}\n") + + # ---- A. several Lexicon rows -> one file -------------------------------- + by_path = collections.defaultdict(list) + for t in tracks: + if t["path"]: + by_path[t["path"]].append(t) + multi_row = {p: v for p, v in by_path.items() if len(v) > 1} + extra_rows = sum(len(v) - 1 for v in multi_row.values()) + + # ---- B. one recording -> several distinct files ------------------------- + by_key = collections.defaultdict(list) + for t in tracks: + by_key[norm_key(t["artist"], t["title"])].append(t) + multi_file = {} + for k, v in by_key.items(): + paths = {t["path"] for t in v if t["path"]} + if len(paths) > 1: + multi_file[k] = v + + # ---- C. archive candidates --------------------------------------------- + archive_hits = collections.defaultdict(list) + if args.archive: + print("scanning archive roots (this walks ~100 GB over SMB)...") + for root in ARCHIVE_ROOTS: + if not os.path.isdir(root): + print(f" skip {root} (not mounted)") + continue + n = 0 + for dp, dn, fn in os.walk(root): + dn[:] = [d for d in dn if d != "@eaDir"] + for f in fn: + if f.startswith(".") or os.path.splitext(f)[1].lower() not in AUDIO_EXT: + continue + n += 1 + a, t = split_archive_stem(os.path.splitext(f)[0]) + key = norm_key(a, t) + if key in by_key and key != ("", ""): + full = os.path.join(dp, f) + try: + sz = os.path.getsize(full) + except OSError: + continue + archive_hits[key].append({"path": nfc(full), "size": sz}) + print(f" {root}: {n} audio files scanned") + + # ---- optional real codec facts ----------------------------------------- + if args.probe: + targets = [t for v in multi_file.values() for t in v] + targets += [r for v in archive_hits.values() for r in v] + print(f"\nffprobing {len(targets)} candidate files...") + for i, rec in enumerate(targets, 1): + if rec.get("path") and os.path.exists(rec["path"]): + rec.update(probe(rec["path"])) + if i % 50 == 0: + print(f" {i}/{len(targets)}") + + # ---- report ------------------------------------------------------------- + print("\n" + "=" * 72) + print("A. MULTI-ROW — several Lexicon rows, ONE file (nothing to delete on disk)") + print("=" * 72) + print(f" files with >1 row : {len(multi_row)}") + print(f" redundant rows : {extra_rows}") + + print("\n" + "=" * 72) + print("B. MULTI-FILE — one recording, SEVERAL distinct files (the real duplicates)") + print("=" * 72) + reclaim = 0 + rows_b = [] + true_dupes, mix_variants = [], [] + for k, v in sorted(multi_file.items()): + uniq = {} + for t in v: + if t["path"]: + uniq.setdefault(t["path"], t) + ranked = sorted(uniq.values(), key=quality_key, reverse=True) + # If the filenames carry DIFFERENT remix descriptors these are almost + # certainly separate records that merely share a Lexicon title. + token_sets = {mix_tokens(r["path"]) for r in ranked} + differing_mix = len(token_sets) > 1 + (mix_variants if differing_mix else true_dupes).append((k, ranked)) + keep, drop = ranked[0], ranked[1:] + if not differing_mix: + reclaim += sum(d.get("size", 0) for d in drop) + for r in ranked: + rows_b.append({ + "group": f"{k[0]}|{k[1]}", + "verdict": ("DIFFERENT MIX - review" if differing_mix + else ("KEEP" if r is keep else "candidate")), + "artist": r["artist"], "title": r["title"], "path": r["path"], + "ext": os.path.splitext(r["path"])[1], "bytes": r.get("size", 0), + "bitrate": r.get("bitrate", 0), "sample_rate": r.get("sample_rate", 0), + "codec": r.get("codec", ""), "added": r.get("added", ""), + "mix_tokens": "|".join(sorted(mix_tokens(r["path"]))), + }) + print(f" groups sharing artist+title : {len(multi_file)}") + print(f" likely TRUE duplicates : {len(true_dupes)} reclaimable {human(reclaim)}") + print(f" different mixes (keep!) : {len(mix_variants)} <- filenames carry different remix tags") + print("\n --- likely true duplicates ---") + for k, ranked in true_dupes[:15]: + print(f"\n {ranked[0]['artist']} — {ranked[0]['title']}") + for i, r in enumerate(ranked): + tag = "KEEP " if i == 0 else " dup" + print(f" {tag}{os.path.splitext(r['path'])[1]:>5} {human(r.get('size',0)):>9} " + f"{r.get('sample_rate',0) or '?'}Hz {r.get('bitrate',0) or '?'}kbps " + f"...{r['path'][-58:]}") + + rows_c = [] + if args.archive: + print("\n" + "=" * 72) + print("C. ARCHIVE — pre-WaxFlow files matching a Lexicon track (upgrade candidates)") + print("=" * 72) + verdict_counts = collections.Counter() + for k, cands in sorted(archive_hits.items()): + lex = sorted((t for t in by_key[k] if t["path"]), key=quality_key, reverse=True) + if not lex: + continue + best_lex = lex[0] + for c in cands: + verdict, why = archive_verdict(c, best_lex) + verdict_counts[verdict] += 1 + rows_c.append({ + "group": f"{k[0]}|{k[1]}", "verdict": verdict, "reason": why, + "artist": best_lex["artist"], "title": best_lex["title"], + "archive_path": c["path"], "archive_bytes": c["size"], + "archive_ext": os.path.splitext(c["path"])[1], + "lexicon_path": best_lex["path"], "lexicon_bytes": best_lex.get("size", 0), + "lexicon_ext": os.path.splitext(best_lex["path"])[1], + }) + print(f" archive files matching a Lexicon track : {sum(len(v) for v in archive_hits.values())}") + for v, n in verdict_counts.most_common(): + print(f" {v:<12} {n}") + for label in ("UPGRADE", "LARGER"): + hits = [r for r in rows_c if r["verdict"] == label] + if not hits: + continue + print(f"\n --- {label} ({len(hits)}) ---") + for r in hits[:10]: + print(f" {r['artist']} — {r['title']}") + print(f" have {r['lexicon_ext']:>5} {human(r['lexicon_bytes']):>9} ...{r['lexicon_path'][-54:]}") + print(f" archive {r['archive_ext']:>5} {human(r['archive_bytes']):>9} ...{r['archive_path'][-54:]}") + + # ---- CSVs --------------------------------------------------------------- + outs = [] + if rows_b: + p = os.path.join(args.out, "dedupe-B-multifile.csv") + with open(p, "w", newline="", encoding="utf-8") as fh: + w = csv.DictWriter(fh, fieldnames=list(rows_b[0].keys())); w.writeheader(); w.writerows(rows_b) + outs.append(p) + if rows_c: + p = os.path.join(args.out, "dedupe-C-archive-upgrades.csv") + with open(p, "w", newline="", encoding="utf-8") as fh: + w = csv.DictWriter(fh, fieldnames=list(rows_c[0].keys())); w.writeheader(); w.writerows(rows_c) + outs.append(p) + if multi_row: + p = os.path.join(args.out, "dedupe-A-multirow.csv") + with open(p, "w", newline="", encoding="utf-8") as fh: + w = csv.writer(fh); w.writerow(["path", "row_count", "track_ids"]) + for path, v in sorted(multi_row.items()): + w.writerow([path, len(v), ";".join(str(t["id"]) for t in v)]) + outs.append(p) + + print("\nwrote:") + for p in outs: + print(" ", p) + print("\nNOTHING WAS DELETED OR MOVED. Review the CSVs before any action.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/deep-repair.sh b/scripts/deep-repair.sh index f1049b1..a5a8aa1 100755 --- a/scripts/deep-repair.sh +++ b/scripts/deep-repair.sh @@ -49,8 +49,19 @@ fi if [ "$NEEDS_REPAIR" = true ]; then log "REPAIR NEEDED: $REASON" - # Hook: Add your alerting/repair dispatch here - # Example: curl -X POST your-webhook-url -d "{\"reason\": \"$REASON\"}" + # Dispatch. This used to be a commented-out example, so the script computed a + # verdict every run and then threw it away — it could never have alerted anyone. + # Uses the same env var as monitor-parity.sh so one setting covers both. + if [ -n "${WAXFLOW_ALERT_WEBHOOK:-}" ]; then + curl -s -m 5 -X POST "$WAXFLOW_ALERT_WEBHOOK" \ + -H 'Content-Type: application/json' \ + -d "{\"source\":\"deep-repair\",\"severity\":\"warning\",\"reason\":\"${REASON}\",\"parity_pct\":${PCT:-null},\"errors\":${ERRORS:-null}}" \ + >/dev/null 2>&1 \ + && log "alert posted to WAXFLOW_ALERT_WEBHOOK" \ + || log "WARN: alert POST failed" + else + log "WARN: repair needed but WAXFLOW_ALERT_WEBHOOK is unset — nobody was told" + fi else log "All clear, no repair needed" fi diff --git a/scripts/ensure-music-mount.sh b/scripts/ensure-music-mount.sh index b8cecde..f5f8f82 100755 --- a/scripts/ensure-music-mount.sh +++ b/scripts/ensure-music-mount.sh @@ -7,7 +7,7 @@ # Deployed on the Lexicon host Mac at ~/.waxflow/ensure-music-mount.sh, run every # ~2 min by LaunchAgent com.waxflow.mount-music. THIS repo copy is canonical — # deploy changes with: -# scp scripts/ensure-music-mount.sh willcurran@192.168.1.116:.waxflow/ensure-music-mount.sh +# scp scripts/ensure-music-mount.sh @:.waxflow/ensure-music-mount.sh # # v2 (2026-07-20) — WRONG-MOUNTPOINT HEAL. Root cause of the Jul-18 sleep incident: # when the Mac sleeps, the SMB session drops; on wake macOS auto-remounts the share @@ -18,15 +18,83 @@ # same share mounted at a wrong mountpoint, unmounts it, clears any stale dir, and # remounts at the canonical path. MP="/Volumes/music" -SHARE_HOST="CCPD-Database._smb._tcp.local" -URL="smb://${SHARE_HOST}/music" +# v5 (2026-08-09) — ADDRESS THE NAS BY IP, NOT BY BONJOUR SERVICE NAME. +# This was "CCPD-Database._smb._tcp.local", the Bonjour SERVICE-INSTANCE name. +# That is not a hostname: it resolves only through service discovery, and when +# the advertisement went stale (after the NAS's Container Manager restart) every +# connection attempt HUNG instead of failing — Finder sat on "Connecting to +# smb://CCPD-...local/music" indefinitely, and so did `mount volume`. +# +# Measured at the time: `CCPD-Database._smb._tcp.local` did not resolve at all, +# while `CCPD-Database.local` and 192.168.1.221 both resolved instantly and +# `smbutil status 192.168.1.221` negotiated fine. Mounting by IP succeeded in 1s. +# +# Configure per host. Put your NAS address in ~/.waxflow/waxflow.conf: +# WAXFLOW_SHARE_HOST=192.168.1.50 # IP is the most reliable +# WAXFLOW_SHARE_NAME=music +# An IP or a plain hostname (nas.local) both work. A Bonjour SERVICE name does +# not — that is the bug described above. +[ -r "$HOME/.waxflow/waxflow.conf" ] && . "$HOME/.waxflow/waxflow.conf" +SHARE_HOST="${WAXFLOW_SHARE_HOST:-}" +SHARE_NAME="${WAXFLOW_SHARE_NAME:-music}" +if [ -z "$SHARE_HOST" ]; then + echo "$(date '+%Y-%m-%dT%H:%M:%S') CONFIG MISSING: set WAXFLOW_SHARE_HOST in ~/.waxflow/waxflow.conf (e.g. your NAS IP)" >>"$HOME/.waxflow/mount-music.log" + exit 64 +fi +URL="smb://${SHARE_HOST}/${SHARE_NAME}" LOG="$HOME/.waxflow/mount-music.log" ts() { date "+%Y-%m-%dT%H:%M:%S"; } # 1) Healthy already? (mounted at the canonical path AND readable) +# +# v3 (2026-08-09) — DO NOT UNMOUNT ON A SINGLE SLOW ls. +# The v2 check treated one failed `ls` as proof of a stale handle and unmounted. +# Under NAS load (observed at load ~10 after a Container Manager restart) a +# healthy share intermittently fails a single `ls`, so this script would unmount +# a WORKING mount and then fail to restore it: `osascript mount volume` cannot +# authenticate from a launchd agent (no GUI/keychain access), which is why the log +# filled with "MOUNT FAILED" every 2 minutes and the sync agent aborted on 32 of +# its last 40 passes while manual runs from a terminal always succeeded. +# +# Require several consecutive failures, spaced out, before touching the mount. A +# genuinely stale handle stays broken; a briefly-slow server recovers on retry. if mount | grep -q " on ${MP} (smbfs"; then - if ls "${MP}" >/dev/null 2>&1; then exit 0; fi - echo "$(ts) stale mount detected at ${MP}, remounting" >>"$LOG" + readable=0; lserr="" + for attempt in 1 2 3; do + lserr="$(ls "${MP}" 2>&1 >/dev/null)" + if [ -z "$lserr" ]; then readable=1; break; fi + [ "$attempt" -lt 3 ] && sleep 3 + done + if [ "$readable" -eq 1 ]; then exit 0; fi + + # EPERM = macOS TCC denying THIS PROCESS access to a network volume. The mount + # is healthy. Fix is one-time and GUI-only: System Settings -> Privacy & + # Security -> Full Disk Access -> add /bin/bash. + case "$lserr" in + *"Operation not permitted"*) + echo "$(ts) EPERM reading ${MP} — TCC is denying this process, mount is FINE. Not unmounting. Grant Full Disk Access to /bin/bash." >>"$LOG" + exit 2 + ;; + esac + + # NEVER unmount by default. + # + # v4 (2026-08-09). v3 still unmounted on any non-EPERM failure, and that is how + # the share was lost: a transient read failure -> unmount -> remount hangs. From + # a launchd agent `mount volume` cannot reach the keychain, and `mount_smbfs` + # gets "Authentication error", so the ONLY way back is a human in Finder. + # Trading a possibly-stale mount for a definitely-unmountable one is a bad deal: + # macOS usually re-establishes a dropped SMB session by itself, and a stale + # handle costs one sync cycle whereas a failed remount costs every cycle until + # someone notices. + # + # Set WAXFLOW_ALLOW_REMOUNT=1 to opt into the old destructive behaviour when + # running interactively (where remounting actually works). + if [ "${WAXFLOW_ALLOW_REMOUNT:-0}" != "1" ]; then + echo "$(ts) ${MP} unreadable (${lserr}) — leaving the mount ALONE (remount is unreliable from launchd). Re-run interactively with WAXFLOW_ALLOW_REMOUNT=1, or reconnect in Finder." >>"$LOG" + exit 3 + fi + echo "$(ts) stale mount at ${MP} (${lserr}) — remounting (WAXFLOW_ALLOW_REMOUNT=1)" >>"$LOG" umount "${MP}" 2>/dev/null || diskutil unmount "${MP}" >/dev/null 2>&1 \ || diskutil unmount force "${MP}" >/dev/null 2>&1 fi @@ -51,10 +119,25 @@ if [ -d "${MP}" ] && ! mount | grep -q " on ${MP} (smbfs"; then fi # 4) Mount (Finder/keychain credentials) and verify it landed at the canonical path. -/usr/bin/osascript -e "try" -e "mount volume \"${URL}\"" -e "end try" >>"$LOG" 2>&1 +# +# HARD TIMEOUT, because `osascript mount volume` can block FOREVER: when the SMB +# session needs re-authentication it waits on NetAuthAgent for a credential +# dialog, which never appears in a launchd context. Observed 2026-08-09 — two +# osascript processes wedged for minutes, blocking BOTH agents and holding the +# sync lock, so the sync stopped running entirely. An agent that hangs is worse +# than one that fails: the failure retries next cycle, the hang never does. +# (macOS has no coreutils `timeout`, hence the watchdog-subshell.) +MOUNT_TIMEOUT="${WAXFLOW_MOUNT_TIMEOUT:-45}" +/usr/bin/osascript -e "try" -e "mount volume \"${URL}\"" -e "end try" >>"$LOG" 2>&1 & +OSPID=$! +( sleep "$MOUNT_TIMEOUT"; kill -9 "$OSPID" 2>/dev/null ) >/dev/null 2>&1 & +WATCHDOG=$! +wait "$OSPID" 2>/dev/null +kill "$WATCHDOG" 2>/dev/null + sleep 3 if mount | grep -q " on ${MP} (smbfs" && ls "${MP}" >/dev/null 2>&1; then echo "$(ts) mounted OK at ${MP}" >>"$LOG"; exit 0 -else - echo "$(ts) MOUNT FAILED (share not at ${MP} after mount attempt)" >>"$LOG"; exit 1 fi +echo "$(ts) MOUNT FAILED (not at ${MP} after ${MOUNT_TIMEOUT}s). If this repeats, the SMB session needs re-auth: reconnect once in Finder (Go > Connect to Server > ${URL})." >>"$LOG" +exit 1 diff --git a/scripts/merge-duplicate-lexicon-rows.py b/scripts/merge-duplicate-lexicon-rows.py new file mode 100755 index 0000000..a784a41 --- /dev/null +++ b/scripts/merge-duplicate-lexicon-rows.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +"""Merge Lexicon Track rows that point at the SAME file. + +WHY + Engine DJ's Track table declares `CONSTRAINT C_path UNIQUE (path)`, so it can + hold at most one row per file. Lexicon had 5,852 rows covering only 5,611 + distinct files — 241 redundant rows that can NEVER sync to Engine. (The proof: + the last good Engine export contained exactly 5,611 tracks.) Those extra rows + are also a plausible contributor to the "SqliteError: FOREIGN KEY constraint + failed" that Lexicon->Engine sync now fails with. + +WHY NOT JUST DELETE THE EXTRAS + Because both rows carry real data. Measured on this library: in 239 of the 241 + groups the row we would drop holds playlist memberships the surviving row does + NOT have — 849 memberships that a naive DELETE would silently destroy. So we + MIGRATE first, then delete. + +WHAT IT DOES (per group of rows sharing one `location`) + 1. Pick the KEEP row: most playlist links + cuepoints, tie-broken by row id. + 2. Re-point every LinkTrackPlaylist row of the other rows onto the KEEP row, + skipping playlists the KEEP row already belongs to (no duplicate entries in + a playlist), preserving `position`. + 3. DELETE the redundant Track rows. Every child table (Cuepoint, Tempomarker, + LinkTrackPlaylist, LinkTagTrack, CloudFile, AlbumartPreview, Waveform) + declares ON DELETE CASCADE from Track, so their leftovers go with them. + PRAGMA foreign_keys=ON is set explicitly — SQLite defaults it OFF, and + without it the cascade silently does not happen. + + Cuepoints/tempo markers/waveforms are NOT merged: the surviving row is chosen + for having the richest set, and merging cue positions between two analyses + would produce nonsense. + +SAFETY + * Refuses to run while Lexicon is open (it caches rows and would clobber us). + * Dry run by default; --apply to write; --limit N for a small first batch. + * integrity_check AND foreign_key_check before and after. + * Verifies afterwards that no playlist membership was lost. + +USAGE + ./merge-duplicate-lexicon-rows.py # dry run + full report + ./merge-duplicate-lexicon-rows.py --apply --limit 10 + ./merge-duplicate-lexicon-rows.py --apply +""" +from __future__ import annotations + +import argparse +import collections +import os +import sqlite3 +import subprocess +import sys +import unicodedata + +DB = os.path.expanduser("~/Library/Application Support/lexicon/main.db") + + +def nfc(s: str) -> str: + return unicodedata.normalize("NFC", s or "") + + +def lexicon_running() -> bool: + r = subprocess.run( + ["pgrep", "-f", "Lexicon.app/Contents/MacOS/Lexicon"], + capture_output=True, text=True, + ) + return r.returncode == 0 and bool(r.stdout.strip()) + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--apply", action="store_true") + ap.add_argument("--limit", type=int, default=0) + ap.add_argument("--db", default=DB) + args = ap.parse_args() + + if not os.path.isfile(args.db): + print(f"REFUSE: no Lexicon DB at {args.db}", file=sys.stderr) + return 1 + if args.apply and lexicon_running(): + print("REFUSE: Lexicon is running. Quit it fully, then re-run --apply.", file=sys.stderr) + return 1 + + conn = sqlite3.connect(args.db, timeout=30) + conn.execute("PRAGMA foreign_keys=ON") # REQUIRED: cascades are off by default + + print("integrity_before :", conn.execute("PRAGMA integrity_check").fetchone()[0]) + print("fk_check_before :", len(list(conn.execute("PRAGMA foreign_key_check"))), "violation(s)") + + groups: dict[str, list[int]] = collections.defaultdict(list) + for tid, loc in conn.execute("SELECT id, location FROM Track"): + if loc: + groups[nfc(loc)].append(tid) + dupes = {k: v for k, v in groups.items() if len(v) > 1} + + def playlists(tid: int) -> dict[int, int]: + return {p: pos for p, pos in + conn.execute("SELECT playlistId, position FROM LinkTrackPlaylist WHERE trackId=?", (tid,))} + + def cues(tid: int) -> int: + return conn.execute("SELECT COUNT(*) FROM Cuepoint WHERE trackId=?", (tid,)).fetchone()[0] + + # Snapshot EVERY (playlist, file) membership in the library, not just the ones + # inside duplicate groups. Scoping this to `dupes` produced a spurious + # "LOST 15" on a run where the full-library comparison proved nothing was lost + # — a verification that cries wolf is worse than none, because the next real + # loss gets waved through. + def all_memberships(c) -> set: + return {(pid, nfc(loc)) for pid, loc in c.execute( + "SELECT l.playlistId, t.location FROM LinkTrackPlaylist l " + "JOIN Track t ON t.id = l.trackId WHERE t.location IS NOT NULL")} + + before_pairs = all_memberships(conn) + + migrated = removed = skipped_existing = 0 + processed = 0 + for loc, ids in sorted(dupes.items()): + ranked = sorted(ids, key=lambda t: (len(playlists(t)) + cues(t), t), reverse=True) + keep, drops = ranked[0], ranked[1:] + keep_pls = set(playlists(keep)) + for d in drops: + for pid, pos in playlists(d).items(): + if pid in keep_pls: + skipped_existing += 1 + continue + if args.apply: + conn.execute( + "UPDATE LinkTrackPlaylist SET trackId=? WHERE trackId=? AND playlistId=?", + (keep, d, pid), + ) + keep_pls.add(pid) + migrated += 1 + if args.apply: + conn.execute("DELETE FROM Track WHERE id=?", (d,)) + removed += 1 + processed += 1 + if args.limit and processed >= args.limit: + break + + if args.apply: + conn.commit() + + print() + print(f"duplicate groups : {len(dupes)}") + print(f"groups processed : {processed}") + print(f"playlist links migrated : {migrated}") + print(f"links already on keep row : {skipped_existing}") + print(f"redundant Track rows {'removed' if args.apply else 'to remove'} : {removed}") + print(f"mode : {'APPLY' if args.apply else 'DRY RUN'}") + + print() + print("integrity_after :", conn.execute("PRAGMA integrity_check").fetchone()[0]) + print("fk_check_after :", len(list(conn.execute("PRAGMA foreign_key_check"))), "violation(s)") + total = conn.execute("SELECT COUNT(*) FROM Track").fetchone()[0] + distinct = conn.execute("SELECT COUNT(DISTINCT location) FROM Track WHERE location IS NOT NULL").fetchone()[0] + print(f"Track rows : {total} distinct locations: {distinct}") + + if args.apply: + after_pairs = all_memberships(conn) + lost = before_pairs - after_pairs + print(f"playlist memberships before {len(before_pairs)} / after {len(after_pairs)} / LOST {len(lost)}") + if lost: + print(" WARNING: memberships lost:", list(lost)[:5], file=sys.stderr) + return 1 + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/repoint-lexicon-local.sh b/scripts/repoint-lexicon-local.sh index 32d6f64..d600971 100755 --- a/scripts/repoint-lexicon-local.sh +++ b/scripts/repoint-lexicon-local.sh @@ -1,46 +1,64 @@ #!/bin/bash -# WaxFlow — re-point Lexicon Track locations from /Volumes/* to canonical LOCAL -# /Users/willcurran/Music/* paths so Engine DJ export includes ALL tracks. +# WaxFlow — re-point Lexicon Track locations to canonical LOCAL +# ~/Music/Database/* paths (paths are derived from $HOME; nothing is hardcoded). # # WHY -# Every Track.location in Lexicon's main.db uses a /Volumes/* prefix: -# • /Volumes/Macintosh HD/Users/willcurran/Music/... (symlink to /, LOCAL disk) -# • /Volumes/music/... (SMB mount of the NAS share) -# Engine DJ export can't ingest /Volumes/* (network/removable-style) paths, so only a -# fraction of the library shows up. The files themselves already exist locally under -# /Users/willcurran/Music (Synology Drive replica), so re-pointing to the canonical -# /Users/... path fixes the export WITHOUT touching a single audio file. +# Engine DJ cannot ingest /Volumes/* locations (network/removable-style paths). +# Historically Lexicon rows carried three non-canonical prefixes: +# • /Volumes/Macintosh HD/Users//Music/... (symlink to /, LOCAL disk) +# • /Volumes/music/... (SMB mount of the NAS share) +# • ~/Music//... (share ROOT replica, i.e. +# outside the library root) +# The files already exist locally under ~/Music/Database, so +# re-pointing fixes the export WITHOUT touching a single audio file. +# +# STATUS (2026-08-08): THIS SCRIPT IS NOW A ONE-SHOT, NOT A RECURRING CHORE. +# It was written on 2026-07-13, fixed the problem, and then the problem came back +# because nothing automated it and new imports kept arriving with /Volumes paths. +# That root cause is now fixed at the source: lexicon_library_path is +# /Users/willcurran/Music/Database and the worker writes into +# MUSIC_LIBRARY_PATH=/music/Database, so every NEW track is canonical on arrival +# (see sync-worker/tasks/process_pipeline.py::_container_to_mac_path and +# tasks/sync_gate.py). Run this once to clean up the legacy rows. If it ever finds +# work again, that is a REGRESSION SIGNAL — something reverted the path contract. # # WHAT IT DOES (NON-DESTRUCTIVE to files; touches ONLY Track.location) -# For each track: -# /Volumes/Macintosh HD/Users/... -> /Users/... (strip the symlink prefix) -# /Volumes/music/ -> /Users/willcurran/Music/ -# ONLY if the resulting local file EXISTS on disk (os.path.isfile). If the local file -# is not present yet (sync incomplete), the track is LEFT UNCHANGED. +# /Volumes/Macintosh HD/Users/... -> /Users/... (strip prefix) +# /Volumes/music/Database/ -> ~/Music/Database/ +# /Volumes/music/Input/ -> ~/Music/Input/ +# /Volumes/music/ -> ~/Music/Database/ (post-consolidation) +# ~/Music// -> ~/Music/Database// +# ONLY if the resulting local file EXISTS on disk (os.path.isfile). If it is not +# present (sync incomplete, or the folder deliberately stays at the share root — +# e.g. Disorganized/), the track is LEFT UNCHANGED. Idempotent: rows already +# canonical are skipped. # Updates ONLY the `location` column. `locationUnique` is Lexicon's immutable -# import-identity key (it already legitimately diverges from `location` on ~5121 rows), -# so it is deliberately NOT modified — this avoids the UNIQUE index entirely. -# Idempotent: rows already at /Users/... are skipped. +# import-identity key (it already legitimately diverges on ~5121 rows), so it is +# deliberately NOT modified — this avoids the UNIQUE index entirely. # # SAFETY GATES (all enforced; refuses otherwise) -# 1. A fresh verified DB backup must exist (heartbeat integrity==ok, Track>0, recent), +# 1. A fresh verified DB backup must exist (heartbeat integrity==ok, Track>0), # or pass --skip-backup-check only if you JUST ran scripts/backup-lexicon-db.sh. -# 2. Lexicon MUST be quit on the Mac (a running Lexicon caches rows and would clobber -# or contend with the write). The script refuses if it sees the Lexicon process. +# 2. Lexicon MUST be quit (a running Lexicon caches rows and would clobber or +# contend with the write). The script refuses if it sees the Lexicon process. # 3. Dry-run by default. Requires --apply to write. --limit N does a SMALL batch first. # 4. PRAGMA integrity_check before and after; per-row old->new audit log. # -# USAGE (run on the ops Mac; SSHes into the Lexicon Mac): -# scripts/backup-lexicon-db.sh # gate: fresh verified backup -# # -> quit Lexicon on 192.168.1.116 <- -# scripts/repoint-lexicon-local.sh # dry-run, full report -# scripts/repoint-lexicon-local.sh --apply --limit 20 # small batch, then reopen+verify +# USAGE +# Runs LOCALLY when invoked on the Lexicon host Mac, or over SSH from an ops box. +# Local mode is auto-detected (the Lexicon DB is present); force with LEXICON_SSH=local. +# +# scripts/backup-lexicon-db.sh # gate: fresh verified backup +# # -> quit Lexicon <- +# scripts/repoint-lexicon-local.sh # dry-run, full report +# scripts/repoint-lexicon-local.sh --apply --limit 20 # small batch, then verify # scripts/repoint-lexicon-local.sh --apply # full run # set -euo pipefail -LEXICON_SSH="${LEXICON_SSH:-willcurran@192.168.1.116}" -LEXICON_DB="${LEXICON_DB:-\$HOME/Library/Application Support/lexicon/main.db}" +# Only used when running from a REMOTE ops box; on the Lexicon host itself +# local mode is auto-detected and this is ignored. +LEXICON_SSH="${LEXICON_SSH:-local}" HEARTBEAT="${HEARTBEAT:-$HOME/.waxflow/logs/lexicon-backup-heartbeat.json}" AUDIT_DIR="${AUDIT_DIR:-$HOME/.waxflow/logs}" APPLY=0; LIMIT=0; SKIP_BACKUP_CHECK=0 @@ -51,11 +69,20 @@ while [ $# -gt 0 ]; do case "$1" in *) echo "unknown arg: $1" >&2; exit 2;; esac; shift; done +# Local mode: we are already ON the Lexicon host, so skip SSH entirely. +LOCAL_DB="$HOME/Library/Application Support/lexicon/main.db" +if [ "$LEXICON_SSH" = "local" ] || [ -f "$LOCAL_DB" ]; then + MODE="local" +else + MODE="ssh" +fi +echo "[repoint] mode=$MODE apply=$APPLY limit=${LIMIT:-none}" + # Gate 1 — fresh verified backup if [ "$SKIP_BACKUP_CHECK" -eq 0 ]; then if [ ! -f "$HEARTBEAT" ]; then echo "REFUSE: no backup heartbeat ($HEARTBEAT). Run scripts/backup-lexicon-db.sh first." >&2; exit 1; fi python3 - "$HEARTBEAT" <<'PY' || exit 1 -import json,sys,time +import json,sys h=json.load(open(sys.argv[1])) assert h.get("status")=="ok" and h.get("integrity")=="ok" and int(h.get("track_count",0))>0, "backup heartbeat not verified" print("backup gate OK: Track=%s integrity=%s file=%s"%(h["track_count"],h["integrity"],h["file"])) @@ -64,8 +91,13 @@ fi # Gate 2 — Lexicon must be quit (enforced only for a real write; dry-run reads are safe) if [ "$APPLY" -eq 1 ]; then - if ssh "$LEXICON_SSH" 'pgrep -x Lexicon >/dev/null 2>&1 || pgrep -f "Lexicon.app/Contents/MacOS/Lexicon" >/dev/null 2>&1'; then - echo "REFUSE: Lexicon is RUNNING on $LEXICON_SSH. Quit Lexicon fully, then re-run --apply." >&2 + if [ "$MODE" = "local" ]; then + RUNNING=$(pgrep -x Lexicon >/dev/null 2>&1 && echo yes || (pgrep -f "Lexicon.app/Contents/MacOS/Lexicon" >/dev/null 2>&1 && echo yes || echo no)) + else + RUNNING=$(ssh "$LEXICON_SSH" 'pgrep -x Lexicon >/dev/null 2>&1 || pgrep -f "Lexicon.app/Contents/MacOS/Lexicon" >/dev/null 2>&1' && echo yes || echo no) + fi + if [ "$RUNNING" = "yes" ]; then + echo "REFUSE: Lexicon is RUNNING. Quit Lexicon fully, then re-run --apply." >&2 exit 1 fi fi @@ -73,30 +105,35 @@ fi TS=$(date +%Y%m%d-%H%M%S) AUDIT="$AUDIT_DIR/repoint-lexicon-$TS.log" mkdir -p "$AUDIT_DIR" -echo "audit -> $AUDIT ; apply=$APPLY limit=$LIMIT" +echo "audit -> $AUDIT" -ssh "$LEXICON_SSH" 'bash -s' "$APPLY" "$LIMIT" <<'REMOTE' | tee "$AUDIT" -set -euo pipefail -APPLY="$1"; LIMIT="$2" -# Standard Lexicon DB path on the Mac (built from remote $HOME so the space in -# "Application Support" is preserved). -DB="$HOME/Library/Application Support/lexicon/main.db" -python3 - "$DB" "$APPLY" "$LIMIT" <<'PY' -import sqlite3, os, sys +read -r -d '' PYSRC <<'PY' || true +import sqlite3, os, sys, unicodedata db, apply, limit = sys.argv[1], sys.argv[2]=="1", int(sys.argv[3]) -HOME="/Users/willcurran" +HOME=os.path.expanduser("~") +MUSIC=HOME+"/Music" +LIB=MUSIC+"/Database" +INP=MUSIC+"/Input" + def newpath(loc): if loc.startswith("/Volumes/Macintosh HD/"): return loc[len("/Volumes/Macintosh HD"):] - if loc.startswith("/Volumes/music/"): return HOME+"/Music/"+loc[len("/Volumes/music/"):] + if loc.startswith("/Volumes/music/Database/"): return LIB+"/"+loc[len("/Volumes/music/Database/"):] + if loc.startswith("/Volumes/music/Input/"): return INP+"/"+loc[len("/Volumes/music/Input/"):] + if loc.startswith("/Volumes/music/"): return LIB+"/"+loc[len("/Volumes/music/"):] + # share-ROOT replica: ~/Music//... but NOT already under Database/ or Input/ + if loc.startswith(MUSIC+"/") and not loc.startswith(LIB+"/") and not loc.startswith(INP+"/"): + return LIB+"/"+loc[len(MUSIC+"/"):] return None + c=sqlite3.connect(db) print("integrity_before=", c.execute("PRAGMA integrity_check").fetchone()[0]) rows=c.execute("SELECT id,location FROM Track ORDER BY id").fetchall() changed=skipped_missing=already=0; n=0 for i,loc in rows: + if not loc: continue np=newpath(loc) if np is None: - if loc.startswith("/Users/"): already+=1 + already+=1 continue if not os.path.isfile(np): skipped_missing+=1; print("SKIP_MISSING id=%d %s"%(i,np)); continue @@ -107,7 +144,14 @@ for i,loc in rows: if limit and n>=limit: break if apply: c.commit() print("integrity_after=", c.execute("PRAGMA integrity_check").fetchone()[0]) -print("SUMMARY changed=%d skipped_missing=%d already_local=%d applied=%s limit=%s"%(changed,skipped_missing,already,apply,limit or "none")) +print("SUMMARY changed=%d skipped_missing=%d already_canonical=%d applied=%s limit=%s"%(changed,skipped_missing,already,apply,limit or "none")) PY -REMOTE + +if [ "$MODE" = "local" ]; then + python3 -c "$PYSRC" "$LOCAL_DB" "$APPLY" "$LIMIT" | tee "$AUDIT" +else + ssh "$LEXICON_SSH" "python3 - \"\$HOME/Library/Application Support/lexicon/main.db\" $APPLY $LIMIT" < ~/Music/Database +# NAS /volume1/music/Input ---> ~/Music/Input +# +# WHY THIS EXISTS (2026-08-08) +# Synology Drive was two-way-syncing the ENTIRE music share (session: share +# 'music', remote '/', local ~/Music/, sync_direction=0, rename_conflict=1). +# Three things went wrong, and all three are structural, not bad luck: +# 1. Engine DJ's library lives at ~/Music/Engine Library. Two-way sync of a +# live 595 MB SQLite file produced 12 "_Conflict" copies (~9.4 GB) and left +# the real m.db with 2 tracks in it. +# 2. ~/Music/SoundSwitch/default.ssproj/*.ssfile is rejected by the server +# ('System error'), and Drive retried it forever — jamming the whole queue +# so genuinely new tracks never arrived. +# 3. Syncing the full 1.1 TB share (DJ Will See 450 G, Processing 95 G, ...) +# filled the Mac to 86%. +# +# ONE-WAY is the entire point. A pull-only replica cannot produce a conflict +# copy, cannot push a half-written Engine database back to the NAS, and cannot +# be jammed by a local file the server won't accept. +# +# TRANSPORT — why this is a hybrid +# Data moves over the SMB mount (/Volumes/music). Change DETECTION happens over +# SSH, because the two costs are wildly different: +# * full rsync scan over SMB ....... 5 m 46 s (3,736 dirs x round-trips) +# * `find -newermt` over SSH ....... 0.9 s (NAS walks its own disk) +# We cannot rsync over SSH: DSM refuses `rsync --server` for non-admin users +# ("Permission denied, please try again"), which needs a Control Panel -> +# File Services -> rsync toggle. Detection-over-SSH gets ~99% of the win +# without touching DSM config. +# +# INCREMENTAL pass (every run): ask the NAS which files changed since the last +# successful pass, copy just those over SMB. Typically <2 s. +# RECONCILE pass (every RECONCILE_SECONDS, default 6 h, and whenever state is +# missing/stale/SSH is down): full rsync. This is the safety net that catches +# anything detection missed — including /volume1/music/Database/Aktive, the one +# directory of 3,736 that is mode 000 to the SSH user but readable over SMB. +# +# WHAT IT DOES NOT DO +# * Never touches ~/Music/Engine Library — Engine's library is LOCAL ONLY. +# (Engine stores track paths as '../Database//...', relative to the +# Engine Library folder, so a NAS copy is meaningless as well as dangerous.) +# * No --delete. A deletion is the one thing a sync bug cannot undo, so v1 +# never removes anything from the Mac. Revisit after a week of clean runs. +# * No push. Nothing on the Mac is ever written back to the NAS. +# +# WHY --size-only +# SMB mtimes round-trip unreliably, and library audio is immutable once the +# worker has moved it into place (_move_to_library uses an atomic same-volume +# rename). Size is a sufficient and much cheaper comparison than checksums over +# 200 GB. CAVEAT: a same-size re-encode or an in-place tag edit on the NAS will +# NOT be detected by the reconcile pass. Don't edit tags NAS-side; do it in +# Lexicon. (The incremental pass is mtime-based and *will* catch those.) +# +# PARTIAL FILES +# rsync transfers to a temp name and renames into place, so a partially copied +# file is never visible at its final path. If the worker is mid-write on the NAS +# we may copy a short file; the next cycle sees a size/mtime mismatch and +# re-copies. Phase 3's import gate (the heartbeat below) is what stops Lexicon +# importing inside that window. +# +# Deployed on the Lexicon host Mac at ~/.waxflow/sync-nas-to-mac.sh, run every +# 120 s by LaunchAgent com.waxflow.sync-database. THIS repo copy is canonical: +# scp scripts/sync-nas-to-mac.sh @:.waxflow/sync-nas-to-mac.sh +# +# Usage: sync-nas-to-mac.sh [--dry-run] [--reconcile] + +set -uo pipefail + +[ -r "$HOME/.waxflow/waxflow.conf" ] && . "$HOME/.waxflow/waxflow.conf" +SRC="${WAXFLOW_SYNC_SRC:-/Volumes/music}" +DST="${WAXFLOW_SYNC_DST:-$HOME/Music}" +# SSH target used ONLY for fast change detection (see TRANSPORT). Set +# WAXFLOW_NAS_SSH in ~/.waxflow/waxflow.conf. If it cannot connect, the +# script simply falls back to a full reconcile, so this is optional. +NAS_SSH="${WAXFLOW_NAS_SSH:-}" +NAS_ROOT="${WAXFLOW_NAS_ROOT:-/volume1/music}" +FOLDERS=("Database" "Input") +RECONCILE_SECONDS="${WAXFLOW_RECONCILE_SECONDS:-21600}" # 6 h +DETECT_SLACK_SECONDS=300 # re-ask a little further back than strictly needed + +STATE_DIR="$HOME/.waxflow" +LOG="$STATE_DIR/sync-nas-to-mac.log" +LOCK="$STATE_DIR/sync-nas-to-mac.lock" +STATE="$STATE_DIR/sync-nas-to-mac.state" +HEARTBEAT="$SRC/Input/.waxflow-sync-heartbeat" +MOUNT_HELPER="$STATE_DIR/ensure-music-mount.sh" +MAX_LOG_BYTES=$((5 * 1024 * 1024)) + +# GNU rsync if available: macOS ships openrsync, which silently drops -e and does +# not implement --files-from. Homebrew rsync is required for the incremental pass. +RSYNC="/opt/homebrew/bin/rsync"; [ -x "$RSYNC" ] || RSYNC="$(command -v rsync)" + +DRY_RUN=0; FORCE_RECONCILE=0 +for a in "$@"; do + case "$a" in + --dry-run) DRY_RUN=1 ;; + --reconcile) FORCE_RECONCILE=1 ;; + esac +done + +mkdir -p "$STATE_DIR" +ts() { date "+%Y-%m-%dT%H:%M:%S"; } +log() { echo "$(ts) $*" >>"$LOG"; } + +# --- log rotation (the sibling mount-music.log grew to 728 KB with none) ------- +if [ -f "$LOG" ] && [ "$(stat -f%z "$LOG" 2>/dev/null || echo 0)" -gt "$MAX_LOG_BYTES" ]; then + mv -f "$LOG" "$LOG.1" +fi + +# --- single instance ---------------------------------------------------------- +# A slow reconcile (200 GB over SMB) must not have a second copy stacked on it. +if ! mkdir "$LOCK" 2>/dev/null; then + if [ -f "$LOCK/pid" ] && kill -0 "$(cat "$LOCK/pid" 2>/dev/null)" 2>/dev/null; then + exit 0 # previous run still going; not an error + fi + log "stale lock (pid $(cat "$LOCK/pid" 2>/dev/null || echo '?') gone) — reclaiming" + rm -rf "$LOCK"; mkdir "$LOCK" 2>/dev/null || exit 0 +fi +echo $$ >"$LOCK/pid" +trap 'rm -rf "$LOCK"' EXIT INT TERM + +# --- precondition: the SMB mount must be healthy ------------------------------ +[ -x "$MOUNT_HELPER" ] && "$MOUNT_HELPER" >/dev/null 2>&1 +if ! mount | grep -q " on ${SRC} (smbfs"; then + log "ABORT: ${SRC} is not mounted (ensure-music-mount.sh could not heal it)" + exit 1 +fi +LSERR="$(ls "$SRC" 2>&1 >/dev/null)" +if [ -n "$LSERR" ]; then + case "$LSERR" in + *"Operation not permitted"*) + # macOS TCC, not a broken mount. The share is healthy; THIS process is + # denied. Nothing the script can do about it, and retrying forever just + # fills the log — say exactly how to fix it and stop. + log "ABORT: TCC denies this process access to ${SRC} (mount is healthy). One-time fix: System Settings -> Privacy & Security -> Full Disk Access -> add /bin/bash, then: launchctl kickstart -k gui/\$(id -u)/com.waxflow.sync-database" + ;; + *) + log "ABORT: ${SRC} unreadable (${LSERR})" + ;; + esac + exit 1 +fi + +RSYNC_COMMON=( + -rt --size-only --modify-window=2 + --no-perms --no-owner --no-group --omit-dir-times + --exclude=.DS_Store --exclude=._* --exclude=@eaDir + --exclude=.SynologyWorkingDirectory --exclude=#recycle + --exclude=*_Conflict* # never replicate Synology conflict artefacts + --exclude=.waxflow-sync-heartbeat +) +[ "$DRY_RUN" -eq 1 ] && RSYNC_COMMON+=(--dry-run) + +# --- decide pass type --------------------------------------------------------- +NOW=$(date +%s) +LAST_OK=0; LAST_RECONCILE=0 +# shellcheck disable=SC1090 +[ -f "$STATE" ] && . "$STATE" 2>/dev/null +MODE="incremental" +[ "$LAST_OK" -eq 0 ] && MODE="reconcile" +[ $((NOW - LAST_RECONCILE)) -ge "$RECONCILE_SECONDS" ] && MODE="reconcile" +[ "$FORCE_RECONCILE" -eq 1 ] && MODE="reconcile" + +START_EPOCH=$NOW +TOTAL_FILES=0 +STATUS="ok" + +if [ "$MODE" = "incremental" ]; then + # Ask the NAS what changed. -newermt with an absolute timestamp so clock skew + # between the two hosts cannot silently narrow the window. + SINCE=$(( LAST_OK - DETECT_SLACK_SECONDS )) + SINCE_FMT=$(date -u -r "$SINCE" "+%Y-%m-%d %H:%M:%S") + LIST="$STATE_DIR/.sync-changed.$$" + # One SSH call per folder, cd'ing in first so `find .` yields paths already + # relative to the folder; sed re-prefixes them to be relative to $SRC, which + # is exactly what --files-from wants. (`find -printf` is GNU-only and this NAS + # is busybox-ish, so the cd+sed form is the portable one.) + # NOTE: newline-in-filename would break the line-oriented list. None exist in + # this library, and the reconcile pass would catch such a file anyway. + : >"$LIST" + if [ -n "$NAS_SSH" ] && ssh -o BatchMode=yes -o ConnectTimeout=10 "$NAS_SSH" true 2>/dev/null; then + # The filter MUST live in the find, not in rsync: --files-from bypasses + # rsync's --exclude rules for explicitly listed paths, so a Synology + # @eaDir entry (extended-attribute streams that do not exist over SMB) + # fails the whole pass with "link_stat ... No such file or directory". + for f in "${FOLDERS[@]}"; do + ssh -o BatchMode=yes -o ConnectTimeout=10 "$NAS_SSH" \ + "cd '$NAS_ROOT/$f' 2>/dev/null && find . -type f -newermt '$SINCE_FMT UTC' \ + ! -path '*/@eaDir/*' ! -name '.DS_Store' ! -name '._*' \ + ! -name '*@SynoEAStream' ! -name '*@SynoResource' \ + ! -name '*_Conflict*' 2>/dev/null | sed 's|^\./|$f/|'" \ + >>"$LIST" 2>/dev/null + done + N=$(grep -c . "$LIST" 2>/dev/null || echo 0) + if [ "$N" -gt 0 ]; then + OUT=$("$RSYNC" "${RSYNC_COMMON[@]}" --files-from="$LIST" "$SRC/" "$DST/" 2>&1) + RC=$? + if [ $RC -ne 0 ]; then + STATUS="error" + log "ERROR incremental rsync rc=$RC: $(echo "$OUT" | tail -3 | tr '\n' ' ')" + else + TOTAL_FILES=$N + log "incremental: $N changed file(s) since $SINCE_FMT UTC" + fi + fi + rm -f "$LIST" + else + log "SSH detection unavailable — falling back to reconcile" + rm -f "$LIST" + MODE="reconcile" + fi +fi + +if [ "$MODE" = "reconcile" ]; then + for folder in "${FOLDERS[@]}"; do + if [ ! -d "$SRC/$folder" ]; then + log "SKIP $folder — not present on the NAS"; continue + fi + mkdir -p "$DST/$folder" + OUT=$("$RSYNC" "${RSYNC_COMMON[@]}" --stats "$SRC/$folder/" "$DST/$folder/" 2>&1) + RC=$? + # macOS openrsync says "Number of files transferred:"; GNU rsync 3.x says + # "Number of regular files transferred:". Match either. + N=$(echo "$OUT" | awk '/Number of (regular )?files transferred:/ {gsub(/,/,"",$NF); print $NF; exit}') + N=${N:-0} + TOTAL_FILES=$((TOTAL_FILES + N)) + if [ $RC -ne 0 ]; then + STATUS="error" + log "ERROR reconcile rsync $folder rc=$RC: $(echo "$OUT" | tail -3 | tr '\n' ' ')" + fi + done +fi + +ELAPSED=$(( $(date +%s) - START_EPOCH )) +# Log every pass, not just the ones that moved data — a silent log is +# indistinguishable from a dead agent, which is exactly how the July 13 fix +# rotted unnoticed. Rotation above keeps this bounded. +log "$MODE pass complete: $TOTAL_FILES file(s) in ${ELAPSED}s (status=$STATUS)" + +# --- persist state ------------------------------------------------------------ +if [ "$DRY_RUN" -eq 0 ] && [ "$STATUS" = "ok" ]; then + [ "$MODE" = "reconcile" ] && LAST_RECONCILE=$START_EPOCH + printf 'LAST_OK=%s\nLAST_RECONCILE=%s\n' "$START_EPOCH" "$LAST_RECONCILE" >"$STATE" +fi + +# --- heartbeat ---------------------------------------------------------------- +# Written NAS-side, inside Input/, so the WaxFlow worker container can read it at +# /downloads/.waxflow-sync-heartbeat. The organizing stage holds any track whose +# file is newer than completed_at — that is what makes importing by LOCAL +# /Users/... path safe despite replication lag. +if [ "$DRY_RUN" -eq 0 ] && [ -d "$SRC/Input" ]; then + # Truncate-in-place, NOT write-tmp-then-mv. The share has Synology's recycle + # bin enabled, and every replace-by-rename was being captured as a deletion — + # one #recycle entry every 120 s, forever. A torn read is harmless here: + # sync_gate.py fails open on unparseable JSON by design. + printf '%s\n' "{\"status\":\"$STATUS\",\"mode\":\"$MODE\",\"completed_at\":$(date +%s),\"completed_at_iso\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\",\"files_transferred\":$TOTAL_FILES,\"elapsed_seconds\":$ELAPSED,\"host\":\"$(scutil --get LocalHostName 2>/dev/null || hostname)\",\"folders\":\"${FOLDERS[*]}\"}" \ + >"$HEARTBEAT" 2>/dev/null || true +fi + +# --- Lexicon path-drift watchdog (read-only) ---------------------------------- +# Lexicon CANONICALISES imported paths through the boot-volume symlink, rewriting +# the /Users/... path WaxFlow hands it into /Volumes/Macintosh HD/Users/... +# +# SEVERITY MATTERS, and the two cases are NOT the same: +# +# /Volumes/Macintosh HD/... COSMETIC. That is a symlink to /, so the path +# resolves to the identical file. Verified 2026-08-09 +# against a real Engine library: all 40 rows carrying +# this prefix were present in Engine. It does NOT +# break Engine DJ export, contrary to the original +# diagnosis. Tidy it up whenever; nothing is broken. +# +# /Volumes//... REAL. An SMB path breaks the moment the share is +# unmounted, and the file exists only on the NAS. +# +# Reporting both at the same volume would train you to ignore the line that +# matters. Read-only (mode=ro), so it is safe while Lexicon is running. +LEXICON_DB="$HOME/Library/Application Support/lexicon/main.db" +if [ -f "$LEXICON_DB" ]; then + DRIFT=$(python3 - "$LEXICON_DB" <<'PY' 2>/dev/null +import sqlite3, sys +try: + c = sqlite3.connect(f"file:{sys.argv[1]}?mode=ro", uri=True) + sym = c.execute("SELECT COUNT(*) FROM Track WHERE location LIKE '/Volumes/Macintosh HD/%'").fetchone()[0] + smb = c.execute("SELECT COUNT(*) FROM Track WHERE location LIKE '/Volumes/%' " + "AND location NOT LIKE '/Volumes/Macintosh HD/%'").fetchone()[0] + print(f"{sym} {smb}") +except Exception: + print("") +PY +) + set -- $DRIFT + SYMLINK_ROWS="${1:-0}"; SMB_ROWS="${2:-0}" + if [ "${SMB_ROWS:-0}" -gt 0 ] 2>/dev/null; then + log "LEXICON PATH DRIFT (action needed): $SMB_ROWS row(s) on an SMB /Volumes path — these break when the share unmounts. Quit Lexicon and run scripts/repoint-lexicon-local.sh --apply" + elif [ "${SYMLINK_ROWS:-0}" -gt 20 ] 2>/dev/null; then + log "note: $SYMLINK_ROWS row(s) carry the /Volumes/Macintosh HD symlink prefix (cosmetic — resolves fine, Engine accepts it). Tidy with repoint-lexicon-local.sh when convenient." + fi +fi + +[ "$STATUS" = "ok" ] || exit 1 +exit 0 diff --git a/sync-api/db.py b/sync-api/db.py index 1234075..9e4e7e9 100644 --- a/sync-api/db.py +++ b/sync-api/db.py @@ -10,10 +10,16 @@ def get_connection() -> sqlite3.Connection: # timeout=30 + PRAGMA busy_timeout make the API wait for a lock instead of # instantly raising "database is locked" when the worker holds a write lock # (the root cause of intermittent 500s on approve/reject during heavy soak). + # + # busy_timeout MUST match the connect timeout. It was 5000 ms against a + # timeout=30 s, and busy_timeout is what actually governs, so the effective + # wait was 5 s, not 30 — PATCH /api/settings still returned + # {"detail":"database is locked"} whenever the worker was mid-index and had + # to be retried by hand. Both are 30 s now. conn = sqlite3.connect(DB_PATH, timeout=30) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA busy_timeout=5000") + conn.execute("PRAGMA busy_timeout=30000") conn.execute("PRAGMA foreign_keys=ON") return conn diff --git a/sync-api/init_db.py b/sync-api/init_db.py index 5f9eac5..ae70d97 100644 --- a/sync-api/init_db.py +++ b/sync-api/init_db.py @@ -1,6 +1,8 @@ #!/usr/bin/env python3 """Initialize the sync database with all required tables.""" +import os + from db import get_db @@ -140,15 +142,24 @@ def init(): ('lexicon_post_processing', 'analyze,cues,tags,cloud'), ('sync_mode', 'scan'), ('webhook_url', ''), - -- Path PREFIX the Lexicon host Mac uses to read the NAS music tree. - -- The worker writes to container /music (== NAS /volume1/music), which - -- the Mac reads over an SMB mount at /Volumes/music (live, instant — - -- no sync lag or ACL dependency). Downloads ALSO propagate to the Mac's - -- Synology-Drive replica (~/Music) because finished files keep the - -- inherited Synology ACL (see the _download_track_via_tiddl delivery - -- note: chmod would strip that ACL and strand the file). - ('lexicon_library_path', '/Volumes/music'), + -- Path PREFIX the Lexicon host Mac uses to read the music tree. The + -- worker rewrites its own container path to this before telling Lexicon + -- where a file is (process_pipeline._container_to_mac_path). + -- + -- Default is the SMB mount, which needs no replication and works out of + -- the box. If you ALSO export to Engine DJ, consider pointing these at a + -- LOCAL path on the Mac instead (e.g. ~/Music/Database) and running + -- scripts/sync-nas-to-mac.sh -- see "Local paths and Engine DJ" in the + -- README. Set them in Settings, or seed via LEXICON_LIBRARY_PATH / + -- LEXICON_INPUT_PATH. Do NOT hardcode a username here: this seeds every + -- new install. + ('lexicon_library_path', '/Volumes/music/Database'), ('lexicon_input_path', '/Volumes/music/Input'), + -- Replication gate (see tasks/sync_gate.py). Fails open by design. + ('sync_gate_enabled', '1'), + ('sync_gate_heartbeat_path', '/downloads/.waxflow-sync-heartbeat'), + ('sync_gate_max_hold_seconds', '3600'), + ('sync_gate_heartbeat_max_age_seconds', '1800'), ('tidal_download_quality', 'max'), ('downloads_path', '/downloads'), ('lexicon_api_url', ''), @@ -202,7 +213,7 @@ def init(): ('metadata_fallback_enabled', '1'), ('metadata_fallback_batch', '8'), ('metadata_fallback_interval_seconds', '3600'), - ('musicbrainz_user_agent', 'WaxFlow/2.9 (https://github.com/rancur/waxflow)'), + ('musicbrainz_user_agent', 'WaxFlow/2.11 (https://github.com/rancur/waxflow)'), -- Acoustic-fingerprint fallback (tasks/acoustid_fallback.py). fpcalc is -- in the image; provide a free AcoustID key here + flip enabled to -- activate (both read live, no redeploy). OFF until provisioned. @@ -585,6 +596,30 @@ def init(): conn.execute("ALTER TABLE import_queue ADD COLUMN next_retry_at TEXT") print("Added import_queue.next_retry_at column.") + # Host-specific paths from the environment. These are the ONE pair of + # settings that cannot have a correct universal default: they describe + # where the Lexicon *host* sees the music tree, which differs per machine + # (SMB mount vs a local replica, and any local path contains a username). + # Seeded above with the SMB default; override here without editing SQL. + # Applied on every start so .env stays the source of truth, but only when + # the variable is actually set. + for env_key, cfg_key in ( + ("LEXICON_LIBRARY_PATH", "lexicon_library_path"), + ("LEXICON_INPUT_PATH", "lexicon_input_path"), + ): + val = (os.environ.get(env_key) or "").strip() + if val: + cur = conn.execute( + "SELECT value FROM app_config WHERE key = ?", (cfg_key,) + ).fetchone() + if cur is None or cur[0] != val: + conn.execute( + "INSERT INTO app_config (key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value = excluded.value", + (cfg_key, val), + ) + print(f"Set {cfg_key} from {env_key}: {val}") + print("Database initialized successfully.") diff --git a/sync-api/main.py b/sync-api/main.py index 42d63f5..414cbcf 100644 --- a/sync-api/main.py +++ b/sync-api/main.py @@ -1,4 +1,5 @@ import os +import pathlib from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -16,10 +17,18 @@ from routes.status import router as status_router from routes.wanted import router as wanted_router +# Version comes from the VERSION file baked into the image at build time, so +# it can never drift from the release tag (it used to be hardcoded "2.1.0" +# while VERSION said 2.10.1). +try: + __version__ = pathlib.Path("/app/VERSION").read_text().strip() +except OSError: + __version__ = os.environ.get("VERSION", "0.0.0") + app = FastAPI( title="WaxFlow API", description="All your music, flowing home. Spotify Liked Songs to Lexicon DJ.", - version="2.1.0", + version=__version__, ) _cors_origins_env = os.environ.get("CORS_ORIGINS", "") @@ -49,4 +58,4 @@ @app.get("/") async def root(): - return {"service": "waxflow", "version": "2.1.0"} + return {"service": "waxflow", "version": __version__} diff --git a/sync-api/routes/admin.py b/sync-api/routes/admin.py index 5892e60..e7af976 100644 --- a/sync-api/routes/admin.py +++ b/sync-api/routes/admin.py @@ -273,7 +273,7 @@ async def check_update(): return { "current_version": current, "latest_version": latest, - "update_available": latest != current and latest > current, + "update_available": _is_newer(latest, current), "release_url": data.get("html_url"), "release_notes": data.get("body", "")[:500], "published_at": data.get("published_at"), @@ -284,6 +284,32 @@ async def check_update(): return {"current_version": current, "update_available": False} + +def _version_tuple(v: str) -> tuple: + """Parse a dotted version into comparable ints, ignoring any suffix.""" + parts = [] + for chunk in (v or "").split("."): + digits = "" + for ch in chunk: + if ch.isdigit(): + digits += ch + else: + break + parts.append(int(digits) if digits else 0) + return tuple(parts) + + +def _is_newer(latest: str, current: str) -> bool: + """True when `latest` is a strictly newer release than `current`. + + A plain string compare is wrong across a version-component boundary: + "2.9.0" > "2.10.1" is True lexicographically, so the update banner was + inverted for exactly the 2.9 -> 2.10 upgrade this deployment was sitting on. + """ + if not latest or not current: + return False + return _version_tuple(latest) > _version_tuple(current) + # ============================================================ # Config Backup System # ============================================================ diff --git a/sync-api/services/downloader.py b/sync-api/services/downloader.py deleted file mode 100644 index 39979ee..0000000 --- a/sync-api/services/downloader.py +++ /dev/null @@ -1,160 +0,0 @@ -import os -import json -import httpx -from datetime import datetime, timezone - -from db import get_db - -TIDARR_API = os.environ.get("TIDARR_URL", "http://localhost:8484") # optional legacy fallback -MUSIC_PATH = os.environ.get("MUSIC_LIBRARY_PATH", "/music") - - -class DownloaderService: - """Manages track downloads via tiddl CLI (primary) with optional Tidarr fallback.""" - - def __init__(self): - self.tidarr_url = TIDARR_API - self.music_path = MUSIC_PATH - - async def download_track(self, track: dict, tidal_id: str) -> dict: - """ - Queue a download via Tidal. Returns status dict. - NOTE: This legacy code path uses the Tidarr API as fallback. - Primary downloads now go through tiddl CLI in the sync-worker. - """ - track_id = track["id"] - now = datetime.now(timezone.utc).isoformat() - - with get_db() as conn: - # Create or update queue entry - existing = conn.execute( - "SELECT id, attempts FROM download_queue WHERE track_id = ?", (track_id,) - ).fetchone() - - if existing: - if existing["attempts"] >= 3: - return {"status": "max_attempts", "track_id": track_id} - conn.execute( - """UPDATE download_queue SET - status = 'downloading', started_at = ?, attempts = attempts + 1 - WHERE track_id = ?""", - (now, track_id), - ) - else: - conn.execute( - """INSERT INTO download_queue - (track_id, priority, source, status, attempts, started_at) - VALUES (?, 0, 'tidarr', 'downloading', 1, ?)""", - (track_id, now), - ) - - conn.execute( - "UPDATE tracks SET download_status = 'downloading', download_attempts = download_attempts + 1, updated_at = datetime('now') WHERE id = ?", - (track_id,), - ) - - # Submit to Tidarr (legacy fallback — primary path uses tiddl CLI in worker) - try: - async with httpx.AsyncClient(timeout=30.0) as client: - resp = await client.post( - f"{self.tidarr_url}/api/download/track", - json={"id": tidal_id}, - ) - - if resp.status_code in (200, 201, 202): - data = resp.json() - - with get_db() as conn: - conn.execute( - "UPDATE download_queue SET status = 'queued' WHERE track_id = ?", - (track_id,), - ) - conn.execute( - "UPDATE tracks SET download_status = 'queued', updated_at = datetime('now') WHERE id = ?", - (track_id,), - ) - conn.execute( - "INSERT INTO activity_log (event_type, track_id, message, details) VALUES (?, ?, ?, ?)", - ("download_queued", track_id, - f"Download queued via Tidal for tidal_id {tidal_id}", - json.dumps(data)), - ) - - return {"status": "queued", "track_id": track_id, "tidal_response": data} - else: - error = f"Tidal download returned HTTP {resp.status_code}: {resp.text}" - await self._mark_failed(track_id, error) - return {"status": "failed", "track_id": track_id, "error": error} - - except Exception as e: - error = str(e) - await self._mark_failed(track_id, error) - return {"status": "failed", "track_id": track_id, "error": error} - - async def check_download_status(self, track: dict) -> dict: - """Check if a download has completed by looking for the file.""" - track_id = track["id"] - artist = track.get("artist", "Unknown") - title = track.get("title", "Unknown") - album = track.get("album", "Unknown") - - # Check common download output paths - possible_paths = [ - os.path.join(self.music_path, artist, album, f"{title}.flac"), - os.path.join(self.music_path, artist, f"{title}.flac"), - os.path.join(self.music_path, artist, album), - ] - - for path in possible_paths: - if os.path.isfile(path): - return await self._mark_complete(track_id, path) - if os.path.isdir(path): - # Look for FLAC files in directory - for fname in os.listdir(path): - if fname.lower().endswith(".flac"): - full = os.path.join(path, fname) - if title.lower() in fname.lower(): - return await self._mark_complete(track_id, full) - - return {"status": "pending", "track_id": track_id} - - async def _mark_complete(self, track_id: int, file_path: str) -> dict: - now = datetime.now(timezone.utc).isoformat() - with get_db() as conn: - conn.execute( - """UPDATE download_queue SET status = 'complete', completed_at = ? - WHERE track_id = ?""", - (now, track_id), - ) - conn.execute( - """UPDATE tracks SET - download_status = 'complete', file_path = ?, - pipeline_stage = 'verifying', updated_at = datetime('now') - WHERE id = ?""", - (file_path, track_id), - ) - conn.execute( - "INSERT INTO activity_log (event_type, track_id, message) VALUES (?, ?, ?)", - ("download_complete", track_id, f"Download complete: {file_path}"), - ) - return {"status": "complete", "track_id": track_id, "file_path": file_path} - - async def _mark_failed(self, track_id: int, error: str): - with get_db() as conn: - conn.execute( - "UPDATE download_queue SET status = 'failed', error = ? WHERE track_id = ?", - (error, track_id), - ) - conn.execute( - """UPDATE tracks SET - download_status = 'failed', download_error = ?, - pipeline_stage = 'error', pipeline_error = ?, - updated_at = datetime('now') - WHERE id = ?""", - (error, error, track_id), - ) - conn.execute( - "INSERT INTO activity_log (event_type, track_id, message, details) VALUES (?, ?, ?, ?)", - ("download_failed", track_id, f"Download failed for track {track_id}", - json.dumps({"error": error})), - ) diff --git a/sync-api/services/matcher.py b/sync-api/services/matcher.py deleted file mode 100644 index c4e8e6c..0000000 --- a/sync-api/services/matcher.py +++ /dev/null @@ -1,157 +0,0 @@ -import os -import json -import httpx - -from db import get_db - -TIDARR_API = os.environ.get("TIDARR_URL", "http://localhost:8484") # optional legacy fallback - - -class MatcherService: - """Matches Spotify tracks to Tidal equivalents for lossless download. - - NOTE: This legacy matcher uses the Tidarr API for search. - Primary matching now uses Tidal API directly in the sync-worker. - """ - - def __init__(self): - self.tidarr_url = TIDARR_API - - async def match_track(self, track: dict) -> dict: - """ - Try to match a track. Returns dict with: - - matched: bool - - tidal_id: str or None - - confidence: float - - source: str (isrc|metadata|failed) - """ - # Strategy 1: ISRC lookup (highest confidence) - if track.get("isrc"): - result = await self._match_by_isrc(track["isrc"]) - if result: - return { - "matched": True, - "tidal_id": result["tidal_id"], - "confidence": 0.95, - "source": "isrc", - } - - # Strategy 2: Metadata search - result = await self._match_by_metadata(track) - if result: - return { - "matched": True, - "tidal_id": result["tidal_id"], - "confidence": result["confidence"], - "source": "metadata", - } - - return { - "matched": False, - "tidal_id": None, - "confidence": 0.0, - "source": "failed", - } - - async def _match_by_isrc(self, isrc: str) -> dict | None: - """Search Tidal by ISRC (via legacy Tidarr API).""" - try: - async with httpx.AsyncClient(timeout=15.0) as client: - resp = await client.get( - f"{self.tidarr_url}/api/search", - params={"query": isrc, "type": "track"}, - ) - if resp.status_code == 200: - data = resp.json() - tracks = data.get("tracks", data.get("items", [])) - if tracks: - return {"tidal_id": str(tracks[0].get("id", ""))} - except Exception: - pass - return None - - async def _match_by_metadata(self, track: dict) -> dict | None: - """Search by artist + title.""" - artist = track.get("artist", "") - title = track.get("title", "") - if not artist or not title: - return None - - query = f"{artist} {title}" - - try: - async with httpx.AsyncClient(timeout=15.0) as client: - resp = await client.get( - f"{self.tidarr_url}/api/search", - params={"query": query, "type": "track"}, - ) - if resp.status_code == 200: - data = resp.json() - tracks = data.get("tracks", data.get("items", [])) - if tracks: - best = tracks[0] - confidence = self._compute_confidence(track, best) - if confidence >= 0.7: - return { - "tidal_id": str(best.get("id", "")), - "confidence": confidence, - } - - # Log fallback attempt - with get_db() as conn: - conn.execute( - """INSERT INTO fallback_attempts - (track_id, source, status, search_query, result_count) - VALUES (?, 'tidal_metadata', 'no_match', ?, ?)""", - (track.get("id", 0), query, len(tracks)), - ) - except Exception: - pass - return None - - def _compute_confidence(self, spotify_track: dict, tidal_track: dict) -> float: - """Compute match confidence based on metadata similarity.""" - score = 0.0 - checks = 0 - - # Title comparison - sp_title = (spotify_track.get("title") or "").lower().strip() - ti_title = (tidal_track.get("title", tidal_track.get("name", ""))).lower().strip() - if sp_title and ti_title: - checks += 1 - if sp_title == ti_title: - score += 1.0 - elif sp_title in ti_title or ti_title in sp_title: - score += 0.7 - - # Artist comparison - sp_artist = (spotify_track.get("artist") or "").lower().strip() - ti_artist = "" - if "artist" in tidal_track: - if isinstance(tidal_track["artist"], dict): - ti_artist = tidal_track["artist"].get("name", "").lower().strip() - else: - ti_artist = str(tidal_track["artist"]).lower().strip() - elif "artists" in tidal_track: - names = [a.get("name", "") for a in tidal_track.get("artists", [])] - ti_artist = ", ".join(names).lower().strip() - - if sp_artist and ti_artist: - checks += 1 - if sp_artist == ti_artist: - score += 1.0 - elif sp_artist.split(",")[0].strip() in ti_artist: - score += 0.7 - - # Duration comparison (within 3 seconds) - sp_dur = spotify_track.get("duration_ms", 0) - ti_dur = tidal_track.get("duration", 0) * 1000 if tidal_track.get("duration") else 0 - if sp_dur and ti_dur: - checks += 1 - diff = abs(sp_dur - ti_dur) - if diff < 3000: - score += 1.0 - elif diff < 10000: - score += 0.5 - - return round(score / max(checks, 1), 2) diff --git a/sync-api/services/verifier.py b/sync-api/services/verifier.py deleted file mode 100644 index 24d0c38..0000000 --- a/sync-api/services/verifier.py +++ /dev/null @@ -1,146 +0,0 @@ -import json -import subprocess -import os - - -class VerifierService: - """Verifies downloaded audio files for lossless quality.""" - - async def verify_lossless(self, file_path: str) -> dict: - """ - Run ffprobe to check codec, sample rate, and bit depth. - Returns verification result dict. - """ - if not os.path.isfile(file_path): - return { - "status": "fail", - "error": f"File not found: {file_path}", - "codec": None, - "sample_rate": None, - "bit_depth": None, - } - - try: - result = subprocess.run( - [ - "ffprobe", "-v", "quiet", - "-print_format", "json", - "-show_streams", - "-select_streams", "a:0", - file_path, - ], - capture_output=True, text=True, timeout=30, - ) - - if result.returncode != 0: - return { - "status": "fail", - "error": f"ffprobe failed: {result.stderr}", - "codec": None, - "sample_rate": None, - "bit_depth": None, - } - - data = json.loads(result.stdout) - streams = data.get("streams", []) - if not streams: - return { - "status": "fail", - "error": "No audio streams found", - "codec": None, - "sample_rate": None, - "bit_depth": None, - } - - stream = streams[0] - codec = stream.get("codec_name", "") - sample_rate = int(stream.get("sample_rate", 0)) - bit_depth = int(stream.get("bits_per_raw_sample", stream.get("bits_per_sample", 0))) - - # Lossless codecs with >= 16-bit and >= 44100 Hz - lossless_codecs = ( - "flac", "alac", "wav", "aiff", - "pcm_s16le", "pcm_s24le", "pcm_s32le", - "pcm_s16be", "pcm_s24be", "pcm_s32be", - "pcm_f32le", "pcm_f64le", - ) - is_lossless = codec in lossless_codecs - is_hires = sample_rate >= 44100 and bit_depth >= 16 - - status = "pass" if (is_lossless and is_hires) else "fail" - - return { - "status": status, - "codec": codec, - "sample_rate": sample_rate, - "bit_depth": bit_depth, - "is_genuine_lossless": is_lossless and is_hires, - } - - except subprocess.TimeoutExpired: - return {"status": "fail", "error": "ffprobe timed out", "codec": None, "sample_rate": None, "bit_depth": None} - except Exception as e: - return {"status": "fail", "error": str(e), "codec": None, "sample_rate": None, "bit_depth": None} - - async def analyze_spectrum(self, file_path: str) -> dict: - """ - Spectral analysis to detect transcoded files (lossy re-encoded as lossless). - Uses sox to check for frequency cutoff typical of lossy codecs. - """ - if not os.path.isfile(file_path): - return {"suspicious": False, "error": f"File not found: {file_path}"} - - try: - result = subprocess.run( - ["sox", file_path, "-n", "stat"], - capture_output=True, text=True, timeout=60, - ) - - # sox stat outputs to stderr - stats_text = result.stderr - - # Parse maximum frequency — if it cuts off below 20kHz, likely transcoded - # This is a heuristic; full spectral analysis would need numpy/scipy - lines = stats_text.strip().split("\n") - stats = {} - for line in lines: - parts = line.split(":", 1) - if len(parts) == 2: - stats[parts[0].strip()] = parts[1].strip() - - return { - "suspicious": False, # Basic check — would need spectral peak analysis for real detection - "stats": stats, - } - - except subprocess.TimeoutExpired: - return {"suspicious": False, "error": "sox timed out"} - except Exception as e: - return {"suspicious": False, "error": str(e)} - - async def generate_fingerprint(self, file_path: str) -> dict: - """Generate a Chromaprint acoustic fingerprint.""" - if not os.path.isfile(file_path): - return {"fingerprint": None, "error": f"File not found: {file_path}"} - - try: - result = subprocess.run( - ["fpcalc", "-json", file_path], - capture_output=True, text=True, timeout=60, - ) - - if result.returncode != 0: - return {"fingerprint": None, "error": f"fpcalc failed: {result.stderr}"} - - data = json.loads(result.stdout) - return { - "fingerprint": data.get("fingerprint"), - "duration": data.get("duration"), - } - - except subprocess.TimeoutExpired: - return {"fingerprint": None, "error": "fpcalc timed out"} - except FileNotFoundError: - return {"fingerprint": None, "error": "fpcalc not found (install libchromaprint-tools)"} - except Exception as e: - return {"fingerprint": None, "error": str(e)} diff --git a/sync-worker/tasks/process_pipeline.py b/sync-worker/tasks/process_pipeline.py index 4d2c4ae..8abc3ba 100644 --- a/sync-worker/tasks/process_pipeline.py +++ b/sync-worker/tasks/process_pipeline.py @@ -2045,6 +2045,18 @@ def _process_organizing(db_path: str): # Soulseek fallback. Protects Will's lossless standard at the single chokepoint. if _reject_nonlossless_for_import(db_path, track): continue + # Replication gate: WaxFlow hands Lexicon a LOCAL Mac path, so the file must + # have reached the Mac before we import. Holding here (track stays in + # 'organizing', retried next cycle) is how we avoid the silent + # HTTP-200-imported-0-tracks failure. Fails open — see tasks/sync_gate.py. + try: + from tasks.sync_gate import is_replicated + ok, reason = is_replicated(db_path, track.get("file_path"), track) + if not ok: + log.info("organizing: holding track %d — %s", track["id"], reason) + continue + except Exception as e: + log.warning("sync_gate error (continuing on normal path): %s", e) try: _organize_track(db_path, track) synced_count += 1 @@ -2154,14 +2166,29 @@ def _container_to_mac_path( """Translate a worker-container file path into the path the Lexicon host Mac reads it by, so POST /v1/tracks points at a file Lexicon can actually open. - Delivery model (2026-07-11): the worker writes finished audio into container - ``/music`` (== NAS /volume1/music), which the Mac reads over the SMB mount at - ``lexicon_library_path`` (/Volumes/music) — LIVE, no sync lag. A separate - upload/staging flow lands in ``/downloads`` (== NAS /volume1/music/Input), - read at ``lexicon_input_path``. Map by container prefix: + Delivery model (2026-08-08) — LOCAL paths, not SMB: + + container /music/Database == NAS /volume1/music/Database + == Mac ~/Music/Database (one-way rsync replica) + container /downloads == NAS /volume1/music/Input + == Mac ~/Music/Input + + /downloads/ -> / + / -> / + + ``lexicon_library_path`` may be a LOCAL host path (e.g. ``~/Music/Database``) — a + path. This is the whole point: Engine DJ refuses ``/Volumes/*`` locations, so + every track imported under the previous SMB model (``/Volumes/music/...``) or + via the ``/Volumes/Macintosh HD/`` symlink was invisible to Engine export. Local + paths make ``scripts/repoint-lexicon-local.sh`` unnecessary for new tracks. + + The cost is replication lag, which ``tasks/sync_gate.py`` handles by holding the + import until the file has actually landed on the Mac. - /downloads/ -> / - /music/ -> / + NOTE: ``MUSIC_LIBRARY_PATH`` is ``/music/Database`` (not ``/music``). The bind + mount deliberately still points at the share ROOT so that Plex path translation + (plex_music_container_prefix=/music -> /volume1/music) and the 4,135 existing + ``/music/Database/...`` file_path rows all keep resolving unchanged. """ if downloads_dir and file_path.startswith(downloads_dir): relative_path = os.path.relpath(file_path, downloads_dir) diff --git a/sync-worker/tasks/sync_gate.py b/sync-worker/tasks/sync_gate.py new file mode 100644 index 0000000..0c63f3b --- /dev/null +++ b/sync-worker/tasks/sync_gate.py @@ -0,0 +1,154 @@ +"""Replication gate — hold a Lexicon import until the file exists on the Mac. + +WHY THIS EXISTS (2026-08-08) + WaxFlow can hand Lexicon a LOCAL host path (e.g. ~/Music/Database/...) instead + of an SMB path (/Volumes/music/...). Local paths are preferable because an SMB + path stops resolving the moment the share unmounts, and the file then exists + only on the NAS. + + (An earlier version of this note claimed Engine DJ refuses /Volumes paths. + Tested 2026-08-09: it does not. All 40 rows carrying a '/Volumes/Macintosh HD/' + prefix were present in the Engine library — that prefix is a symlink to / and + resolves fine. The missing-tracks symptom had a different cause: a two-way sync + destroying Engine's database.) + + The cost of local paths is replication lag. The worker writes to the NAS; a + one-way rsync agent on the Mac (scripts/sync-nas-to-mac.sh, every 120 s) pulls it + down. Between those two events the local path does not exist yet, and importing + then is exactly the silent failure mode WaxFlow already fights: Lexicon returns + HTTP 200 having imported 0 tracks (see lexicon_health.note_empty_import). + + So: before importing, check that the file predates the last completed sync pass. + The sync agent publishes a heartbeat into the shared Input/ directory, which the + worker sees at /downloads/.waxflow-sync-heartbeat. + +DESIGN NOTES + * FAIL OPEN. A missing, malformed, or stale heartbeat must never stall the + pipeline — if the sync agent dies, imports proceed and the existing + empty-import + import_catchup machinery handles any fallout. A gate that can + deadlock the pipeline is worse than the lag it prevents. + * BOUNDED HOLD. A track is held for at most sync_gate_max_hold_seconds; past + that we import anyway rather than spin forever on a file that will never + replicate (e.g. it lives in Database/Aktive, which the sync agent's SSH-side + change detection cannot see). + * NO NEW STATE. Held tracks simply stay in 'organizing' and are retried on the + next pipeline cycle (~10 s). No queue rows, no error state, nothing to drain. + +Config (all read live from app_config, no redeploy needed): + sync_gate_enabled default 1 + sync_gate_heartbeat_path default /downloads/.waxflow-sync-heartbeat + sync_gate_max_hold_seconds default 3600 + sync_gate_heartbeat_max_age_seconds default 1800 +""" + +from __future__ import annotations + +import json +import logging +import os +import sqlite3 +import time + +log = logging.getLogger(__name__) + +DEFAULT_HEARTBEAT = "/downloads/.waxflow-sync-heartbeat" +DEFAULT_MAX_HOLD = 3600 +DEFAULT_HEARTBEAT_MAX_AGE = 1800 + + +def get_config(db_path: str, key: str) -> str | None: + """Read one app_config value. + + Deliberately NOT tasks.helpers.get_config: importing helpers pulls in spotipy + and the whole Spotify client stack, which would make this leaf module + un-importable in a bare test environment. The read is three lines; the coupling + is not worth it. + """ + try: + conn = sqlite3.connect(db_path, timeout=10) + try: + row = conn.execute( + "SELECT value FROM app_config WHERE key = ?", (key,) + ).fetchone() + return row[0] if row else None + finally: + conn.close() + except sqlite3.Error: + return None + + +def _int_config(db_path: str, key: str, default: int) -> int: + try: + return int(str(get_config(db_path, key) or default).strip()) + except (TypeError, ValueError): + return default + + +def is_enabled(db_path: str) -> bool: + return str(get_config(db_path, "sync_gate_enabled") or "1").strip().lower() in ( + "1", "true", "yes", "on", + ) + + +def read_heartbeat(db_path: str) -> dict | None: + """Last sync-agent heartbeat, or None if absent/unparseable/stale.""" + path = (get_config(db_path, "sync_gate_heartbeat_path") or DEFAULT_HEARTBEAT).strip() + try: + with open(path, encoding="utf-8") as fh: + hb = json.load(fh) + except (OSError, ValueError): + return None + try: + completed = int(hb.get("completed_at", 0)) + except (TypeError, ValueError): + return None + if completed <= 0: + return None + max_age = _int_config(db_path, "sync_gate_heartbeat_max_age_seconds", + DEFAULT_HEARTBEAT_MAX_AGE) + if time.time() - completed > max_age: + return None # agent is dead or wedged -> fail open + hb["completed_at"] = completed + return hb + + +def is_replicated(db_path: str, file_path: str | None, track: dict | None = None) -> tuple[bool, str]: + """Has ``file_path`` had time to reach the Mac? + + Returns (ok_to_import, reason). ``ok_to_import`` is True whenever we cannot + prove the file is still in flight — see FAIL OPEN above. + """ + if not is_enabled(db_path): + return True, "gate disabled" + if not file_path: + return True, "no file_path" + + hb = read_heartbeat(db_path) + if hb is None: + return True, "no usable sync heartbeat (failing open)" + if hb.get("status") != "ok": + return True, f"sync heartbeat status={hb.get('status')} (failing open)" + + try: + mtime = os.path.getmtime(file_path) + except OSError: + # The worker cannot see its own file; that is a different problem and the + # normal import path will surface it. + return True, "file not stat-able by worker" + + if mtime <= hb["completed_at"]: + return True, "replicated" + + # Still in flight — but do not hold forever. + max_hold = _int_config(db_path, "sync_gate_max_hold_seconds", DEFAULT_MAX_HOLD) + if time.time() - mtime > max_hold: + log.warning( + "sync_gate: %s still not replicated after %ss — importing anyway", + file_path, max_hold, + ) + return True, f"max hold {max_hold}s exceeded" + + return False, ( + f"awaiting replication (file {int(time.time() - mtime)}s old, " + f"last sync {int(time.time() - hb['completed_at'])}s ago)" + ) diff --git a/sync-worker/tests/test_sync_gate.py b/sync-worker/tests/test_sync_gate.py new file mode 100644 index 0000000..0fc346b --- /dev/null +++ b/sync-worker/tests/test_sync_gate.py @@ -0,0 +1,147 @@ +"""Tests for the replication gate (tasks/sync_gate.py). + +The gate exists because WaxFlow now hands Lexicon a LOCAL Mac path, so an import +fired before the one-way rsync has landed the file produces Lexicon's silent +HTTP-200-imported-0-tracks failure. + +The contract these tests pin down: + * a file older than the last completed sync pass imports immediately, + * a file newer than it is HELD (still in flight), + * every degenerate heartbeat case FAILS OPEN — missing, malformed, stale, or + status!=ok must never stall the pipeline, because a gate that can deadlock is + worse than the lag it prevents, + * the hold is bounded: past sync_gate_max_hold_seconds we import anyway rather + than spin forever on a file that will never replicate, + * the whole gate can be switched off live via sync_gate_enabled. +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import time +import unittest + +SYNC_WORKER_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if SYNC_WORKER_DIR not in sys.path: + sys.path.insert(0, SYNC_WORKER_DIR) + +from tasks import sync_gate # noqa: E402 + + +def _db(**config) -> str: + path = tempfile.mktemp(suffix=".db") + conn = sqlite3.connect(path) + conn.executescript( + "CREATE TABLE app_config (key TEXT PRIMARY KEY, value TEXT);" + ) + for k, v in config.items(): + conn.execute("INSERT INTO app_config (key, value) VALUES (?, ?)", (k, str(v))) + conn.commit() + conn.close() + return path + + +def _heartbeat(tmpdir: str, completed_at: float, status: str = "ok") -> str: + p = os.path.join(tmpdir, ".waxflow-sync-heartbeat") + with open(p, "w", encoding="utf-8") as fh: + json.dump({"status": status, "completed_at": int(completed_at)}, fh) + return p + + +def _audio(tmpdir: str, mtime: float) -> str: + p = os.path.join(tmpdir, "track.flac") + with open(p, "wb") as fh: + fh.write(b"\0") + os.utime(p, (mtime, mtime)) + return p + + +class SyncGateTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.now = time.time() + + def test_file_older_than_last_sync_is_replicated(self): + hb = _heartbeat(self.tmp, self.now - 60) + f = _audio(self.tmp, self.now - 600) # written well before the sync + db = _db(sync_gate_heartbeat_path=hb) + ok, reason = sync_gate.is_replicated(db, f) + self.assertTrue(ok) + self.assertEqual(reason, "replicated") + + def test_file_newer_than_last_sync_is_held(self): + hb = _heartbeat(self.tmp, self.now - 300) + f = _audio(self.tmp, self.now - 10) # written after the last pass + db = _db(sync_gate_heartbeat_path=hb) + ok, reason = sync_gate.is_replicated(db, f) + self.assertFalse(ok) + self.assertIn("awaiting replication", reason) + + def test_missing_heartbeat_fails_open(self): + f = _audio(self.tmp, self.now) + db = _db(sync_gate_heartbeat_path=os.path.join(self.tmp, "nope.json")) + ok, reason = sync_gate.is_replicated(db, f) + self.assertTrue(ok) + self.assertIn("no usable sync heartbeat", reason) + + def test_malformed_heartbeat_fails_open(self): + p = os.path.join(self.tmp, "bad.json") + with open(p, "w", encoding="utf-8") as fh: + fh.write("{not json") + f = _audio(self.tmp, self.now) + ok, _ = sync_gate.is_replicated(_db(sync_gate_heartbeat_path=p), f) + self.assertTrue(ok) + + def test_stale_heartbeat_fails_open(self): + # agent died an hour ago; max age is 30 min -> treat as no heartbeat + hb = _heartbeat(self.tmp, self.now - 3600) + f = _audio(self.tmp, self.now) + db = _db(sync_gate_heartbeat_path=hb, sync_gate_heartbeat_max_age_seconds=1800) + ok, reason = sync_gate.is_replicated(db, f) + self.assertTrue(ok) + self.assertIn("no usable sync heartbeat", reason) + + def test_error_status_heartbeat_fails_open(self): + hb = _heartbeat(self.tmp, self.now - 60, status="error") + f = _audio(self.tmp, self.now) + ok, reason = sync_gate.is_replicated(_db(sync_gate_heartbeat_path=hb), f) + self.assertTrue(ok) + self.assertIn("status=error", reason) + + def test_hold_is_bounded_by_max_hold_seconds(self): + # file is newer than the last sync AND older than the max hold -> import + hb = _heartbeat(self.tmp, self.now - 7200) + # keep the heartbeat itself fresh enough to be usable + db = _db(sync_gate_heartbeat_path=hb, + sync_gate_heartbeat_max_age_seconds=99999, + sync_gate_max_hold_seconds=600) + f = _audio(self.tmp, self.now - 3600) + ok, reason = sync_gate.is_replicated(db, f) + self.assertTrue(ok) + self.assertIn("max hold", reason) + + def test_disabled_gate_always_allows(self): + hb = _heartbeat(self.tmp, self.now - 300) + f = _audio(self.tmp, self.now) + db = _db(sync_gate_heartbeat_path=hb, sync_gate_enabled="0") + ok, reason = sync_gate.is_replicated(db, f) + self.assertTrue(ok) + self.assertEqual(reason, "gate disabled") + + def test_no_file_path_allows(self): + ok, reason = sync_gate.is_replicated(_db(), None) + self.assertTrue(ok) + self.assertEqual(reason, "no file_path") + + def test_unstatable_file_fails_open(self): + hb = _heartbeat(self.tmp, self.now - 60) + db = _db(sync_gate_heartbeat_path=hb) + ok, reason = sync_gate.is_replicated(db, os.path.join(self.tmp, "ghost.flac")) + self.assertTrue(ok) + self.assertIn("not stat-able", reason) + + +if __name__ == "__main__": + unittest.main()