diff --git a/main.py b/main.py index 4e6d09c0..0fd1647b 100755 --- a/main.py +++ b/main.py @@ -318,7 +318,8 @@ async def start_single_tasks(app: Sanic): @perpetual_signal(sleep=1) async def perpetual_file_worker_queue(): file_worker.transfer_queue() - await file_worker.process_queue() + processed = await file_worker.process_queue() + file_worker.check_queue_stall(processed=processed) @perpetual_signal(sleep=1) diff --git a/wrolpi/api_utils.py b/wrolpi/api_utils.py index d3b9d405..5ec1deb2 100644 --- a/wrolpi/api_utils.py +++ b/wrolpi/api_utils.py @@ -1,6 +1,7 @@ import asyncio import json import logging +import os from asyncio import CancelledError from datetime import datetime, timezone, date from decimal import Decimal @@ -164,18 +165,113 @@ async def cleanup_session(request: Request, response_: HTTPResponse): PERPETUAL_WORKERS = list() +# Always started in every Sanic worker so file jobs survive the owner process dying. +FILE_WORKER_PERPETUAL_EVENT = 'wrolpi.perpetual.perpetual_file_worker_queue' + + +def _perpetual_owner_is_alive(app) -> bool: + """True if the process that claimed the single-process perpetual loops is still running.""" + try: + pid = app.shared_ctx.perpetual_tasks_owner_pid.value + except Exception: + return False + if not pid: + return False + from wrolpi.cmd import pid_is_running + return pid_is_running(pid) + + +def _claim_perpetual_tasks(app) -> bool: + """Return True if this process should start the single-process perpetual loops. + + ``perpetual_tasks_started`` is a shared Event. If the worker that set it + exits (Sanic auto_reload in docker, crash), the Event stays set and every + replacement ``after_server_start`` used to no-op — file processing, downloads, + and switches stayed dead until a full API restart. Track the owner PID and + retake the claim when that process is gone. + """ + lock = getattr(app.shared_ctx, 'perpetual_tasks_lock', None) + if lock is None: + return _claim_perpetual_tasks_unlocked(app) + with lock: + return _claim_perpetual_tasks_unlocked(app) + + +def _claim_perpetual_tasks_unlocked(app) -> bool: + if app.shared_ctx.perpetual_tasks_started.is_set() and _perpetual_owner_is_alive(app): + return False + app.shared_ctx.perpetual_tasks_started.set() + try: + app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid() + except Exception: + pass + return True + + +def _perpetual_events_for_this_process(app) -> list[str]: + """Events this worker should dispatch at startup. + + The claiming worker starts every perpetual loop. Every other worker still + starts the file-worker pump so a dead/cancelled owner cannot stall file + operations for the life of the remaining processes. + """ + if _claim_perpetual_tasks(app): + return list(PERPETUAL_WORKERS) + return [FILE_WORKER_PERPETUAL_EVENT] + + +def _app_is_stopping() -> bool: + """True when Sanic is shutting down (perpetual workers should not reschedule). + + Only the ``before_server_stop`` flag (and Sanic's ``is_stopping`` if it is + ever set) count. ``state.is_running`` stays False in some serving modes + (including this project's docker Sanic workers), so treating + ``is_started and not is_running`` as shutdown killed every perpetual + worker after its first tick. + """ + try: + if getattr(api_app.ctx, 'perpetual_shutdown', False): + return True + state = getattr(api_app, 'state', None) + return bool(state is not None and getattr(state, 'is_stopping', False)) + except Exception: + return False + + +def _uncancel_current_task(): + """Allow a perpetual worker to continue after an unexpected CancelledError. + + Python 3.11+ re-raises CancelledError at the next await unless uncancel() + is called. Without this, catching CancelledError cannot reschedule. + """ + task = asyncio.current_task() + if task is not None and hasattr(task, 'uncancel'): + task.uncancel() + + +@api_app.listener('before_server_start') +async def _clear_perpetual_shutdown(app): + app.ctx.perpetual_shutdown = False + + +@api_app.listener('before_server_stop') +async def _set_perpetual_shutdown(app): + app.ctx.perpetual_shutdown = True + + +async def _dispatch_perpetual(event_: str): + """Dispatch a perpetual-signal event. Isolated so tests can stub it.""" + await api_app.dispatch(event_) + @api_app.after_server_start async def start_perpetual_tasks(app: Sanic): - # Only one set of perpetual tasks needs to be started. - if app.shared_ctx.perpetual_tasks_started.is_set(): - return - logger.info('start_perpetual_tasks started') - logger.debug(f'start_perpetual_tasks: {PERPETUAL_WORKERS}') - app.shared_ctx.perpetual_tasks_started.set() + events = _perpetual_events_for_this_process(app) + logger.info(f'start_perpetual_tasks started pid={os.getpid()} events={len(events)}') + logger.debug(f'start_perpetual_tasks: {events}') try: - for event_ in PERPETUAL_WORKERS: + for event_ in events: logger.debug(f'start_perpetual_tasks {event_}') await app.dispatch(event_) except Exception as e: @@ -185,6 +281,58 @@ async def start_perpetual_tasks(app: Sanic): logger.debug('start_perpetual_tasks completed') +async def _run_perpetual_iteration(func: callable, event_: str, sleep: int | float): + """Run one perpetual-worker iteration and reschedule unless shutting down. + + An ordinary Exception is logged and the loop continues. CancelledError is + terminal only during shutdown; any other cancellation is logged and the + worker is rescheduled. Previously any CancelledError skipped reschedule + with no log, silently killing file processing for the life of the process. + """ + logger.trace(f'perpetual_signal {event_}') + start = time() + try: + await func() + except CancelledError: + if _app_is_stopping(): + logger.info(f'Perpetual worker {event_} cancelled during shutdown') + raise + logger.warning( + f'Perpetual worker {event_} cancelled unexpectedly; will reschedule', + exc_info=True, + ) + _uncancel_current_task() + except Exception as e: + logger.error(f'Perpetual worker {event_} had error', exc_info=e) + finally: + if __debug__ and logger.isEnabledFor(TRACE_LEVEL): + elapsed = int(time() - start) + logger.trace(f'perpetual_signal {event_} took {elapsed} seconds') + + if PYTEST: + return + if _app_is_stopping(): + logger.info(f'Perpetual worker {event_} not rescheduling because the app is stopping') + return + + try: + await asyncio.sleep(sleep) + except CancelledError: + if _app_is_stopping(): + logger.info(f'Perpetual worker {event_} stopped during shutdown') + raise + logger.warning( + f'Perpetual worker {event_} cancelled while sleeping; rescheduling immediately' + ) + _uncancel_current_task() + + try: + await _dispatch_perpetual(event_) + except CancelledError: + logger.info(f'Perpetual worker {event_} cancelled while dispatching next run') + raise + + def perpetual_signal(event: str = None, sleep: int | float = 1, run_while_testing: bool = False): """Use Sanic signals to continually call the wrapped function. The wrapped function will continually be called, even if it has errors. If the function is long-running, it will only be called again after it has finished.""" @@ -200,25 +348,7 @@ def perpetual_signal(event: str = None, sleep: int | float = 1, run_while_testin # Wrap the function in a worker that will call it perpetually. @api_app.signal(event_) async def worker(*args, **kwargs): - logger.trace(f'perpetual_signal {event_}') - cancelled = False - start = time() - try: - await func(*args, **kwargs) - except CancelledError: - cancelled = True - raise - except Exception as e: - logger.error(f'Perpetual worker {event_} had error', exc_info=e) - finally: - if __debug__ and logger.isEnabledFor(TRACE_LEVEL): - elapsed = int(time() - start) - logger.trace(f'perpetual_signal {event_} took {elapsed} seconds') - if not cancelled and not PYTEST: - # In PYTEST mode, don't self-dispatch - the test fixture handles this. - # This prevents "Task was destroyed but it is pending!" errors during teardown. - await asyncio.sleep(sleep) - await api_app.dispatch(event_) + await _run_perpetual_iteration(lambda: func(*args, **kwargs), event_, sleep) # Add this new signal to the global list so that a task will be started after server startup. PERPETUAL_WORKERS.append(event_) diff --git a/wrolpi/collections/reorganize.py b/wrolpi/collections/reorganize.py index 4f5700bb..0e80dd21 100644 --- a/wrolpi/collections/reorganize.py +++ b/wrolpi/collections/reorganize.py @@ -978,6 +978,16 @@ def get_reorganization_status(job_id: str) -> dict: 'percent': 0, 'error': None, } + elif job_status == 'failed': + # reset_status() clears worker_status['error'] after the handler + # finishes; the recorded job error is the durable source. + return { + 'status': 'failed', + 'total': worker_status.get('operation_total', 0), + 'completed': worker_status.get('operation_processed', 0), + 'percent': worker_status.get('operation_percent', 0), + 'error': file_worker.get_job_error(job_id) or worker_status.get('error'), + } else: # Job is running return { @@ -1189,7 +1199,7 @@ def get_batch_reorganization_status(batch_job_id: str) -> dict: 'overall_percent': batch_status.get('overall_percent', 0), 'completed': completed_list, 'failed_collection': batch_status.get('failed_collection'), - 'error': batch_status.get('error'), + 'error': file_worker.get_job_error(batch_job_id) or batch_status.get('error'), } else: # Job is running diff --git a/wrolpi/collections/test/test_reorganize.py b/wrolpi/collections/test/test_reorganize.py index 2ac0139c..fcbe3670 100644 --- a/wrolpi/collections/test/test_reorganize.py +++ b/wrolpi/collections/test/test_reorganize.py @@ -179,6 +179,75 @@ async def test_get_reorganization_status_unknown_job(async_client): assert 'error' in status +@pytest.mark.asyncio +async def test_get_reorganization_status_failed_job(async_client): + """A failed reorganize job must report status=failed with the recorded error. + + reset_status() clears worker_status['error'], so the error has to come from + the job record itself — otherwise the UI would show an idle/running job. + """ + from wrolpi.files.worker import file_worker + + file_worker._fail_job('reorg-failed', 'disk full') + status = get_reorganization_status('reorg-failed') + assert status['status'] == 'failed' + assert status['error'] == 'disk full' + + +@pytest.mark.asyncio +async def test_handle_reorganize_marks_job_failed_on_exception( + async_client, test_directory, video_factory +): + """handle_reorganize must mark its job failed when a move raises. + + Regression: the except path sent a failure event but left the job + 'pending', so waiters (and the batch reorganize poller) spun forever. + """ + from unittest import mock + from wrolpi.files.worker import file_worker, FileTask, FileTaskType + + channel_dir = test_directory / 'videos' / 'FailChannel' + channel_dir.mkdir(parents=True, exist_ok=True) + + with get_db_session(commit=True) as session: + collection = Collection( + name='FailChannel', + kind='channel', + directory=channel_dir, + file_format='%(title)s.%(ext)s', + ) + session.add(collection) + session.flush() + channel = Channel(name='FailChannel', collection_id=collection.id, directory=channel_dir) + session.add(channel) + session.commit() + channel_id = channel.id + + video = video_factory(channel_id=channel_id, title='will_fail', with_video_file=True) + with get_db_session() as session: + video_obj = session.query(Video).get(video.id) + old_path = video_obj.file_group.primary_path + + dest = channel_dir / 'renamed_will_fail.mp4' + job_id = 'reorganize-will-fail' + task = FileTask( + task_type=FileTaskType.reorganize, + paths=[], + move_mappings=[(old_path, dest)], + job_id=job_id, + ) + file_worker._set_job_status(job_id, 'pending') + + with mock.patch('wrolpi.files.worker.shutil.move', side_effect=OSError('disk full')): + await file_worker.handle_reorganize(task) + + assert file_worker.get_job_status(job_id) == 'failed' + assert 'disk full' in (file_worker.get_job_error(job_id) or '') + status = get_reorganization_status(job_id) + assert status['status'] == 'failed' + assert 'disk full' in (status['error'] or '') + + def test_filegroup_primary_path_lookup_consistency(test_session, test_directory): """FileGroup.primary_path returns a Path object, but IN queries use strings. diff --git a/wrolpi/contexts.py b/wrolpi/contexts.py index 024d9472..e542ded8 100644 --- a/wrolpi/contexts.py +++ b/wrolpi/contexts.py @@ -61,6 +61,11 @@ def attach_shared_contexts(app: Sanic): app.shared_ctx.single_tasks_started = multiprocessing.Event() app.shared_ctx.flags_initialized = multiprocessing.Event() app.shared_ctx.perpetual_tasks_started = multiprocessing.Event() + # PID of the worker that started the single-process perpetual loops. + # The Event alone survives that worker dying (Sanic auto_reload / crash), + # so replacement workers check this PID before skipping startup. + app.shared_ctx.perpetual_tasks_owner_pid = multiprocessing.Value(ctypes.c_int, 0) + app.shared_ctx.perpetual_tasks_lock = multiprocessing.Lock() # Switches app.shared_ctx.switches = manager.dict() @@ -206,6 +211,8 @@ def reset_shared_contexts(app: Sanic): app.shared_ctx.single_tasks_started.clear() app.shared_ctx.flags_initialized.clear() app.shared_ctx.perpetual_tasks_started.clear() + if hasattr(app.shared_ctx, 'perpetual_tasks_owner_pid'): + app.shared_ctx.perpetual_tasks_owner_pid.value = 0 # Do not start downloads when reloading. app.shared_ctx.download_manager_stopped.set() diff --git a/wrolpi/files/test/test_api.py b/wrolpi/files/test/test_api.py index 9c8fc1ab..d077a5b6 100644 --- a/wrolpi/files/test/test_api.py +++ b/wrolpi/files/test/test_api.py @@ -1261,6 +1261,37 @@ async def test_rename_directory(test_session, test_directory, make_files_structu assert foo.name == 'foo' and qux.name == 'qux' +@pytest.mark.asyncio +async def test_rename_directory_stale_destination_fails_promptly( + test_session, test_directory, make_files_structure, async_client, refresh_files +): + """A directory rename that collides with a stale FileGroup must fail promptly. + + Regression: the move's IntegrityError left the job 'pending', so + POST /api/files/rename hung until wait_for_job's 300s timeout. + """ + import asyncio + import shutil + + make_files_structure({'dir-a/one.txt': 'hello'}) + await refresh_files() + + content = dict(path='dir-a', new_name='dir-b') + request, response = await async_client.post('/api/files/rename', content=json.dumps(content)) + assert response.status_code == HTTPStatus.NO_CONTENT + + # Remove dest on disk without refreshing; recreate the old source. + shutil.rmtree(test_directory / 'dir-b') + (test_directory / 'dir-a').mkdir() + (test_directory / 'dir-a' / 'one.txt').write_text('hello') + + request, response = await asyncio.wait_for( + async_client.post('/api/files/rename', content=json.dumps(content)), + timeout=5, + ) + assert response.status_code == HTTPStatus.CONFLICT, response.json + + @pytest.mark.asyncio async def test_rename_collection_directory(test_session, test_directory, make_files_structure, async_client, refresh_files): diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index c1b8856d..495f4451 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -8,7 +8,16 @@ from sqlalchemy import text from wrolpi.conftest import await_switches from wrolpi.dates import from_timestamp from wrolpi.files.models import FileGroup -from wrolpi.files.worker import compare_file_groups, count_files, file_worker, FileGroupDiff, FileTask, FileTaskType +from wrolpi.files.worker import ( + compare_file_groups, + count_files, + file_worker, + FileGroupDiff, + FileTask, + FileTaskType, + FileWorkerJobFailed, + QUEUE_STALL_SECONDS, +) @pytest.mark.asyncio @@ -673,6 +682,90 @@ async def test_wait_for_job_times_out(async_client, test_session): await file_worker.wait_for_job(job_id, timeout=0.1) +@pytest.mark.asyncio +async def test_wait_for_job_raises_immediately_when_job_failed(async_client, test_session): + """wait_for_job must treat 'failed' as terminal and raise with the recorded error. + + Regression: a failed move left the job 'pending', so wait_for_job spun until + its 300s timeout and POST /api/files/rename hung for 5 minutes. + """ + job_id = 'move-already-failed' + file_worker._fail_job(job_id, 'UNIQUE constraint failed: file_group.primary_path') + + start = asyncio.get_event_loop().time() + with pytest.raises(FileWorkerJobFailed, match='UNIQUE constraint failed'): + await file_worker.wait_for_job(job_id, timeout=5) + elapsed = asyncio.get_event_loop().time() - start + assert elapsed < 1, f'wait_for_job should fail immediately, took {elapsed:.2f}s' + assert file_worker.get_job_status(job_id) == 'failed' + + +@pytest.mark.asyncio +async def test_process_queue_marks_refresh_job_failed(async_client, test_session, test_directory, + make_files_structure): + """A handler exception must mark the tracked job failed, not leave it pending.""" + make_files_structure(['docs/file1.txt']) + job_id = 'refresh-will-fail' + task = FileTask(FileTaskType.refresh, [test_directory], count=1, job_id=job_id) + file_worker._set_job_status(job_id, 'pending') + file_worker.private_queue.put_nowait(task) + + with mock.patch('wrolpi.files.worker.apply_modelers', side_effect=RuntimeError('boom')): + with pytest.raises(RuntimeError, match='boom'): + await file_worker.process_queue() + + assert file_worker.get_job_status(job_id) == 'failed' + assert 'boom' in (file_worker.get_job_error(job_id) or '') + + +@pytest.mark.asyncio +async def test_transfer_queue_survives_empty_race(async_client): + """transfer_queue must not raise when get_nowait races with another process. + + The public queue is a multiprocessing.Queue. qsize() > 0 followed by + get_nowait() is check-then-act; the loser gets queue.Empty. + """ + import queue as queue_mod + + public = mock.Mock() + public.get_nowait.side_effect = queue_mod.Empty + with mock.patch.object(type(file_worker), 'public_queue', new_callable=mock.PropertyMock, + return_value=public): + file_worker.transfer_queue() # must not raise + + +@pytest.mark.asyncio +async def test_check_queue_stall_logs_when_idle_with_queued_jobs(async_client, caplog): + """A non-empty queue while status is idle for N seconds is a logged stall.""" + import logging + import time as time_mod + + file_worker.update_status(status='idle') + file_worker._queue_stall_since = time_mod.time() - (QUEUE_STALL_SECONDS + 1) + public = mock.Mock() + public.qsize.return_value = 3 + with mock.patch.object(type(file_worker), 'public_queue', new_callable=mock.PropertyMock, + return_value=public): + with caplog.at_level(logging.ERROR): + file_worker.check_queue_stall(processed=False) + assert 'stalled' in caplog.text.lower() + assert '3' in caplog.text + + +@pytest.mark.asyncio +async def test_check_queue_stall_resets_when_a_job_was_processed(async_client, caplog): + """A tick that handled a job is not a stall, even if more jobs remain.""" + import logging + import time as time_mod + + file_worker.update_status(status='idle') + file_worker._queue_stall_since = time_mod.time() - (QUEUE_STALL_SECONDS + 1) + with caplog.at_level(logging.ERROR): + file_worker.check_queue_stall(processed=True) + assert 'stalled' not in caplog.text.lower() + assert file_worker._queue_stall_since is None + + @pytest.mark.asyncio async def test_file_worker_refresh_inserts_new_files(async_client, test_session, test_directory, make_files_structure, video_file_factory): diff --git a/wrolpi/files/test/test_worker_move.py b/wrolpi/files/test/test_worker_move.py index a2a0a919..2dcc7d9e 100644 --- a/wrolpi/files/test/test_worker_move.py +++ b/wrolpi/files/test/test_worker_move.py @@ -7,7 +7,7 @@ from wrolpi.conftest import production_like_sessions, write_lock_held_briefly from wrolpi.files import worker as worker_module from wrolpi.files.lib import _move_file_group_files from wrolpi.files.models import FileGroup, Directory -from wrolpi.files.worker import file_worker, FileTask, FileTaskType, build_move_plan_bulk +from wrolpi.files.worker import file_worker, FileTask, FileTaskType, FileWorkerJobFailed, build_move_plan_bulk @pytest.mark.asyncio @@ -398,6 +398,74 @@ async def test_file_worker_move_rollback_on_failure( "Destination should be empty after rollback" +@pytest.mark.asyncio +async def test_handle_move_marks_job_failed_on_exception( + async_client, test_session, test_directory, make_files_structure, monkeypatch +): + """A failed move must mark its tracked job 'failed' so wait_for_job can return. + + Regression: handle_move's except path sent a failure event and reverted files + but never touched the job status, leaving it 'pending' forever. + """ + source_file, = make_files_structure(['source/file1.txt']) + FileGroup.from_paths(test_session, source_file) + test_session.commit() + + dest = test_directory / 'destination' + dest.mkdir() + + def failing_move(fg, new_path): + raise IOError('Simulated disk failure') + + monkeypatch.setattr('wrolpi.files.worker._move_file_group_files', failing_move) + + job_id = 'move-will-fail' + task = FileTask(FileTaskType.move, [source_file], destination=dest, job_id=job_id) + file_worker._set_job_status(job_id, 'pending') + file_worker.private_queue.put_nowait(task) + + await file_worker.process_queue() + + assert file_worker.get_job_status(job_id) == 'failed' + assert 'Simulated disk failure' in (file_worker.get_job_error(job_id) or '') + assert source_file.exists(), 'Source should be reverted' + + +@pytest.mark.asyncio +async def test_wait_for_job_fails_promptly_on_stale_destination_filegroup( + async_client, test_session, test_directory, make_files_structure +): + """A move that collides with a stale destination FileGroup must fail promptly. + + Reproduction from the incident: rename dir-a -> dir-b, delete dir-b on disk + without a refresh, recreate dir-a/one.txt, rename again. The move plan + INSERTs a FileGroup for the new source then UPDATEs primary_path onto the + stale destination row and hits UNIQUE constraint. The waiter must raise + FileWorkerJobFailed immediately, not spin until the 300s timeout. + """ + source, = make_files_structure({'dir-a/one.txt': 'hello'}) + FileGroup.from_paths(test_session, source) + test_session.commit() + + dest_b = test_directory / 'dir-b' + first_job = file_worker.queue_move(dest_b, [source]) + await file_worker.wait_for_job(first_job, timeout=10) + assert (dest_b / 'one.txt').exists() + + # Dest gone on disk; DB still holds dir-b/one.txt. Recreate the old source. + # The first move leaves the empty source directory in place. + import shutil + shutil.rmtree(dest_b) + source.parent.mkdir(exist_ok=True) + source.write_text('hello') + + second_job = file_worker.queue_move(dest_b, [source]) + with pytest.raises(FileWorkerJobFailed): + await file_worker.wait_for_job(second_job, timeout=5) + assert file_worker.get_job_status(second_job) == 'failed' + assert source.exists(), 'Failed move should revert the source file' + + @pytest.mark.asyncio async def test_file_worker_move_unindexed_files( async_client, test_session, test_directory, make_files_structure diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 650cdf1e..0b02f959 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -8,6 +8,7 @@ import asyncio import json import os import pathlib +import queue import shlex import shutil import stat as stat_module @@ -30,6 +31,7 @@ from wrolpi.dates import now from wrolpi.db import get_db_session, get_db_curs from wrolpi.errors import NoPrimaryFile from wrolpi.events import Events +from wrolpi.vars import PYTEST from wrolpi.files.lib import ( split_path_stem_and_suffix, _upsert_files, get_unique_files_by_stem, glob_shared_stem, group_files_by_stem, get_primary_file, delete_directory, apply_indexers, @@ -42,6 +44,9 @@ logger = logger.getChild(__name__) # Update status every N items to avoid excessive overhead PROGRESS_UPDATE_INTERVAL = 100 +# Log if jobs sit queued while the worker reports idle for this long. +QUEUE_STALL_SECONDS = 30 + # Maximum mtime difference (seconds) before a file is considered "modified". # 0.5s accommodates SD card / FAT32 / exFAT timestamp granularity on Pi 4. MTIME_TOLERANCE_SECONDS = 0.5 @@ -49,12 +54,17 @@ MTIME_TOLERANCE_SECONDS = 0.5 __all__ = [ 'FileGroupDiff', 'FileComparisonResult', + 'FileWorkerJobFailed', 'compare_file_groups', 'count_files', 'file_worker', ] +class FileWorkerJobFailed(RuntimeError): + """A tracked file-worker job finished unsuccessfully.""" + + @dataclass class FileGroupDiff: """Represents changes needed for a FileGroup.""" @@ -795,6 +805,7 @@ class FileWorker: def __init__(self): # Use last-in, first out so a task that inserts another task is handled next. self.private_queue = asyncio.LifoQueue() + self._queue_stall_since: float | None = None @property def public_queue(self) -> Queue: @@ -831,27 +842,50 @@ class FileWorker: from wrolpi.api_utils import api_app return api_app.shared_ctx.file_worker_jobs - def _set_job_status(self, job_id: str, status: str): - """Set the status of a tracked job.""" - self._jobs[job_id] = status + def _set_job_status(self, job_id: str, status: str, error: str | None = None): + """Set the status of a tracked job. + + Failed jobs store ``{'status': 'failed', 'error': ...}`` so waiters can + surface the recorded error after ``reset_status()`` clears worker status. + """ + if not job_id: + return + if error is not None: + self._jobs[job_id] = {'status': status, 'error': error} + else: + self._jobs[job_id] = status def _complete_job(self, job_id: str): """Mark a job as complete.""" - if job_id: - self._set_job_status(job_id, 'complete') + self._set_job_status(job_id, 'complete') + + def _fail_job(self, job_id: str, error: str | BaseException): + """Mark a job as failed and record the error for waiters.""" + self._set_job_status(job_id, 'failed', error=str(error)) def get_job_status(self, job_id: str) -> str | None: """Get the status of a tracked job.""" job_status = self._jobs.get(job_id) if not job_status: raise RuntimeError(f'Job {job_id} does not exist') + if isinstance(job_status, dict): + return job_status.get('status') return job_status - async def wait_for_job(self, job_id: str, timeout: float = 300): - """Wait for a job to complete. + def get_job_error(self, job_id: str) -> str | None: + """Return the recorded error for a failed job, if any.""" + job_status = self._jobs.get(job_id) + if isinstance(job_status, dict): + return job_status.get('error') + return None - Transfers and processes the queue while waiting to ensure jobs are actually executed. - This is especially important in tests and synchronous contexts. + async def wait_for_job(self, job_id: str, timeout: float = 300): + """Wait for a job to complete or fail. + + In tests (no perpetual file-worker loop) this drains the shared queue + so the job actually runs. Production request processes only poll: + stealing from the public queue reorders work (the private queue is + LIFO) and can starve the perpetual loop. Args: job_id: The job ID to wait for @@ -859,8 +893,8 @@ class FileWorker: Raises: TimeoutError: If job doesn't complete within timeout + FileWorkerJobFailed: If the job is marked failed """ - import time start = time.time() iteration = 0 while time.time() - start < timeout: @@ -868,9 +902,13 @@ class FileWorker: status = self.get_job_status(job_id) if status == 'complete': return - # Transfer from public to private queue, then process - self.transfer_queue() - await self.process_queue() + if status == 'failed': + error = self.get_job_error(job_id) or 'unknown error' + logger.error(f'wait_for_job: job_id={job_id} failed: {error}') + raise FileWorkerJobFailed(f'Job {job_id} failed: {error}') + if PYTEST: + self.transfer_queue() + await self.process_queue() await asyncio.sleep(0.01) logger.error(f'wait_for_job: TIMEOUT job_id={job_id} after {iteration} iterations') raise TimeoutError(f'Job {job_id} did not complete in time') @@ -1024,12 +1062,50 @@ class FileWorker: self.reset_status() def transfer_queue(self): - """Transfer items from the public queue to the private queue.""" - while self.public_queue.qsize() > 0: - item = self.public_queue.get_nowait() + """Transfer items from the public queue to the private queue. + + Uses ``get_nowait()`` in a loop rather than checking ``qsize()`` first: + the public queue is a multiprocessing.Queue drained by every process, so + ``qsize() > 0`` followed by ``get_nowait()`` races and raises + ``queue.Empty``. + """ + while True: + try: + item = self.public_queue.get_nowait() + except queue.Empty: + break self.private_queue.put_nowait(item) - async def process_queue(self): + def check_queue_stall(self, processed: bool = False) -> None: + """Log if jobs sit queued while the worker looks idle. + + ``process_queue`` handles one task per tick, so a shrinking backlog is + not a stall. Only warn when a tick processed nothing and the queues + stayed non-empty while status is idle. + """ + if processed: + self._queue_stall_since = None + return + try: + queued = self.public_queue.qsize() + self.private_queue.qsize() + except Exception: + return + if queued and self.status.get('status') == 'idle': + now = time.time() + if self._queue_stall_since is None: + self._queue_stall_since = now + elif now - self._queue_stall_since >= QUEUE_STALL_SECONDS: + logger.error( + 'File worker appears stalled: %s queued job(s) while status is idle for %ss', + queued, + int(now - self._queue_stall_since), + ) + self._queue_stall_since = now + else: + self._queue_stall_since = None + + async def process_queue(self) -> bool: + """Process one queued task. Returns True if a task was handled.""" try: task: FileTask = self.private_queue.get_nowait() except asyncio.QueueEmpty: @@ -1037,7 +1113,7 @@ class FileWorker: if flags.file_worker_busy.is_set() and self.public_queue.qsize() == 0: flags.file_worker_busy.clear() await asyncio.sleep(0) - return + return False # Set flag before processing any task (if not already set) if not flags.file_worker_busy.is_set(): @@ -1056,6 +1132,7 @@ class FileWorker: case FileTaskType.batch_reorganize: await self.handle_batch_reorganize(task) except asyncio.CancelledError: + self._fail_job(task.job_id, 'File worker task was cancelled') raise except Exception as e: # `handle_count`/`handle_refresh` reset their own status on success (or intentionally @@ -1066,11 +1143,13 @@ class FileWorker: # 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)) + self._fail_job(task.job_id, e) raise finally: # Clear flag if both queues are now empty if self.private_queue.qsize() == 0 and self.public_queue.qsize() == 0: flags.file_worker_busy.clear() + return True async def handle_count(self, task: FileTask): """Count files in the task's paths and chain to next task if specified.""" @@ -1994,12 +2073,14 @@ class FileWorker: if not destination or not sources: logger.error('Invalid move task: missing destination or sources') + self._fail_job(task.job_id, 'Invalid move task: missing destination or sources') self.reset_status() return media_directory = get_media_directory() if not self._validate_move_paths(sources, destination, media_directory): + self._fail_job(task.job_id, 'Move paths are not within the media directory') return # Estimate total for planning phase based on number of sources @@ -2059,11 +2140,18 @@ class FileWorker: except asyncio.CancelledError: logger.warning('Move task was cancelled') + try: + if revert_plan: + self._revert_move(revert_plan, created_directories, destination, destination_existed) + except Exception as revert_error: + logger.error(f'Failed to revert cancelled move: {revert_error}') + self._fail_job(task.job_id, 'Move task was cancelled') raise except Exception as e: logger.error(f'Move failed: {e}, reverting {len(revert_plan)} items', exc_info=e) self.update_status(status='reverting', error=str(e)) self._revert_move(revert_plan, created_directories, destination, destination_existed) + self._fail_job(task.job_id, e) Events.send_file_move_failed(f'Move to {destination} failed: {e}') finally: self.reset_status() @@ -2354,12 +2442,14 @@ class FileWorker: except asyncio.CancelledError: logger.warning('Reorganize task was cancelled') + self._fail_job(task.job_id, 'Reorganize task was cancelled') raise except Exception as e: logger.error(f'Reorganize failed: {e}', exc_info=e) self.update_status(status='error', error=str(e)) # Note: We don't attempt automatic rollback for reorganize tasks # as the partial state may be complex. User should refresh files. + self._fail_job(task.job_id, e) Events.send_file_move_failed(f'Reorganize failed: {e}') finally: self.reset_status() @@ -2497,7 +2587,7 @@ class FileWorker: error=str(e), batch_status=batch_status, ) - self._set_job_status(task.job_id, 'failed') + self._fail_job(task.job_id, e) Events.send_file_move_failed( f'Batch reorganization failed on {collection_name}: {e}' ) @@ -2512,12 +2602,13 @@ class FileWorker: except asyncio.CancelledError: logger.warning('Batch reorganize task was cancelled') + self._fail_job(task.job_id, 'Batch reorganize task was cancelled') raise except Exception as e: logger.error(f'Batch reorganize failed: {e}', exc_info=e) batch_status['error'] = str(e) self.update_status(status='error', error=str(e), batch_status=batch_status) - self._set_job_status(task.job_id, 'failed') + self._fail_job(task.job_id, e) Events.send_file_move_failed(f'Batch reorganization failed: {e}') finally: self.reset_status() @@ -2544,8 +2635,14 @@ class FileWorker: status = self.get_job_status(job_id) if status == 'complete': return + if status == 'failed': + error = self.get_job_error(job_id) or 'unknown error' + raise FileWorkerJobFailed( + f'Collection {collection_name} reorganization failed: {error}' + ) - # Transfer and process queue FIRST to ensure status is updated + # Nested under process_queue (batch reorganize); must drain the + # sub-job ourselves or this wait deadlocks. self.transfer_queue() await self.process_queue() @@ -2571,5 +2668,7 @@ class FileWorker: raise TimeoutError(f'Collection {collection_name} reorganization timed out') -# All processes create a FileWorker, but only one receives the signals and actually does work. +# Each Sanic worker constructs a FileWorker. The perpetual signal is started +# once, but request handlers (and wait_for_job in tests) can also drain the +# shared public queue in this process. file_worker = FileWorker() diff --git a/wrolpi/test/test_api_utils.py b/wrolpi/test/test_api_utils.py new file mode 100644 index 00000000..c8464866 --- /dev/null +++ b/wrolpi/test/test_api_utils.py @@ -0,0 +1,163 @@ +"""Tests for perpetual_signal cancellation / reschedule behavior.""" +import asyncio +from unittest import mock + +import pytest + +from wrolpi.api_utils import ( + _app_is_stopping, + _run_perpetual_iteration, + api_app, +) + + +def test_app_is_stopping_false_by_default(): + api_app.ctx.perpetual_shutdown = False + api_app.state.is_stopping = False + # is_started/is_running are not a shutdown signal: docker Sanic workers + # often have is_started=True and is_running=False while serving. + api_app.state.is_started = True + api_app.state.is_running = False + assert _app_is_stopping() is False + + +def test_app_is_stopping_when_shutdown_flag_set(): + api_app.ctx.perpetual_shutdown = True + try: + assert _app_is_stopping() is True + finally: + api_app.ctx.perpetual_shutdown = False + + +def test_app_is_stopping_when_sanic_is_stopping(): + api_app.ctx.perpetual_shutdown = False + api_app.state.is_stopping = True + try: + assert _app_is_stopping() is True + finally: + api_app.state.is_stopping = False + + +@pytest.mark.asyncio +async def test_perpetual_iteration_reschedules_after_unexpected_cancel(monkeypatch): + """CancelledError must not silently kill the perpetual loop. + + Regression: perpetual_signal caught CancelledError, set cancelled=True, + and skipped re-dispatch with no log. The file-worker pump died for the + life of the process. + """ + dispatched = [] + + async def boom(): + raise asyncio.CancelledError() + + async def fake_dispatch(event): + dispatched.append(event) + + monkeypatch.setattr('wrolpi.api_utils.PYTEST', False) + monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: False) + monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch) + monkeypatch.setattr('wrolpi.api_utils.asyncio.sleep', mock.AsyncMock()) + + await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0) + assert dispatched == ['wrolpi.perpetual.test_worker'] + + +@pytest.mark.asyncio +async def test_perpetual_iteration_stops_on_shutdown_cancel(monkeypatch): + """CancelledError during shutdown must stay terminal (do not reschedule).""" + dispatched = [] + + async def boom(): + raise asyncio.CancelledError() + + async def fake_dispatch(event): + dispatched.append(event) + + monkeypatch.setattr('wrolpi.api_utils.PYTEST', False) + monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: True) + monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch) + + with pytest.raises(asyncio.CancelledError): + await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0) + assert dispatched == [] + + +def test_claim_perpetual_tasks_when_never_started(async_client): + """The first worker to start must claim the perpetual loops.""" + from wrolpi.api_utils import _claim_perpetual_tasks, api_app + + api_app.shared_ctx.perpetual_tasks_started.clear() + assert _claim_perpetual_tasks(api_app) is True + + +def test_claim_perpetual_tasks_skips_when_owner_is_alive(async_client): + """A live owner keeps the single-process loops from starting twice.""" + import os + from wrolpi.api_utils import _claim_perpetual_tasks, api_app + + api_app.shared_ctx.perpetual_tasks_started.set() + api_app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid() + assert _claim_perpetual_tasks(api_app) is False + + +def test_claim_perpetual_tasks_when_owner_process_is_dead(async_client): + """A dead owner must not block a replacement worker from restarting the loops. + + ``perpetual_tasks_started`` is a shared Event. Sanic docker runs with + auto_reload, and a worker crash/recycle leaves that Event set. The + replacement process's after_server_start then no-ops, and file processing + (plus downloads, switches, …) is dead until a full API restart. + + This is the unconfirmed wedge from the 2026-08-18 incident: last file-worker + activity from one PID, then silence across every process, recovered only by + restarting the API. + """ + from wrolpi.api_utils import _claim_perpetual_tasks, api_app + + api_app.shared_ctx.perpetual_tasks_started.set() + # A PID that cannot be running on this host. + api_app.shared_ctx.perpetual_tasks_owner_pid.value = 2**31 - 1 + assert _claim_perpetual_tasks(api_app) is True + assert api_app.shared_ctx.perpetual_tasks_owner_pid.value != 2**31 - 1, \ + 'claimer must record this process as the new owner' + + +def test_perpetual_events_include_file_worker_when_not_claimed(async_client): + """Every Sanic worker must run the file-worker pump, even if it does not + own the other perpetual tasks. + + Otherwise a live owner whose *task* died (CancelledError before the + reschedule fix) still wedges file operations in every other process. + """ + import os + from wrolpi.api_utils import ( + FILE_WORKER_PERPETUAL_EVENT, + _perpetual_events_for_this_process, + api_app, + ) + + api_app.shared_ctx.perpetual_tasks_started.set() + api_app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid() + events = _perpetual_events_for_this_process(api_app) + assert events == [FILE_WORKER_PERPETUAL_EVENT] + + +@pytest.mark.asyncio +async def test_perpetual_iteration_reschedules_after_error(monkeypatch): + """Ordinary exceptions are logged and the loop continues.""" + dispatched = [] + + async def boom(): + raise RuntimeError('worker exploded') + + async def fake_dispatch(event): + dispatched.append(event) + + monkeypatch.setattr('wrolpi.api_utils.PYTEST', False) + monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: False) + monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch) + monkeypatch.setattr('wrolpi.api_utils.asyncio.sleep', mock.AsyncMock()) + + await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0) + assert dispatched == ['wrolpi.perpetual.test_worker']