perpetual: clear in-progress flags after a crash; downloads only in the perpetual process

Found by a kill -9 stress test on a device:

A SIGKILL inside `with flags.file_worker_counting` (or any of the file
worker's progress flags) left the shared flag set, so every later refresh
raised "flag is already set" until the whole API was restarted.
reconcile_after_predecessor now clears the perpetual process's in-progress
flags (file_worker_*, global_refresh_active) and reports it.

create_downloads started a dispatch in whatever process called it, so an
API worker executed downloads (and could take transcode_lock) despite the
single-process design.  It now dispatches only inside the perpetual
process (or under pytest); the download loop picks new rows up within
seconds otherwise.

The shared restart cooldown drops from 60 s to 15 s: it only has to
outlast the manager acting on a request, and a replacement that dies
right away was left dead for the rest of the minute.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

wrolpi/downloader.py

@@ -998,8 +998,12 @@ class DownloadManager:
 
         session.flush(downloads)
         try:
-            # Start downloading ASAP.
-            background_task(self.dispatch_downloads())
+            from wrolpi import perpetual
+            if perpetual.IN_PERPETUAL_PROCESS or PYTEST:
+                # Start downloading ASAP.  Only the perpetual process runs downloads; a request worker leaves
+                # them for perpetual_download_worker's next cycle (a few seconds) rather than running a download
+                # inside the API worker.
+                background_task(self.dispatch_downloads())
             # Save the config now that new Downloads exist.
             save_downloads_config.activate_switch()
         except RuntimeError:

wrolpi/perpetual.py

@@ -36,8 +36,10 @@ PERPETUAL_PROCESS_IDENT = f'Sanic-{PERPETUAL_PROCESS_NAME}-0'
 # A managed (non-server) process never reaches ACKED; these are the states the manager reports for one that
 # is gone.  See sanic/worker/constants.py ProcessState.
 PERPETUAL_PROCESS_DEAD_STATES = frozenset({'FAILED', 'COMPLETED', 'TERMINATED'})
-# A restart request is idempotent to the manager, but five workers noticing one death should not log five times.
-PERPETUAL_RESTART_COOLDOWN = 60
+# One restart request per death across all workers.  The cooldown only has to outlast the manager acting on the
+# request (the replacement appears within a second or two); a longer one leaves the loops dead that much longer
+# when the replacement dies too.
+PERPETUAL_RESTART_COOLDOWN = 15
 # How long the runner waits for cancelled loops before abandoning them at shutdown.
 PERPETUAL_CANCEL_TIMEOUT = 15
 # Shared multiprocessing locks whose critical sections are short.  A predecessor SIGKILLed inside one leaves it
@@ -45,6 +47,15 @@ PERPETUAL_CANCEL_TIMEOUT = 15
 SHARED_LOCK_NAMES = ('jobs_lock', 'switches_lock', 'config_save_lock', 'config_update_lock', 'events_lock',
                      'secure_cookies_lock', 'transcode_lock')
 SHARED_LOCK_HEAL_TIMEOUT = 10
+# Shared flags that mean "the perpetual process is in the middle of this".  They are set and cleared by context
+# managers in this process only; a SIGKILL inside one leaves it set, and the next attempt raises "flag is already
+# set" forever (every file refresh, for instance).
+PERPETUAL_PROGRESS_FLAGS = ('file_worker_busy', 'file_worker_counting', 'file_worker_discovery',
+                            'file_worker_modeling', 'file_worker_indexing', 'file_worker_cleanup',
+                            'global_refresh_active')
+
+# True only inside the process started by `run_perpetual_process`.
+IN_PERPETUAL_PROCESS = False
 
 
 @dataclass
@@ -157,14 +168,26 @@ def reconcile_after_predecessor(app) -> dict:
     refresh that is not happening.  On a first start all of this is already clean, so this is idempotent.
     Each step is independent; one failing (e.g. no database yet) must not stop the others.
     """
+    from wrolpi import flags
     from wrolpi.downloader import download_manager, Download, DownloadStatus
     from wrolpi.files.worker import file_worker
     from wrolpi.db import get_db_session
 
-    report = dict(locks=[], jobs=0, downloads=0, processing_domains=0, file_worker_jobs=0)
+    report = dict(locks=[], flags=[], jobs=0, downloads=0, processing_domains=0, file_worker_jobs=0,
+                  file_worker_status=False)
 
     report['locks'] = heal_shared_locks(app)
 
+    for name in PERPETUAL_PROGRESS_FLAGS:
+        try:
+            flag = getattr(flags, name)
+            if flag.is_set():
+                logger.warning(f'reconcile: clearing in-progress flag {name} left set by a dead process')
+                flag.clear()
+                report['flags'].append(name)
+        except Exception as e:
+            logger.error(f'reconcile: failed to clear flag {name}', exc_info=e)
+
     try:
         from wrolpi.jobs import fail_orphaned_jobs
         report['jobs'] = fail_orphaned_jobs()
@@ -200,6 +223,7 @@ def reconcile_after_predecessor(app) -> dict:
                 report['file_worker_jobs'] += 1
         if report['file_worker_jobs'] or (file_worker.status and file_worker.status.get('status') != 'idle'):
             file_worker.reset_status()
+            report['file_worker_status'] = True
     except Exception as e:
         logger.error('reconcile: failed to reset the file worker status', exc_info=e)
 
@@ -239,6 +263,8 @@ def run_perpetual_process():
     the populated `shared_ctx`, and the module singletons, and needs no arguments."""
     # The fork inherited the manager's Python signal handlers; until the event loop installs ours, a signal
     # must do the default thing (exit) rather than run the manager's handler inside this process.
+    global IN_PERPETUAL_PROCESS
+    IN_PERPETUAL_PROCESS = True
     signal.signal(signal.SIGTERM, signal.SIG_DFL)
     signal.signal(signal.SIGINT, signal.SIG_DFL)
     if not PERPETUAL_LOOPS:

wrolpi/test/test_downloader.py

@@ -2158,3 +2158,22 @@ async def test_renew_recurring_downloads_scans_without_the_write_lock(test_sessi
     assert lock_held_during_scan, 'the recurring downloads were never scanned'
     assert not any(lock_held_during_scan), \
         'the write lock was held while scanning recurring downloads'
+
+
+@pytest.mark.asyncio
+async def test_create_downloads_does_not_dispatch_outside_the_perpetual_process(test_session, test_download_manager,
+                                                                                 test_downloader, monkeypatch):
+    """Only the perpetual process runs downloads.  A request worker creating Downloads must not start one itself;
+    perpetual_download_worker picks them up on its next cycle."""
+    from wrolpi import perpetual
+    monkeypatch.setattr('wrolpi.downloader.PYTEST', False)
+    monkeypatch.setattr(perpetual, 'IN_PERPETUAL_PROCESS', False)
+    with mock.patch.object(test_download_manager, 'dispatch_downloads') as dispatch:
+        test_download_manager.create_downloads(test_session, ['https://example.com/worker'], test_downloader.name)
+    dispatch.assert_not_called()
+
+    monkeypatch.setattr(perpetual, 'IN_PERPETUAL_PROCESS', True)
+    with mock.patch.object(test_download_manager, 'dispatch_downloads', new_callable=mock.AsyncMock) as dispatch:
+        test_download_manager.create_downloads(test_session, ['https://example.com/perpetual'], test_downloader.name)
+        await asyncio.sleep(0)
+    dispatch.assert_called_once()

wrolpi/test/test_perpetual.py

@@ -188,9 +188,9 @@ async def test_health_check_without_multiplexer_is_a_noop(async_client, monkeypa
     await perpetual.perpetual_process_health_check()
 
 
-def _wait_for_byte(fd: int, timeout: float) -> bool:
+def _wait_for_byte(fd: int, timeout: float) -> bytes:
     ready, _, _ = select.select([fd], [], [], timeout)
-    return bool(ready) and bool(os.read(fd, 1))
+    return os.read(fd, 1) if ready else b''
 
 
 @pytest.mark.parametrize('sig', [signal.SIGINT, signal.SIGTERM], ids=['SIGINT (manager shutdown)', 'SIGTERM (restart)'])
@@ -202,16 +202,18 @@ def test_run_perpetual_process_exits_cleanly_on_signal(async_client, monkeypatch
     async def loop_with_live_background_task():
         from wrolpi.common import background_task
         background_task(asyncio.sleep(3600))
-        os.write(ready_w, b'x')
+        # The byte doubles as the assertion that the entry point marked this process as the perpetual one.
+        os.write(ready_w, b'x' if perpetual.IN_PERPETUAL_PROCESS else b'n')
 
     monkeypatch.setattr(perpetual, 'PERPETUAL_LOOPS', [PerpetualLoop('bg', loop_with_live_background_task, 3600)])
+    assert perpetual.IN_PERPETUAL_PROCESS is False, 'the test process is not the perpetual process'
     # The forked test process must not touch the test database; the reconcile step has its own tests below.
     monkeypatch.setattr(perpetual, 'reconcile_after_predecessor', lambda app: {})
 
     process = multiprocessing.get_context('fork').Process(target=run_perpetual_process)
     process.start()
     try:
-        assert _wait_for_byte(ready_r, 10), 'the perpetual process did not start its loop'
+        assert _wait_for_byte(ready_r, 10) == b'x', 'the perpetual process did not start its loop as the perpetual process'
         os.kill(process.pid, sig)
         process.join(15)
         assert process.exitcode == 0, f'perpetual process exited {process.exitcode}'
@@ -293,12 +295,12 @@ def test_sanic_manager_marks_failed_and_restart_revives(test_directory):
 
 
 @pytest.mark.asyncio
-async def test_reconcile_after_predecessor(async_client, test_session, monkeypatch):
+async def test_reconcile_after_predecessor(async_client, test_session, monkeypatch, flags_lock):
     """A restarted perpetual process cleans up what a predecessor that died mid-work left behind."""
     from wrolpi.api_utils import api_app
     from wrolpi.downloader import Download, download_manager
     from wrolpi.files.worker import file_worker
-    from wrolpi import jobs
+    from wrolpi import jobs, flags
 
     # A domain still claimed, a Download still pending, a Job still running under a dead pid, a file job running,
     # and the jobs lock still held by the dead process.
@@ -312,6 +314,9 @@ async def test_reconcile_after_predecessor(async_client, test_session, monkeypat
                                            created_at='', started_at='', finished_at=None, error=None)
     file_worker._set_job_status('fj', 'running')
     file_worker.update_status(status='refreshing')
+    flags.file_worker_counting.set()
+    flags.global_refresh_active.set()
+    flags.db_up.set()  # not an in-progress flag; must survive
     assert api_app.shared_ctx.jobs_lock.acquire(timeout=1)
     monkeypatch.setattr(perpetual, 'SHARED_LOCK_HEAL_TIMEOUT', 0.1)
 
@@ -329,7 +334,12 @@ async def test_reconcile_after_predecessor(async_client, test_session, monkeypat
     assert {d.url: d.status for d in test_session.query(Download)} == \
            {'https://example.com/a': 'new', 'https://example.com/b': 'new'}
     assert report['file_worker_jobs'] == 1 and file_worker._jobs['fj']['status'] == 'failed'
-    assert file_worker.status['status'] == 'idle'
+    assert file_worker.status['status'] == 'idle' and report['file_worker_status'] is True
+    assert report['flags'] == ['file_worker_counting', 'global_refresh_active']
+    assert not flags.file_worker_counting.is_set() and not flags.global_refresh_active.is_set()
+    assert flags.db_up.is_set(), 'only in-progress flags are cleared'
+    with flags.file_worker_counting:  # usable again; this raised "flag is already set" before
+        pass
 
     # Idempotent: a clean state reports nothing.
     assert not any(reconcile_after_predecessor(api_app).values())