Fix FileWorker status stranded mid-phase (stuck on "Cleanup") (#500)

* Fix FileWorker status stranded mid-phase (e.g. stuck on "Cleanup")

The FileWorker stores its status in shared context and relies on an explicit
reset_status() in each handler. Two execution paths set the status (comparing/
upserting/deleting/modeling/indexing/cleanup) but could finish without ever
resetting it, freezing the UI at the last phase:

- refresh_sync() ran _apply_post_processing (final phase sets status='cleanup')
  but never called reset_status(). Deleting files in the UI goes through
  lib.delete -> refresh_sync, so any delete that changed the DB stranded the
  worker at 'cleanup' (observed on 10.0.0.8 after deleting STL files).

- process_queue() wrapped the handler dispatch in try/finally with no except.
  An exception in handle_count/handle_refresh (which, unlike handle_move/
  handle_reorganize, have no try/except) escaped to the perpetual worker loop
  and left the status frozen at whatever phase was running ("any state").

Fixes:
- refresh_sync(): reset_status() in a finally.
- process_queue(): on exception record a terminal 'error' status (so the worker
  is not stranded mid-phase) and re-raise so callers and the perpetual loop
  still observe the failure. This covers both handle_count and handle_refresh.

Tests: add replication tests for the synchronous-refresh, handler-error, and
count-error stranding paths, plus the previously-untested wait_for_job timeout.
* refresh_sync: leave terminal 'error' status on failure

Address Greptile review: refresh_sync reset to idle in a finally that also ran
on failure, so a synchronous refresh that raised after a phase was set reported
a clean 'idle' to the status endpoint instead of the failure (the exception was
still raised to the caller). Make it symmetric with process_queue: reset to idle
only on success, record status='error' and re-raise on failure.

Add test_refresh_sync_records_error_status_on_failure.

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lrnselfreliance 2026-06-28 13:19:59 -06:00 • committed by Roland
parent e8bb7b7fb1
commit 627fbf47e3
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
2 changed files with 141 additions and 10 deletions

View file

@ -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

View file

@ -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: