perpetual: clear in-progress flags after a crash, downloads only in the perpetual process, KillMode=mixed (#726)

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

* systemd: KillMode=mixed for the API so main shuts its own children down

With control-group kill, SIGTERM reaches the multiprocessing Manager at
the same moment as Sanic's main process; main's cleanup then writes
worker state to a dead Manager (broken pipe) and exits 1, so a plain
`systemctl stop` is recorded as a failure whenever a child takes a
moment to stop (a transcode being killed, for instance).  Signal only
main; it stops the workers and the perpetual process itself.

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

* perpetual: clear an in-progress flag only when its holder is dead

`with flag:` now records the holder pid in shared_ctx.flag_holders and
drops it on exit.  The reconcile step clears the flags only the perpetual
process ever enters (file_worker_counting, file_worker_discovery,
global_refresh_active) whenever they are set, and the flags API workers
also enter (file_worker_modeling / indexing / cleanup via refresh_sync,
map_search_building via the map rebuild endpoints) only when the
recorded holder is dead.  file_worker_busy is no longer touched: a
collection move in an API worker holds it as a mutex, and
process_queue drops a stale one itself.

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

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

etc/raspberrypios/wrolpi-api-dev.service

@@ -15,6 +15,10 @@ Group=wrolpi
 User=wrolpi
 WorkingDirectory=/opt/wrolpi/
 Restart=on-failure
+# Signal only Sanic's main process on stop; it shuts its workers, the perpetual process and the multiprocessing
+# Manager down itself.  Signalling the whole cgroup at once kills the Manager under main's own cleanup (broken
+# pipe, exit 1).  Stragglers still get SIGKILL at TimeoutStopSec.
+KillMode=mixed
 # Higher than the App so the API can start.
 Nice=4
 

etc/raspberrypios/wrolpi-api.service

@@ -15,6 +15,10 @@ Group=wrolpi
 User=wrolpi
 WorkingDirectory=/opt/wrolpi/
 Restart=on-failure
+# Signal only Sanic's main process on stop; it shuts its workers, the perpetual process and the multiprocessing
+# Manager down itself.  Signalling the whole cgroup at once kills the Manager under main's own cleanup (broken
+# pipe, exit 1).  Stragglers still get SIGKILL at TimeoutStopSec.
+KillMode=mixed
 # Higher than the App so the API can start.
 Nice=4
 

wrolpi/contexts.py

@@ -20,6 +20,8 @@ def attach_shared_contexts(app: Sanic):
     from wrolpi import flags
 
     app.shared_ctx.flags = manager.dict({i: False for i in flags.FLAG_NAMES})
+    # Flag name -> pid of the process inside `with flag:` (see Flag.holder_pid).
+    app.shared_ctx.flag_holders = manager.dict()
 
     # ConfigFile multiprocessing_dict's.
     # Shared Configs
@@ -230,6 +232,8 @@ def reset_shared_contexts(app: Sanic):
     del app.shared_ctx.events_history[:]
     app.shared_ctx.single_tasks_started.clear()
     app.shared_ctx.flags_initialized.clear()
+    if hasattr(app.shared_ctx, 'flag_holders'):
+        app.shared_ctx.flag_holders.clear()
     if hasattr(app.shared_ctx, 'perpetual_restart_requested_at'):
         app.shared_ctx.perpetual_restart_requested_at.value = 0.0
 

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/flags.py

@@ -1,6 +1,7 @@
 import asyncio
 import contextlib
 import multiprocessing
+import os
 from datetime import datetime
 from pathlib import Path
 
@@ -68,6 +69,27 @@ class Flag:
         from wrolpi.api_utils import api_app
         return api_app.shared_ctx.flags[self.name]
 
+    def holder_pid(self) -> int | None:
+        """The pid of the process inside `with flag:`, or None if no process is (or the holder was not recorded).
+
+        A process killed inside the block never reaches `__exit__`; the perpetual process uses this on startup to
+        tell an orphaned flag from one a live process is still holding."""
+        if PYTEST and not TESTING_LOCK.is_set():
+            return None
+        from wrolpi.api_utils import api_app
+        holders = getattr(api_app.shared_ctx, 'flag_holders', None)
+        return holders.get(self.name) if holders is not None else None
+
+    def _record_holder(self, pid: int | None):
+        from wrolpi.api_utils import api_app
+        holders = getattr(api_app.shared_ctx, 'flag_holders', None)
+        if holders is None:
+            return
+        if pid is None:
+            holders.pop(self.name, None)
+        else:
+            holders[self.name] = pid
+
     def __enter__(self):
         if PYTEST and not TESTING_LOCK.is_set():
             # Testing, but the test does not need flags.
@@ -76,6 +98,7 @@ class Flag:
         if self.is_set():
             raise ValueError(f'{self} flag is already set!')
         self.set()
+        self._record_holder(os.getpid())
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         if PYTEST and not TESTING_LOCK.is_set():
@@ -83,6 +106,7 @@ class Flag:
             return
 
         self.clear()
+        self._record_holder(None)
 
     def _save(self, value: bool):
         if self.store_db:

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,20 @@ 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
+# In-progress flags.  A SIGKILL inside `with flag:` leaves it set and the next enter raises "flag is already set"
+# forever (every file refresh, or every map search-index build).  Two kinds:
+#  * entered or set only by the perpetual process (FileWorker.process_queue, handle_refresh): at this process's
+#    startup any of them still set is orphaned, whoever the recorded holder is;
+#  * also entered by API workers (refresh_sync from the delete endpoints, the map rebuild endpoints): cleared only
+#    when the recorded holder pid is dead, because a live worker may be inside the block right now.
+# file_worker_busy is deliberately absent: the collection-move path in an API worker holds it as a mutex, and
+# process_queue already drops a stale set on its next idle tick.
+PERPETUAL_ONLY_FLAGS = ('file_worker_counting', 'file_worker_discovery', 'global_refresh_active')
+SHARED_PROGRESS_FLAGS = ('file_worker_modeling', 'file_worker_indexing', 'file_worker_cleanup',
+                         'map_search_building')
+
+# True only inside the process started by `run_perpetual_process`.
+IN_PERPETUAL_PROCESS = False
 
 
 @dataclass
@@ -149,6 +165,37 @@ def heal_shared_locks(app, timeout: float = SHARED_LOCK_HEAL_TIMEOUT) -> list[st
     return healed
 
 
+def clear_orphaned_flags() -> list[str]:
+    """Clear in-progress flags whose holder is gone; returns the names cleared.  See PERPETUAL_ONLY_FLAGS."""
+    from wrolpi import flags
+    from wrolpi.cmd import pid_is_running
+
+    cleared = []
+    for name in PERPETUAL_ONLY_FLAGS + SHARED_PROGRESS_FLAGS:
+        try:
+            flag = getattr(flags, name)
+            if not flag.is_set():
+                continue
+            holder = flag.holder_pid()
+            if name in PERPETUAL_ONLY_FLAGS:
+                reason = f'holder pid={holder}' if holder else 'no live holder possible'
+            elif holder is None:
+                logger.warning(f'reconcile: flag {name} is set with no recorded holder; leaving it')
+                continue
+            elif pid_is_running(holder) and holder != os.getpid():
+                logger.info(f'reconcile: flag {name} is held by live pid={holder}; leaving it')
+                continue
+            else:
+                reason = f'holder pid={holder} is dead'
+            logger.warning(f'reconcile: clearing in-progress flag {name} ({reason})')
+            flag.clear()
+            flag._record_holder(None)
+            cleared.append(name)
+        except Exception as e:
+            logger.error(f'reconcile: failed to check flag {name}', exc_info=e)
+    return cleared
+
+
 def reconcile_after_predecessor(app) -> dict:
     """Undo what a perpetual process that died mid-work left in shared state and the DB.
 
@@ -157,14 +204,18 @@ 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)
 
+    report['flags'] = clear_orphaned_flags()
+
     try:
         from wrolpi.jobs import fail_orphaned_jobs
         report['jobs'] = fail_orphaned_jobs()
@@ -200,6 +251,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 +291,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,15 @@ 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()          # perpetual-only: cleared whoever held it
+    flags.global_refresh_active.set()         # perpetual-only, set() not entered: cleared
+    flags.file_worker_modeling.set()          # shared, holder dead: cleared
+    api_app.shared_ctx.flag_holders['file_worker_modeling'] = 2 ** 31 - 1
+    flags.map_search_building.set()           # shared, holder is a live API worker: kept
+    api_app.shared_ctx.flag_holders['map_search_building'] = os.getppid()
+    flags.file_worker_indexing.set()          # shared, no holder recorded: kept (cannot tell)
+    flags.file_worker_busy.set()              # not a reconciled flag (a live move may hold it): kept
+    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 +340,32 @@ 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', 'file_worker_modeling']
+    assert not flags.file_worker_counting.is_set() and not flags.global_refresh_active.is_set()
+    assert not flags.file_worker_modeling.is_set() and 'file_worker_modeling' not in api_app.shared_ctx.flag_holders
+    assert flags.map_search_building.is_set(), 'a flag held by a live process is not touched'
+    assert flags.file_worker_indexing.is_set(), 'a shared flag with no recorded holder is not touched'
+    assert flags.file_worker_busy.is_set(), 'file_worker_busy is not reconciled'
+    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
+    for name in ('map_search_building', 'file_worker_indexing', 'file_worker_busy'):
+        getattr(flags, name).clear()
+    api_app.shared_ctx.flag_holders.clear()
 
     # Idempotent: a clean state reports nothing.
     assert not any(reconcile_after_predecessor(api_app).values())
+
+
+def test_flag_context_manager_records_its_holder(async_client, flags_lock):
+    """`with flag:` records this pid so a later reconcile can tell an orphaned flag from a live holder."""
+    from wrolpi import flags
+    from wrolpi.api_utils import api_app
+
+    assert flags.file_worker_modeling.holder_pid() is None
+    with flags.file_worker_modeling:
+        assert flags.file_worker_modeling.holder_pid() == os.getpid()
+        assert api_app.shared_ctx.flag_holders['file_worker_modeling'] == os.getpid()
+    assert flags.file_worker_modeling.holder_pid() is None
+    assert not flags.file_worker_modeling.is_set()