From 811793d76738d38fa8b51c6e15b4f0a7bbd69655 Mon Sep 17 00:00:00 2001 From: lrnselfreliance <33263708+lrnselfreliance@users.noreply.github.com> Date: Sat, 26 Sep 2026 11:22:58 -0600 Subject: [PATCH] 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 * 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 * 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 --------- Co-authored-by: Claude Fable 5.1 --- etc/raspberrypios/wrolpi-api-dev.service | 4 ++ etc/raspberrypios/wrolpi-api.service | 4 ++ wrolpi/contexts.py | 4 ++ wrolpi/downloader.py | 8 +++- wrolpi/flags.py | 24 ++++++++++ wrolpi/perpetual.py | 60 ++++++++++++++++++++++-- wrolpi/test/test_downloader.py | 19 ++++++++ wrolpi/test/test_perpetual.py | 50 +++++++++++++++++--- 8 files changed, 161 insertions(+), 12 deletions(-) diff --git a/etc/raspberrypios/wrolpi-api-dev.service b/etc/raspberrypios/wrolpi-api-dev.service index 4e8de8ff..2f4c8fae 100644 --- a/etc/raspberrypios/wrolpi-api-dev.service +++ b/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 diff --git a/etc/raspberrypios/wrolpi-api.service b/etc/raspberrypios/wrolpi-api.service index dbc5e015..724e606f 100644 --- a/etc/raspberrypios/wrolpi-api.service +++ b/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 diff --git a/wrolpi/contexts.py b/wrolpi/contexts.py index d961cdf2..2f55cabc 100644 --- a/wrolpi/contexts.py +++ b/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 diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index b6e937a6..36937f17 100644 --- a/wrolpi/downloader.py +++ b/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: diff --git a/wrolpi/flags.py b/wrolpi/flags.py index dbc59a3b..049a288f 100644 --- a/wrolpi/flags.py +++ b/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: diff --git a/wrolpi/perpetual.py b/wrolpi/perpetual.py index d5cfbaeb..afa76abb 100644 --- a/wrolpi/perpetual.py +++ b/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: diff --git a/wrolpi/test/test_downloader.py b/wrolpi/test/test_downloader.py index 9abecc50..00810417 100644 --- a/wrolpi/test/test_downloader.py +++ b/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() diff --git a/wrolpi/test/test_perpetual.py b/wrolpi/test/test_perpetual.py index 423e9ce9..233704c8 100644 --- a/wrolpi/test/test_perpetual.py +++ b/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()