diff --git a/moss_transcribe_diarize/app/jobs.py b/moss_transcribe_diarize/app/jobs.py index e1c7f7f..271d1f0 100644 --- a/moss_transcribe_diarize/app/jobs.py +++ b/moss_transcribe_diarize/app/jobs.py @@ -159,6 +159,7 @@ def __init__( self.temperature = temperature self._jobs: dict[str, JobRecord] = {} self._queue: queue.Queue[str] = queue.Queue() + self._state_lock = threading.RLock() self._render_lock = threading.Lock() self._progress_save_times: dict[str, float] = {} self._load_existing_jobs() @@ -305,15 +306,21 @@ def update_segments( style_payload: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: job = self.get_job(job_id) - segments = coerce_subtitle_segments(payload) - style = SubtitleStyle.from_dict(style_payload) if style_payload is not None else None - if style is not None: - job.subtitle_style = style.to_dict() - self._write_subtitle_files(job, segments, style=style) - if job.status == "done": - self._set_status(job, "waiting_review", 0.95, error=None) - else: - self._touch(job, error=None) + with self._state_lock: + if job.status == "rendering": + raise JobManagerError( + "job_running", + "Cannot update subtitles while the job is rendering.", + ) + segments = coerce_subtitle_segments(payload) + style = SubtitleStyle.from_dict(style_payload) if style_payload is not None else None + if style is not None: + job.subtitle_style = style.to_dict() + self._write_subtitle_files(job, segments, style=style) + if job.status == "done": + self._set_status(job, "waiting_review", 0.95, error=None) + else: + self._touch(job, error=None) return [segment.to_dict() for segment in segments] def render(self, job_id: str, style_payload: dict[str, Any] | None = None) -> JobRecord: @@ -322,12 +329,25 @@ def render(self, job_id: str, style_payload: dict[str, Any] | None = None) -> Jo raise JobManagerError("ffmpeg_unavailable", "ffmpeg and ffprobe are not available on PATH.") if not job.segments_path.exists(): raise JobManagerError("subtitles_unavailable", "No subtitle segments are available for this job.") - threading.Thread( + style = SubtitleStyle.from_dict(style_payload) + thread = threading.Thread( target=self._render_job, - args=(job.id, SubtitleStyle.from_dict(style_payload)), + args=(job.id, style), name=f"mtd-render-{job.id}", daemon=True, - ).start() + ) + with self._state_lock: + if job.status == "rendering": + raise JobManagerError("job_running", "This job is already rendering.") + previous_status = job.status + previous_progress = job.progress + previous_error = job.error + self._set_status(job, "rendering", 0.97, error=None) + try: + thread.start() + except Exception: + self._set_status(job, previous_status, previous_progress, error=previous_error) + raise return job def download_path(self, job_id: str, kind: str) -> Path: @@ -362,7 +382,13 @@ def _load_existing_jobs(self) -> None: job = JobRecord.from_dict(data) if not job.job_dir: job.job_dir = str(path.parent) - if job.status in {"queued", "loading_model", "transcribing", "postprocessing", "rendering"}: + if job.status == "rendering" and self._has_complete_segments(job): + job.status = "waiting_review" + job.progress = 0.95 + job.error = "Rendering was interrupted by the previous server shutdown. You can retry rendering." + job.updated_at = time.time() + self._save_job(job) + elif job.status in {"queued", "loading_model", "transcribing", "postprocessing", "rendering"}: job.status = "failed" job.progress = 1.0 job.error = "Interrupted by previous server shutdown." @@ -372,6 +398,17 @@ def _load_existing_jobs(self) -> None: except Exception: continue + @staticmethod + def _has_complete_segments(job: JobRecord) -> bool: + try: + payload = json.loads(job.segments_path.read_text(encoding="utf-8")) + if not isinstance(payload, list): + return False + coerce_subtitle_segments(payload) + except (OSError, TypeError, ValueError, KeyError, AttributeError): + return False + return True + def _process_job(self, job: JobRecord) -> None: try: def update(status: str, progress: float | None, generated_tokens: int | None = None) -> None: @@ -408,7 +445,6 @@ def _render_job(self, job_id: str, style: SubtitleStyle) -> None: job = self.get_job(job_id) with self._render_lock: try: - self._set_status(job, "rendering", 0.97, error=None) segments = [SubtitleSegment.from_dict(item) for item in self.list_segments(job.id)] width, height = probe_video_size(job.input_path) write_text(job.ass_path, export_ass(segments, style=style, video_width=width, video_height=height)) @@ -448,9 +484,10 @@ def _set_status( error: str | None = None, save: bool = True, ) -> None: - job.status = status - job.progress = max(0.0, min(1.0, progress)) - self._touch(job, error=error, save=save) + with self._state_lock: + job.status = status + job.progress = max(0.0, min(1.0, progress)) + self._touch(job, error=error, save=save) def _resolve_inference_options( self, diff --git a/moss_transcribe_diarize/app/static/app.js b/moss_transcribe_diarize/app/static/app.js index ce6f784..87f9bee 100644 --- a/moss_transcribe_diarize/app/static/app.js +++ b/moss_transcribe_diarize/app/static/app.js @@ -8,7 +8,7 @@ import { } from './i18n.js'; const RUNNING_STATES = new Set(['queued', 'loading_model', 'transcribing', 'postprocessing', 'rendering']); -const EDIT_STATES = new Set(['waiting_review', 'rendering', 'done']); +const EDIT_STATES = new Set(['waiting_review', 'done']); const TERMINAL_STATES = new Set(['waiting_review', 'done', 'failed', 'cancelled']); const fileInput = document.querySelector('#file'); const importTitleEl = document.querySelector('#importTitle'); diff --git a/tests/test_app_api.py b/tests/test_app_api.py index dc78b2a..308c8d7 100644 --- a/tests/test_app_api.py +++ b/tests/test_app_api.py @@ -215,6 +215,151 @@ def test_running_job_exposes_live_token_progress(self): self.assertEqual(finished["status"], "waiting_review") self.assertEqual(finished["usage"]["generated_tokens"], 5) + def test_rendering_job_rejects_subtitle_updates(self): + from fastapi.testclient import TestClient + from moss_transcribe_diarize.app.server import create_app + + with tempfile.TemporaryDirectory() as tmpdir: + app = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8) + app.state.manager.model_runner = FakeRunner() + client = TestClient(app) + + created = client.post( + "/api/jobs", + files={"file": ("sample.wav", b"audio", "audio/wav")}, + ) + job_id = created.json()["id"] + for _ in range(40): + job = client.get(f"/api/jobs/{job_id}").json() + if job["status"] == "waiting_review": + break + time.sleep(0.05) + self.assertEqual(job["status"], "waiting_review") + + original = client.get(f"/api/jobs/{job_id}/segments").json()["segments"] + edited = [dict(item) for item in original] + edited[0]["text"] = "must not be saved" + app.state.manager._set_status( + app.state.manager.get_job(job_id), "rendering", 0.97, error=None + ) + + updated = client.put( + f"/api/jobs/{job_id}/segments", + json={"segments": edited}, + ) + + self.assertEqual(updated.status_code, 409) + self.assertIn("while the job is rendering", updated.json()["detail"]) + self.assertEqual( + client.get(f"/api/jobs/{job_id}/segments").json()["segments"], + original, + ) + self.assertIn( + "const EDIT_STATES = new Set(['waiting_review', 'done']);", + client.get("/assets/app.js").text, + ) + + def test_duplicate_render_request_is_rejected_before_starting_worker(self): + from fastapi.testclient import TestClient + from moss_transcribe_diarize.app.server import create_app + + with tempfile.TemporaryDirectory() as tmpdir: + app = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8) + app.state.manager.model_runner = FakeRunner() + client = TestClient(app) + + created = client.post( + "/api/jobs", + files={"file": ("sample.wav", b"audio", "audio/wav")}, + ) + job_id = created.json()["id"] + for _ in range(40): + job = client.get(f"/api/jobs/{job_id}").json() + if job["status"] == "waiting_review": + break + time.sleep(0.05) + self.assertEqual(job["status"], "waiting_review") + + class Available: + available = True + + render_started = threading.Event() + release_render = threading.Event() + + def block_render(*_args): + render_started.set() + release_render.wait(timeout=2) + + with ( + patch("moss_transcribe_diarize.app.jobs.detect_ffmpeg", return_value=Available()), + patch.object(app.state.manager, "_render_job", side_effect=block_render) as render_job, + ): + first = client.post(f"/api/jobs/{job_id}/render", json={"style": {}}) + self.assertEqual(first.status_code, 200) + self.assertEqual(first.json()["status"], "rendering") + self.assertTrue(render_started.wait(timeout=2)) + + duplicate = client.post(f"/api/jobs/{job_id}/render", json={"style": {}}) + + self.assertEqual(duplicate.status_code, 409) + self.assertIn("already rendering", duplicate.json()["detail"]) + self.assertEqual(render_job.call_count, 1) + release_render.set() + + def test_restart_recovers_interrupted_render_for_review(self): + from fastapi.testclient import TestClient + from moss_transcribe_diarize.app.server import create_app + + with tempfile.TemporaryDirectory() as tmpdir: + app = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8) + app.state.manager.model_runner = FakeRunner() + client = TestClient(app) + + job_ids = [] + for name in ("rendering.wav", "transcribing.wav"): + created = client.post( + "/api/jobs", + files={"file": (name, b"audio", "audio/wav")}, + ) + job_ids.append(created.json()["id"]) + for job_id in job_ids: + for _ in range(40): + job = client.get(f"/api/jobs/{job_id}").json() + if job["status"] == "waiting_review": + break + time.sleep(0.05) + self.assertEqual(job["status"], "waiting_review") + + rendering_id, transcribing_id = job_ids + app.state.manager._set_status( + app.state.manager.get_job(rendering_id), "rendering", 0.97, error=None + ) + app.state.manager._set_status( + app.state.manager.get_job(transcribing_id), "transcribing", 0.5, error=None + ) + + restarted = create_app(model_path="fake-model", runs_dir=tmpdir, max_new_tokens=8) + restarted_client = TestClient(restarted) + recovered = restarted_client.get(f"/api/jobs/{rendering_id}").json() + + self.assertEqual(recovered["status"], "waiting_review") + self.assertEqual(recovered["progress"], 0.95) + self.assertIn("interrupted", recovered["error"]) + self.assertIn("retry rendering", recovered["error"]) + segments = restarted_client.get(f"/api/jobs/{rendering_id}/segments").json()["segments"] + edited = [dict(item) for item in segments] + edited[0]["text"] = "editable after restart" + updated = restarted_client.put( + f"/api/jobs/{rendering_id}/segments", + json={"segments": edited}, + ) + self.assertEqual(updated.status_code, 200) + + failed = restarted_client.get(f"/api/jobs/{transcribing_id}").json() + self.assertEqual(failed["status"], "failed") + self.assertEqual(failed["progress"], 1.0) + self.assertEqual(failed["error"], "Interrupted by previous server shutdown.") + if __name__ == "__main__": unittest.main()