diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 9db1bc20..cbbd0c60 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -571,6 +571,108 @@ async def test_file_worker_count_only_resets_status(async_client, test_session, assert status['operation_percent'] == 0 +@pytest.mark.asyncio +async def test_refresh_sync_resets_status(async_client, test_session, test_directory, make_files_structure): + """refresh_sync must reset status to idle when it finishes. + + Regression: deleting files in the UI calls `lib.delete` -> `file_worker.refresh_sync()`. + When that refresh actually changes the DB it runs `_apply_post_processing`, whose final + phase sets status='cleanup'. `refresh_sync` never calls `reset_status()`, so the worker is + stranded showing 'cleanup' forever (observed on 10.0.0.8 after deleting STL files). + """ + paths = make_files_structure(['docs/file1.txt']) + file_path = paths[0] + + # Index the file so the DB knows about it. A queued refresh resets to idle on its own. + task = FileTask(FileTaskType.refresh, [test_directory], count=1) + file_worker.private_queue.put_nowait(task) + await file_worker.process_queue() + assert file_worker.status['status'] == 'idle' + + # Delete the file on disk, then synchronously refresh so post-processing runs. + file_path.unlink() + await file_worker.refresh_sync([file_path]) + + # Without the fix the status is stranded at 'cleanup'. + assert file_worker.status['status'] == 'idle' + + +@pytest.mark.asyncio +async def test_refresh_sync_records_error_status_on_failure(async_client, test_session, test_directory, + make_files_structure): + """A failed synchronous refresh must report a terminal 'error' status, not look idle. + + `refresh_sync` resets to idle on success, but if it raises after a phase status was set the + status endpoint must surface the failure (status='error') rather than reporting a clean 'idle' + that hides the failed operation. The exception is still re-raised to the synchronous caller. + """ + # A new file on disk makes the refresh "changed", so post-processing actually runs. + paths = make_files_structure(['docs/file1.txt']) + + # Fail during the modeling phase of post-processing. + with mock.patch('wrolpi.files.worker.apply_modelers', side_effect=RuntimeError('boom')): + with pytest.raises(RuntimeError): + await file_worker.refresh_sync(paths) + + assert file_worker.status['status'] == 'error' + assert 'boom' in (file_worker.status['error'] or '') + + +@pytest.mark.asyncio +async def test_process_queue_resets_status_on_handler_error(async_client, test_session, test_directory, + make_files_structure): + """A handler raising mid-phase must not strand the worker status at that phase. + + Regression: `process_queue` had no `except` clause, so an exception raised inside a handler + (here, during the modeling phase) left the status frozen at whatever phase was running and the + UI showed that phase indefinitely. `process_queue` now records a terminal 'error' status (so + the worker is not stranded) and re-raises so callers still see the failure. + """ + make_files_structure(['docs/file1.txt']) + task = FileTask(FileTaskType.refresh, [test_directory], count=1) + file_worker.private_queue.put_nowait(task) + + # Fail during the modeling phase of post-processing. + with mock.patch('wrolpi.files.worker.apply_modelers', side_effect=RuntimeError('boom')): + with pytest.raises(RuntimeError): + await file_worker.process_queue() + + # Status is a terminal 'error', not stranded at 'modeling'. + assert file_worker.status['status'] == 'error' + + +@pytest.mark.asyncio +async def test_process_queue_resets_status_on_count_error(async_client, test_session, test_directory, + make_files_structure): + """A failure during the counting phase must not strand the worker status at 'counting'. + + Regression: `handle_count` (like `handle_refresh`) has no try/except, so an error while + counting left the status frozen at 'counting'. The shared `process_queue` recovery records a + terminal 'error' status for every handler, not just move/reorganize. + """ + make_files_structure(['docs/file1.txt']) + task = FileTask(FileTaskType.count, [test_directory], next_task_type=FileTaskType.refresh) + file_worker.private_queue.put_nowait(task) + + # Fail during the counting phase. + with mock.patch('wrolpi.files.worker.count_files_with_progress', side_effect=RuntimeError('boom')): + with pytest.raises(RuntimeError): + await file_worker.process_queue() + + # Status is a terminal 'error', not stranded at 'counting'. + assert file_worker.status['status'] == 'error' + + +@pytest.mark.asyncio +async def test_wait_for_job_times_out(async_client, test_session): + """wait_for_job raises TimeoutError when a job never completes (previously untested path).""" + job_id = 'refresh-never-completes' + file_worker._set_job_status(job_id, 'pending') + + with pytest.raises(TimeoutError): + await file_worker.wait_for_job(job_id, timeout=0.1) + + @pytest.mark.asyncio async def test_file_worker_refresh_inserts_new_files(async_client, test_session, test_directory, make_files_structure, video_file_factory): @@ -1065,6 +1167,8 @@ async def test_file_worker_refuses_refresh_when_only_ignored_files_exist( await file_worker.process_queue() assert 'empty' in str(exc_info.value).lower() or 'ignored' in str(exc_info.value).lower() + # The guard records a terminal 'error' status rather than stranding the worker mid-phase. + assert file_worker.status['status'] == 'error' @pytest.mark.asyncio diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index ba751948..c4510217 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -1001,16 +1001,31 @@ class FileWorker: """Synchronously refresh specific files. Use sparingly - prefer queue_refresh.""" if not paths: return - result = await self._refresh_files_directly(paths) - self._cleanup_modified_models(result.modified) - await self._upsert_file_groups(result.new + result.modified) - await self._delete_file_groups(result.deleted) - if post_processing and (result.new or result.modified or result.deleted): - # Skip global modelers/indexers when the caller opts out (e.g. `lib.delete` from the - # upload path defers that to the post-upload `upsert_file`) or when nothing changed. - # Without this guard a synchronous caller would block on the entire backlog of - # unindexed files and could exceed Sanic's RESPONSE_TIMEOUT. - await self._apply_post_processing(is_global_refresh=False) + # `_upsert_file_groups`/`_delete_file_groups`/`_apply_post_processing` all advance the + # shared status (upserting/deleting/modeling/indexing/cleanup). Unlike the queued + # handlers this path has no `process_queue` wrapper to reset it, so reset here or the UI + # is stranded at the last phase (e.g. 'cleanup' after deleting files). Mirror + # `process_queue`: reset to idle only on success, leave a terminal 'error' status on + # failure (and re-raise so the synchronous caller still sees it). + try: + result = await self._refresh_files_directly(paths) + self._cleanup_modified_models(result.modified) + await self._upsert_file_groups(result.new + result.modified) + await self._delete_file_groups(result.deleted) + if post_processing and (result.new or result.modified or result.deleted): + # Skip global modelers/indexers when the caller opts out (e.g. `lib.delete` from the + # upload path defers that to the post-upload `upsert_file`) or when nothing changed. + # Without this guard a synchronous caller would block on the entire backlog of + # unindexed files and could exceed Sanic's RESPONSE_TIMEOUT. + await self._apply_post_processing(is_global_refresh=False) + except asyncio.CancelledError: + raise + except Exception as e: + logger.error(f'refresh_sync failed for {len(paths)} paths', exc_info=e) + self.update_status(status='error', error=str(e)) + raise + else: + self.reset_status() def transfer_queue(self): """Transfer items from the public queue to the private queue.""" @@ -1044,6 +1059,18 @@ class FileWorker: await self.handle_reorganize(task) case FileTaskType.batch_reorganize: await self.handle_batch_reorganize(task) + except asyncio.CancelledError: + raise + except Exception as e: + # `handle_count`/`handle_refresh` reset their own status on success (or intentionally + # leave it set when chaining to a follow-up task), but they have no try/except, so an + # error mid-phase would otherwise strand the status (e.g. 'counting'/'cleanup') and the + # UI would show that phase forever. Record a terminal 'error' status so the worker is + # not stranded, then re-raise so callers and the perpetual worker loop still see the + # failure. (handle_move/handle_reorganize handle their own errors and do not reach here.) + logger.error(f'FileWorker task {task.task_type} failed', exc_info=e) + self.update_status(status='error', error=str(e)) + raise finally: # Clear flag if both queues are now empty if self.private_queue.qsize() == 0 and self.public_queue.qsize() == 0: