diff --git a/wrolpi/api_utils.py b/wrolpi/api_utils.py index 5ec1deb2..34daf5af 100644 --- a/wrolpi/api_utils.py +++ b/wrolpi/api_utils.py @@ -165,12 +165,32 @@ 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' +OWNER_WATCHDOG_EVENT = 'wrolpi.perpetual.perpetual_owner_watchdog' +OWNER_HEARTBEAT_STALE_SECONDS = 30 + + +def _touch_heartbeat(app) -> None: + try: + app.shared_ctx.perpetual_tasks_heartbeat.value = time() + except Exception: + logger.error('Failed to update perpetual-tasks heartbeat', exc_info=True) + + +def _i_am_owner(app) -> bool: + try: + return app.shared_ctx.perpetual_tasks_owner_pid.value == os.getpid() + except Exception: + return False def _perpetual_owner_is_alive(app) -> bool: - """True if the process that claimed the single-process perpetual loops is still running.""" + """True if the claiming worker is still running *and* keeping a heartbeat. + + pid_is_running alone is not enough: after a crash the kernel can reuse the + owner PID for an unrelated process. A zero heartbeat means the owner just + claimed and has not ticked yet, so a live PID still counts. + """ try: pid = app.shared_ctx.perpetual_tasks_owner_pid.value except Exception: @@ -178,7 +198,15 @@ def _perpetual_owner_is_alive(app) -> bool: if not pid: return False from wrolpi.cmd import pid_is_running - return pid_is_running(pid) + if not pid_is_running(pid): + return False + try: + heartbeat = app.shared_ctx.perpetual_tasks_heartbeat.value + except Exception: + return True + if not heartbeat: + return True + return (time() - heartbeat) < OWNER_HEARTBEAT_STALE_SECONDS def _claim_perpetual_tasks(app) -> bool: @@ -188,7 +216,7 @@ def _claim_perpetual_tasks(app) -> bool: 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. + a heartbeat, and retake the claim when that process is gone or silent. """ lock = getattr(app.shared_ctx, 'perpetual_tasks_lock', None) if lock is None: @@ -204,20 +232,24 @@ def _claim_perpetual_tasks_unlocked(app) -> bool: try: app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid() except Exception: - pass + logger.error('Failed to record perpetual-tasks owner pid', exc_info=True) + _touch_heartbeat(app) 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. + The claiming worker starts every perpetual loop (one file-worker consumer). + Every other worker starts only the owner watchdog, which reclaims if the + owner dies later — after_server_start will not run again on those processes. """ if _claim_perpetual_tasks(app): - return list(PERPETUAL_WORKERS) - return [FILE_WORKER_PERPETUAL_EVENT] + events = list(PERPETUAL_WORKERS) + if OWNER_WATCHDOG_EVENT not in events: + events.append(OWNER_WATCHDOG_EVENT) + return events + return [OWNER_WATCHDOG_EVENT] def _app_is_stopping() -> bool: @@ -355,3 +387,30 @@ def perpetual_signal(event: str = None, sleep: int | float = 1, run_while_testin return func return wrapper + + +async def perpetual_owner_watchdog(): + """Keep the owner heartbeat fresh, or reclaim the loops if the owner is gone. + + Non-owner Sanic workers run only this tick — not the file-worker pump — so + file jobs stay single-consumer. cancel_background_tasks / cancel_refresh_tasks + do not cancel perpetual signals; the only intentional stop is shutdown. + """ + if _app_is_stopping(): + return + if _i_am_owner(api_app): + _touch_heartbeat(api_app) + return + if not _claim_perpetual_tasks(api_app): + return + logger.warning(f'Perpetual-loop owner is gone; this worker is taking over pid={os.getpid()}') + _touch_heartbeat(api_app) + for event_ in list(PERPETUAL_WORKERS): + if event_ == OWNER_WATCHDOG_EVENT: + continue + await _dispatch_perpetual(event_) + + +# Register after perpetual_signal is defined. In PYTEST the decorator is a +# no-op, so tests call perpetual_owner_watchdog() directly. +perpetual_signal(sleep=5)(perpetual_owner_watchdog) diff --git a/wrolpi/contexts.py b/wrolpi/contexts.py index e542ded8..8d65b821 100644 --- a/wrolpi/contexts.py +++ b/wrolpi/contexts.py @@ -65,6 +65,7 @@ def attach_shared_contexts(app: Sanic): # 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_heartbeat = multiprocessing.Value(ctypes.c_double, 0.0) app.shared_ctx.perpetual_tasks_lock = multiprocessing.Lock() # Switches @@ -213,6 +214,8 @@ def reset_shared_contexts(app: Sanic): 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 + if hasattr(app.shared_ctx, 'perpetual_tasks_heartbeat'): + app.shared_ctx.perpetual_tasks_heartbeat.value = 0.0 # Do not start downloads when reloading. app.shared_ctx.download_manager_stopped.set() diff --git a/wrolpi/files/api.py b/wrolpi/files/api.py index 2a8875f4..2ae851db 100644 --- a/wrolpi/files/api.py +++ b/wrolpi/files/api.py @@ -340,7 +340,8 @@ async def post_rename(request: Request, body: schema.Rename): await lib.rename(request.ctx.session, path, body.new_name, send_events=True) except Exception as e: logger.error(f'Failed to rename {path} to {new_path}', exc_info=e) - raise FileConflict(f'Failed to rename {path} to {new_path}') from e + cause = str(e).split('\n', 1)[0] + raise FileConflict(f'Failed to rename {path} to {new_path}: {cause}') from e # Refresh the parent directory instead of individual files. # This ensures all related files (e.g., subtitles, posters) are discovered together. diff --git a/wrolpi/files/test/test_api.py b/wrolpi/files/test/test_api.py index d077a5b6..4ead049e 100644 --- a/wrolpi/files/test/test_api.py +++ b/wrolpi/files/test/test_api.py @@ -1290,6 +1290,9 @@ async def test_rename_directory_stale_destination_fails_promptly( timeout=5, ) assert response.status_code == HTTPStatus.CONFLICT, response.json + error = (response.json or {}).get('error') or '' + # The recorded move error, not just "Failed to rename …". + assert 'UNIQUE' in error.upper() or 'constraint' in error.lower(), response.json @pytest.mark.asyncio diff --git a/wrolpi/files/test/test_worker.py b/wrolpi/files/test/test_worker.py index 495f4451..2c721ffb 100644 --- a/wrolpi/files/test/test_worker.py +++ b/wrolpi/files/test/test_worker.py @@ -682,6 +682,29 @@ 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_polls_slowly_outside_tests(async_client, monkeypatch): + """Production wait_for_job must not hammer the Manager dict at 100 Hz. + + Each get_job_status is an IPC round-trip. Tests keep a short sleep so + they stay fast; production can wait a tenth of a second between polls. + """ + job_id = 'move-poll-interval' + file_worker._set_job_status(job_id, 'pending') + sleeps = [] + + async def fake_sleep(seconds): + sleeps.append(seconds) + file_worker._complete_job(job_id) + + monkeypatch.setattr('wrolpi.files.worker.PYTEST', False) + monkeypatch.setattr('wrolpi.files.worker.asyncio.sleep', fake_sleep) + + await file_worker.wait_for_job(job_id, timeout=1) + assert sleeps, 'wait_for_job should sleep between polls' + assert sleeps[0] == pytest.approx(0.1, abs=0.01) + + @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. diff --git a/wrolpi/files/worker.py b/wrolpi/files/worker.py index 0b02f959..c175cf7a 100644 --- a/wrolpi/files/worker.py +++ b/wrolpi/files/worker.py @@ -909,7 +909,8 @@ class FileWorker: if PYTEST: self.transfer_queue() await self.process_queue() - await asyncio.sleep(0.01) + # Tests poll tightly; production is an IPC round-trip per check. + await asyncio.sleep(0.01 if PYTEST else 0.1) 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') @@ -1081,7 +1082,8 @@ class FileWorker: ``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. + stayed non-empty while status is idle. This runs inside the pump, so + it cannot report the pump's own death. """ if processed: self._queue_stall_since = None diff --git a/wrolpi/test/test_api_utils.py b/wrolpi/test/test_api_utils.py index c8464866..eb237c0d 100644 --- a/wrolpi/test/test_api_utils.py +++ b/wrolpi/test/test_api_utils.py @@ -123,24 +123,77 @@ def test_claim_perpetual_tasks_when_owner_process_is_dead(async_client): '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. +def test_claim_perpetual_tasks_when_heartbeat_is_stale(async_client): + """A live PID with a stale heartbeat is treated as a dead owner. - Otherwise a live owner whose *task* died (CancelledError before the - reschedule fix) still wedges file operations in every other process. + After a crash the kernel can reuse the owner PID for an unrelated + process, so pid_is_running alone is not enough. The owner must keep + a heartbeat; if it goes quiet, another worker reclaims. """ import os + import time + from wrolpi.api_utils import OWNER_HEARTBEAT_STALE_SECONDS, _claim_perpetual_tasks, api_app + + api_app.shared_ctx.perpetual_tasks_started.set() + api_app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid() + api_app.shared_ctx.perpetual_tasks_heartbeat.value = time.time() - OWNER_HEARTBEAT_STALE_SECONDS - 1 + assert _claim_perpetual_tasks(api_app) is True + + +def test_non_owner_starts_watchdog_not_file_worker(async_client): + """Non-owner workers run a reclaim watchdog, not a second file-worker pump. + + Five concurrent pumps would race on the shared status dict, the busy + flag, and the DB — overlapping refreshes/moves are how stale + primary_path rows appear. File jobs stay single-consumer; other + workers only notice a dead owner and take over. + """ + import os + import time from wrolpi.api_utils import ( FILE_WORKER_PERPETUAL_EVENT, + OWNER_WATCHDOG_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() + api_app.shared_ctx.perpetual_tasks_heartbeat.value = time.time() events = _perpetual_events_for_this_process(api_app) - assert events == [FILE_WORKER_PERPETUAL_EVENT] + assert events == [OWNER_WATCHDOG_EVENT] + assert FILE_WORKER_PERPETUAL_EVENT not in events + + +@pytest.mark.asyncio +async def test_watchdog_reclaims_and_starts_pumps_when_owner_is_dead(async_client, monkeypatch): + """The watchdog, not after_server_start, is what heals a dead owner mid-life. + + Existing workers have already passed start_perpetual_tasks. If the + owner dies later, a periodic tick must claim and dispatch the pumps. + """ + import os + from wrolpi.api_utils import perpetual_owner_watchdog, api_app + + dispatched = [] + + async def fake_dispatch(event): + dispatched.append(event) + + api_app.shared_ctx.perpetual_tasks_started.set() + api_app.shared_ctx.perpetual_tasks_owner_pid.value = 2**31 - 1 + monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch) + monkeypatch.setattr('wrolpi.api_utils.PERPETUAL_WORKERS', [ + 'wrolpi.perpetual.perpetual_file_worker_queue', + 'wrolpi.perpetual.perpetual_owner_watchdog', + ]) + + await perpetual_owner_watchdog() + + assert api_app.shared_ctx.perpetual_tasks_owner_pid.value == os.getpid() + assert api_app.shared_ctx.perpetual_tasks_heartbeat.value > 0 + assert 'wrolpi.perpetual.perpetual_file_worker_queue' in dispatched + assert 'wrolpi.perpetual.perpetual_owner_watchdog' not in dispatched @pytest.mark.asyncio