rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

files: a failed directory rename that hung for five minutes

Browse source 
A directory rename waits inline on a move job. When that move failed
(stale FileGroup rows colliding on primary_path), handle_move reverted
the files and sent a failure event but left the job pending. wait_for_job
only understands 'complete', so POST /api/files/rename spun until its
300s timeout and the Rename modal never closed.

The same sequence could kill file processing for the life of the process.
perpetual_signal treated any CancelledError as terminal and logged
nothing. Separately, perpetual_tasks_started is a shared Event that
survives the worker that set it: docker auto-reloads and worker crashes
left every later after_server_start as a no-op. Queued refreshes sat
untouched while worker_status reported idle. Restarting the API was the
only recovery.

Mark failed jobs failed and raise immediately from wait_for_job. Catch
unexpected cancellation and reschedule unless the app is actually
stopping. Track the perpetual-loop owner PID and retake the claim when
that process is gone; every Sanic worker still runs the file-worker
pump. Harden transfer_queue against the cross-process Empty race, and
stop request processes stealing from the shared public queue.
This commit is contained in:
Roland 2026-08-19 15:25:11 -06:00
parent 2b7bc4533f
commit 85f22b61ca
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
10 changed files with 723 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

3
main.py
View file

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

182
wrolpi/api_utils.py
View file

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

12
wrolpi/collections/reorganize.py
View file

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

69
wrolpi/collections/test/test_reorganize.py
View file

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

7
wrolpi/contexts.py
View file

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

31
wrolpi/files/test/test_api.py
View file

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

95
wrolpi/files/test/test_worker.py
View file

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

70
wrolpi/files/test/test_worker_move.py
View file

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

143
wrolpi/files/worker.py
View file

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

163
wrolpi/test/test_api_utils.py Normal file
View file

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