files: one file-worker consumer, a watchdog for the rest

Starting the file-worker pump in every Sanic worker made 5 processes
race the shared status dict, the busy flag, and the same directories.
Non-owners now run only a reclaim watchdog: they refresh nothing unless
the owner PID is gone or its heartbeat is stale.

Production wait_for_job polls every 0.1s instead of 100 Hz. A failed
rename includes the recorded job error in the 409 body.

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)

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

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.

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

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.

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

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