From 79b5c42a18805478ba3e0d33e64484b175364184 Mon Sep 17 00:00:00 2001 From: Andrew Beveridge Date: Mon, 31 Aug 2026 18:12:41 -0400 Subject: [PATCH] fix(jobs): atomic dot-path writes for file_urls/state_data (lost-update race) update_file_url and update_state_data copied the whole map and re-persisted it, so the parallel audio + screens workers could clobber each other's writes. Real symptom: a backing_vocals stem uploaded to GCS but dropped from file_urls.stems, so the instrumental-review UI couldn't load the backing preview and _analyze_backing_vocals reported 0% backing. - Write single Firestore dot-path fields (file_urls.., state_data.) so siblings merge server-side, matching update_processing_metadata. - _analyze_backing_vocals recovers a stem present at the conventional GCS path but missing from file_urls, and stores a fallback analysis instead of a silent early-return. - Unit tests for dot-path writes + emulator regression tests that fail against the old read-modify-write and pass against the fix. Co-Authored-By: Claude Opus 4.8 --- backend/services/job_manager.py | 26 ++-- .../emulator/test_file_urls_no_clobber.py | 111 ++++++++++++++++++ backend/tests/test_job_manager.py | 78 +++++++++++- backend/tests/test_workers.py | 49 ++++++-- backend/workers/screens_worker.py | 27 ++++- docs/LESSONS-LEARNED.md | 21 ++++ pyproject.toml | 2 +- 7 files changed, 290 insertions(+), 24 deletions(-) create mode 100644 backend/tests/emulator/test_file_urls_no_clobber.py diff --git a/backend/services/job_manager.py b/backend/services/job_manager.py index 63760e2f7..31293df58 100644 --- a/backend/services/job_manager.py +++ b/backend/services/job_manager.py @@ -789,10 +789,12 @@ def update_state_data(self, job_id: str, key: str, value: Any) -> None: logger.error(f"Job {job_id} not found") return - state_data = job.state_data.copy() - state_data[key] = value - - self.update_job(job_id, {'state_data': state_data}) + # Atomic nested-field write. Previously this copied the whole state_data + # map and re-persisted it, so two workers updating different keys + # concurrently could clobber each other (lost-update race). Writing a + # single dot-path field merges without touching sibling keys — same + # approach as update_processing_metadata. + self.update_job(job_id, {f"state_data.{key}": value}) logger.debug(f"Job {job_id} state_data updated: {key} = {value}") def bump_worker_generation(self, job_id: str) -> Optional[int]: @@ -1013,14 +1015,14 @@ def update_file_url(self, job_id: str, category: str, file_type: str, url: str) if not job: logger.error(f"Job {job_id} not found") return - - file_urls = job.file_urls.copy() - if category not in file_urls: - file_urls[category] = {} - - file_urls[category][file_type] = url - - self.update_job(job_id, {'file_urls': file_urls}) + + # Atomic nested-field write. Previously this copied the whole file_urls + # map and re-persisted it, so two workers registering different files + # concurrently could clobber each other. That lost-update race dropped + # freshly-registered stems (e.g. backing_vocals vanishing while the + # audio + screens workers ran in parallel). Writing a single dot-path + # field merges without touching sibling entries. + self.update_job(job_id, {f"file_urls.{category}.{file_type}": url}) logger.debug(f"Job {job_id} file URL updated: {category}.{file_type}") def check_parallel_processing_complete(self, job_id: str) -> bool: diff --git a/backend/tests/emulator/test_file_urls_no_clobber.py b/backend/tests/emulator/test_file_urls_no_clobber.py new file mode 100644 index 000000000..22871ae2f --- /dev/null +++ b/backend/tests/emulator/test_file_urls_no_clobber.py @@ -0,0 +1,111 @@ +""" +Regression test for the file_urls / state_data lost-update race. + +Reproduces the real-world failure where a `backing_vocals` stem was uploaded to +GCS but vanished from the job's ``file_urls`` because two workers registered +different stems concurrently, each re-persisting the whole map from a stale +snapshot. The fix writes atomic dot-path fields, which Firestore merges +server-side. + +To reproduce the race deterministically we hand the *second* writer a stale +snapshot (as if it had read the job before the first writer's value landed) by +patching ``get_job`` for that one call. Against the old copy-and-rewrite +implementation the second write re-persists the stale map and drops the first +writer's value; against the field-path write it cannot. These run on the real +Firestore emulator. + +Run with: scripts/run-emulator-tests.sh +""" + +import pytest +from copy import deepcopy +from datetime import datetime, UTC +from unittest.mock import patch + +from backend.tests.emulator.conftest import emulators_running + +pytestmark = pytest.mark.skipif( + not emulators_running(), + reason="GCP emulators not running. Start with: scripts/start-emulators.sh" +) + +if emulators_running(): + from backend.models.job import Job, JobStatus + from backend.services.job_manager import JobManager + from backend.services.firestore_service import FirestoreService + + +class TestFileUrlsNoClobber: + @pytest.fixture + def job_manager(self): + return JobManager() + + @pytest.fixture + def firestore_service(self): + return FirestoreService() + + def _create_job(self, firestore_service, job_id, file_urls=None, state_data=None): + firestore_service.create_job(Job( + job_id=job_id, + status=JobStatus.SEPARATING_STAGE1, + created_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + artist="Test", + title="Test", + file_urls=file_urls or {}, + state_data=state_data or {}, + )) + + def test_stale_registration_does_not_drop_sibling_stem(self, job_manager, firestore_service): + """The backing_vocals-vanishing bug: a second stem registration working + from a stale snapshot must not clobber a stem written in between.""" + job_id = f"noclobber-stems-{datetime.now(UTC).timestamp()}" + try: + self._create_job( + firestore_service, job_id, + file_urls={"stems": {"lead_vocals": "jobs/x/stems/lead_vocals.flac"}}, + ) + # Snapshot as it looked BEFORE backing_vocals was registered. + stale = deepcopy(firestore_service.get_job(job_id)) + + # Writer 1 registers backing_vocals (lands in Firestore). + job_manager.update_file_url(job_id, "stems", "backing_vocals", "jobs/x/stems/backing_vocals.flac") + + # Writer 2 registers a different stem but only "sees" the stale + # snapshot (missing backing_vocals). The field-path write ignores the + # snapshot contents, so backing_vocals survives. + with patch.object(job_manager, "get_job", return_value=stale): + job_manager.update_file_url(job_id, "stems", "instrumental_with_backing", "jobs/x/stems/instrumental_with_backing.flac") + + stems = firestore_service.get_job(job_id).file_urls.get("stems", {}) + assert stems.get("lead_vocals") == "jobs/x/stems/lead_vocals.flac" + assert stems.get("backing_vocals") == "jobs/x/stems/backing_vocals.flac" + assert stems.get("instrumental_with_backing") == "jobs/x/stems/instrumental_with_backing.flac" + finally: + try: + firestore_service.delete_job(job_id) + except Exception: + pass + + def test_stale_write_does_not_drop_sibling_state_data_key(self, job_manager, firestore_service): + """Same lost-update guard for state_data: a stale write must not drop a + sibling key set in between.""" + job_id = f"noclobber-state-{datetime.now(UTC).timestamp()}" + try: + self._create_job(firestore_service, job_id, state_data={"lyrics_complete": True}) + stale = deepcopy(firestore_service.get_job(job_id)) + + job_manager.update_state_data(job_id, "backing_vocals_analysis", {"has_audible_content": True}) + + with patch.object(job_manager, "get_job", return_value=stale): + job_manager.update_state_data(job_id, "audio_complete", True) + + state = firestore_service.get_job(job_id).state_data + assert state.get("lyrics_complete") is True + assert state.get("backing_vocals_analysis") == {"has_audible_content": True} + assert state.get("audio_complete") is True + finally: + try: + firestore_service.delete_job(job_id) + except Exception: + pass diff --git a/backend/tests/test_job_manager.py b/backend/tests/test_job_manager.py index 982c24475..165bb708e 100644 --- a/backend/tests/test_job_manager.py +++ b/backend/tests/test_job_manager.py @@ -911,10 +911,12 @@ async def test_skips_audio_worker_when_existing_instrumental(self, job_manager, # Lyrics worker should still be triggered mock_worker.trigger_lyrics_worker.assert_called_once_with("test123") - # audio_complete should be set via update_job (update_state_data calls update_job) + # audio_complete should be set via update_state_data, which now writes + # an atomic dot-path field ("state_data.audio_complete") rather than + # re-persisting the whole state_data map. update_calls = mock_firestore_service.update_job.call_args_list audio_complete_set = any( - call.args[1].get('state_data', {}).get('audio_complete') is True + call.args[1].get('state_data.audio_complete') is True for call in update_calls if len(call.args) > 1 and isinstance(call.args[1], dict) ) @@ -988,6 +990,78 @@ async def test_workers_not_triggered_when_reconcile_pauses(self, job_manager, mo mock_worker.trigger_lyrics_worker.assert_not_called() +class TestAtomicNestedUpdates: + """update_state_data / update_file_url must write atomic dot-path fields. + + Previously both copied the whole map and re-persisted it, so two workers + updating different keys concurrently could clobber each other's writes + (lost-update race — e.g. a backing_vocals stem vanishing while the audio and + screens workers ran in parallel). Writing a single nested field path lets + Firestore merge sibling entries server-side. + """ + + def _job(self, job_id="job-1", file_urls=None, state_data=None): + return Job( + job_id=job_id, + status=JobStatus.SEPARATING_STAGE1, + created_at=datetime.now(UTC), + updated_at=datetime.now(UTC), + file_urls=file_urls or {}, + state_data=state_data or {}, + ) + + def test_update_state_data_writes_dotted_field(self, job_manager, mock_firestore_service): + mock_firestore_service.get_job.return_value = self._job() + + job_manager.update_state_data("job-1", "audio_complete", True) + + mock_firestore_service.update_job.assert_called_once() + args = mock_firestore_service.update_job.call_args.args + assert args[0] == "job-1" + assert args[1] == {"state_data.audio_complete": True} + # Must NOT re-persist the whole map. + assert "state_data" not in args[1] + + def test_update_file_url_writes_dotted_field(self, job_manager, mock_firestore_service): + mock_firestore_service.get_job.return_value = self._job() + + job_manager.update_file_url("job-1", "stems", "backing_vocals", "jobs/job-1/stems/backing_vocals.flac") + + mock_firestore_service.update_job.assert_called_once() + args = mock_firestore_service.update_job.call_args.args + assert args[1] == {"file_urls.stems.backing_vocals": "jobs/job-1/stems/backing_vocals.flac"} + assert "file_urls" not in args[1] + + def test_concurrent_stem_registrations_do_not_clobber(self, job_manager, mock_firestore_service): + """Two stem registrations issue independent field-path writes. + + With the old copy-and-rewrite approach, a registration built from a stale + snapshot would drop a sibling stem written in between. Independent dot-path + writes target different fields, so Firestore never loses either. + """ + # Both callers see the same stale snapshot (missing each other's write). + mock_firestore_service.get_job.return_value = self._job( + file_urls={"stems": {"lead_vocals": "jobs/job-1/stems/lead_vocals.flac"}} + ) + + job_manager.update_file_url("job-1", "stems", "backing_vocals", "bv.flac") + job_manager.update_file_url("job-1", "stems", "instrumental_with_backing", "iwb.flac") + + writes = [c.args[1] for c in mock_firestore_service.update_job.call_args_list] + assert {"file_urls.stems.backing_vocals": "bv.flac"} in writes + assert {"file_urls.stems.instrumental_with_backing": "iwb.flac"} in writes + # Neither write rewrites the whole map, so neither can drop the other. + assert all("file_urls" not in w for w in writes) + + def test_missing_job_is_a_noop(self, job_manager, mock_firestore_service): + mock_firestore_service.get_job.return_value = None + + job_manager.update_state_data("gone", "k", "v") + job_manager.update_file_url("gone", "stems", "backing_vocals", "x.flac") + + mock_firestore_service.update_job.assert_not_called() + + if __name__ == "__main__": pytest.main([__file__, "-v"]) diff --git a/backend/tests/test_workers.py b/backend/tests/test_workers.py index aa2e00203..8269e087d 100644 --- a/backend/tests/test_workers.py +++ b/backend/tests/test_workers.py @@ -933,29 +933,64 @@ async def test_analyze_backing_vocals_handles_missing_job(self): mock_job_manager.update_state_data.assert_not_called() @pytest.mark.asyncio - async def test_analyze_backing_vocals_handles_missing_stems(self): - """Test that analysis returns early when stems not found (no error stored).""" + async def test_analyze_backing_vocals_stores_fallback_when_stem_truly_absent(self): + """When the stem is registered nowhere AND absent from GCS, record a + fallback analysis so the review UI can distinguish 'genuinely no backing + vocals' from 'analysis never ran' — instead of silently implying 0%.""" mock_job = MagicMock() - mock_job.file_urls = {} # No stems + mock_job.file_urls = {} # No stems registered mock_job_manager = MagicMock() mock_job_manager.get_job.return_value = mock_job mock_job_manager.update_state_data = MagicMock() mock_storage = MagicMock() + mock_storage.file_exists.return_value = False # Not recoverable from GCS mock_logger = MagicMock() from backend.workers.screens_worker import _analyze_backing_vocals - # Should not raise when stems not found, just log warning and return await _analyze_backing_vocals( "test123", mock_job_manager, mock_storage, mock_logger ) - # Should log warning but not update state (early return) mock_logger.warning.assert_called() - # No state_data update since we return early - mock_job_manager.update_state_data.assert_not_called() + # A fallback analysis is stored (not a silent early return), and no stem + # is re-registered because none exists. + mock_job_manager.update_state_data.assert_called_once() + args = mock_job_manager.update_state_data.call_args.args + assert args[1] == "backing_vocals_analysis" + assert args[2].get("analysis_error") + mock_job_manager.update_file_url.assert_not_called() + + @pytest.mark.asyncio + async def test_analyze_backing_vocals_recovers_stem_missing_from_file_urls(self): + """If the backing_vocals stem exists in GCS but was dropped from file_urls + (the lost-update race), recover by re-registering it from the conventional + path so the review UI can load it and analysis can run.""" + mock_job = MagicMock() + mock_job.file_urls = {"stems": {"lead_vocals": "jobs/test123/stems/lead_vocals.flac"}} + + mock_job_manager = MagicMock() + mock_job_manager.get_job.return_value = mock_job + + mock_storage = MagicMock() + mock_storage.file_exists.return_value = True # Present in GCS + mock_logger = MagicMock() + + # Patch the analysis service so the test asserts the recovery, not the + # (heavy) downstream waveform analysis. + with patch("backend.services.audio_analysis_service.AudioAnalysisService") as MockSvc: + MockSvc.return_value.analyze_and_generate_waveform.side_effect = RuntimeError("stop") + from backend.workers.screens_worker import _analyze_backing_vocals + await _analyze_backing_vocals( + "test123", mock_job_manager, mock_storage, mock_logger + ) + + # The dropped stem is re-registered atomically from the conventional path. + mock_job_manager.update_file_url.assert_called_once_with( + "test123", "stems", "backing_vocals", "jobs/test123/stems/backing_vocals.flac" + ) def test_analysis_service_can_be_instantiated(self): """Test that AudioAnalysisService can be instantiated.""" diff --git a/backend/workers/screens_worker.py b/backend/workers/screens_worker.py index cf8cbb75f..9a86c64a7 100644 --- a/backend/workers/screens_worker.py +++ b/backend/workers/screens_worker.py @@ -592,8 +592,31 @@ async def _analyze_backing_vocals( stems = job.file_urls.get('stems', {}) backing_vocals_path = stems.get('backing_vocals') if not backing_vocals_path: - job_log.warning("No backing vocals file found - skipping analysis") - return + # The stem may exist in GCS but be missing from file_urls if a + # concurrent write dropped its registration (the lost-update race + # this change also fixes at the source). Recover by checking the + # conventional path and re-registering it so the review UI can load + # the preview and the analysis below can run. + conventional_path = f"jobs/{job_id}/stems/backing_vocals.flac" + if storage.file_exists(conventional_path): + backing_vocals_path = conventional_path + stems['backing_vocals'] = conventional_path + job_manager.update_file_url(job_id, 'stems', 'backing_vocals', conventional_path) + job_log.warning( + "backing_vocals missing from file_urls but present in GCS " + "— re-registered from the conventional path" + ) + else: + job_log.warning("No backing vocals file found - skipping analysis") + # Record that analysis was attempted with no stem, so the review + # UI can distinguish "genuinely no backing vocals" from "analysis + # never ran" instead of silently implying 0% backing. + job_manager.update_state_data(job_id, 'backing_vocals_analysis', { + 'has_audible_content': None, + 'analysis_error': 'backing_vocals stem not found', + 'recommended_selection': 'clean', + }) + return job_log.info(f"Analyzing backing vocals: {backing_vocals_path}") diff --git a/docs/LESSONS-LEARNED.md b/docs/LESSONS-LEARNED.md index 0801ffeaa..a53b937b7 100644 --- a/docs/LESSONS-LEARNED.md +++ b/docs/LESSONS-LEARNED.md @@ -6,6 +6,27 @@ Key insights for future AI agents working on this codebase. --- +## Update Firestore Job Maps with Dot-Path Fields, Never Read-Modify-Write (Aug 2026) + +`JobManager.update_file_url` and `update_state_data` used to `get_job()`, copy the +whole `file_urls`/`state_data` map, set one key, and re-persist the entire map. The +audio and screens workers run **in parallel** and both write these maps, so a write +built from a stale snapshot silently dropped a sibling key set in between (a classic +lost update). Real symptom: a `backing_vocals.flac` stem uploaded to GCS but **missing +from `file_urls.stems`** → the instrumental-review UI couldn't load the backing preview, +and `_analyze_backing_vocals` early-returned ("0% backing detected"). Rare — needs a +precise interleaving (~1 in 4 parallel jobs). + +**Fix / rule:** write a single Firestore dot-path field so the server merges siblings — +`update_job(job_id, {f"file_urls.{category}.{file_type}": url})`, exactly like +`update_processing_metadata` already did. Any code that does read-copy-mutate-write on a +nested job map is exposed to the same race. Keys must be plain identifiers (no dots), or +the path is misread as further nesting. Belt-and-suspenders: `_analyze_backing_vocals` +now recovers a stem that exists at the conventional GCS path but is unregistered, and +stores a fallback analysis instead of a silent early-return. + +--- + ## Measure Human Edits by RECONSTRUCTION, Not the edit_log (Aug 2026) The lyrics-review `edit_log_*.json` (typed frontend ops) is **unreliable** as a diff --git a/pyproject.toml b/pyproject.toml index 55e3ba915..f9845dfd9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "karaoke-gen" -version = "0.216.0" +version = "0.216.1" description = "Generate karaoke videos with synchronized lyrics. Handles the entire process from downloading audio and lyrics to creating the final video with title screens." authors = ["Andrew Beveridge "] license = "MIT"