fix(jobs): atomic dot-path writes for file_urls/state_data (lost-update race)

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:

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

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"])
 

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."""

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}")
 

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

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "karaoke-gen"
-version = "0.217.0"
+version = "0.217.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 <andrew@beveridge.uk>"]
 license = "MIT"