diff --git a/main.py b/main.py index 2b84c729..3360886d 100755 --- a/main.py +++ b/main.py @@ -12,7 +12,7 @@ from sanic.signals import Event from modules.videos.lib import get_videos_downloader_config from wrolpi import flags from wrolpi import root_api # noqa -from wrolpi.api_utils import api_app, perpetual_signal +from wrolpi.api_utils import api_app, perpetual_signal, per_worker_task from wrolpi.common import logger, check_media_directory, set_log_level, limit_concurrent, \ cancel_refresh_tasks, cancel_background_tasks, get_wrolpi_config, can_connect_to_server, wrol_mode_enabled, \ create_empty_config_files, TRACE_LEVEL @@ -22,6 +22,7 @@ from modules.videos.cookies import lock_cookies from wrolpi.downloader import download_manager, get_download_manager_config from wrolpi.errors import WROLModeEnabled from wrolpi.files.worker import file_worker +from wrolpi.perpetual import run_perpetual_process, PERPETUAL_PROCESS_NAME, perpetual_process_health_check from wrolpi.vars import PROJECT_DIR, DOCKERIZED, INTERNET_SERVER from wrolpi.version import get_version_string @@ -168,11 +169,12 @@ def main(): @api_app.main_process_ready async def main_process_startup(app: Sanic): """ - Initializes multiprocessing tools, flags, etc. + Initializes multiprocessing tools, flags, etc., then registers the perpetual process. Performed only once when the server starts, this is done before server processes are forked. - @warning: This is NOT run after auto-reload! You must stop and start Sanic. + @warning: This is NOT run after auto-reload! You must stop and start Sanic. (The perpetual process is + `transient`, so the manager does restart it on auto-reload.) """ logger.debug('main_process_startup') @@ -199,6 +201,10 @@ async def main_process_startup(app: Sanic): except Exception as e: logger.error('Failed to create/migrate the database', exc_info=e) + # The singleton background loops run in ONE process the Worker Manager owns (not in a server worker), so no + # worker can claim or take them over. `shared_ctx` is populated above; the fork inherits it. + app.manager.manage(PERPETUAL_PROCESS_NAME, run_perpetual_process, {}, transient=True) + @api_app.listener('after_server_start') # FileConfigs need to be initialized first. async def initialize_configs(app: Sanic): @@ -220,7 +226,9 @@ async def initialize_configs(app: Sanic): @api_app.listener('reload_process_stop') @limit_concurrent(1) async def handle_server_shutdown(*args, **kwargs): - """Stop downloads and securely clear sensitive data when server is shutting down.""" + """Stop downloads and securely clear sensitive data when server is shutting down. + + Runs in each server worker. The perpetual process does the same for itself (wrolpi/perpetual.py).""" logger.warning('Shutting down') try: download_manager.stop() @@ -324,7 +332,7 @@ async def perpetual_file_worker_queue(): file_worker.check_queue_stall(processed=processed) -@perpetual_signal(sleep=1) +@per_worker_task(sleep=1) async def perpetual_check_log_level(): """Copies global log level into this Sanic worker's logger.""" log_level = api_app.shared_ctx.log_level.value @@ -333,6 +341,10 @@ async def perpetual_check_log_level(): set_log_level(log_level, warn_level=False) +# Every server worker watches the perpetual process and asks the manager to restart it if it dies. +per_worker_task(sleep=10)(perpetual_process_health_check) + + @perpetual_signal(sleep=10) async def perpetual_check_db_is_up_worker(): try: @@ -416,7 +428,7 @@ async def perpetual_check_for_updates(): Only runs on native installs (not Docker). """ # Local import to avoid loading upgrade module at startup - from wrolpi.upgrade import check_for_update + from wrolpi.upgrade import refresh_update_status # Don't check for updates in Docker environments if DOCKERIZED: @@ -431,13 +443,8 @@ async def perpetual_check_for_updates(): return try: - result = check_for_update(fetch=True) - # Store in shared_ctx.status (a manager.dict) to share across all workers. - api_app.shared_ctx.status['update_available'] = result.get('update_available', False) - api_app.shared_ctx.status['latest_commit'] = result.get('latest_commit') - api_app.shared_ctx.status['current_commit'] = result.get('current_commit') - api_app.shared_ctx.status['commits_behind'] = result.get('commits_behind', 0) - api_app.shared_ctx.status['git_branch'] = result.get('branch') + # Runs `git fetch` in a thread and stores the result in shared_ctx.status for all workers. + result = await refresh_update_status(api_app.shared_ctx.status) if result.get('update_available'): logger.info(f"Update available: {result['commits_behind']} commits behind on {result['branch']}") diff --git a/modules/videos/channel/test/test_api.py b/modules/videos/channel/test/test_api.py index e608fa6e..883f41bf 100644 --- a/modules/videos/channel/test/test_api.py +++ b/modules/videos/channel/test/test_api.py @@ -211,8 +211,6 @@ async def test_channel_download_requires_refresh( assert channel.refreshed == expected assert bool(channel.info_json) == expected - await async_client.sanic_app.dispatch('wrolpi.download.download') - assert_refreshed(False) test_session.commit() diff --git a/wrolpi/api_utils.py b/wrolpi/api_utils.py index 34daf5af..c07eebf6 100644 --- a/wrolpi/api_utils.py +++ b/wrolpi/api_utils.py @@ -2,18 +2,17 @@ import asyncio import json import logging import os -from asyncio import CancelledError from datetime import datetime, timezone, date from decimal import Decimal from functools import wraps from http import HTTPStatus from pathlib import Path -from time import time from sanic import response, HTTPResponse, Request, Sanic, SanicException -from wrolpi.common import Base, get_media_directory, logger, LOGGING_CONFIG, TRACE_LEVEL +from wrolpi.common import Base, get_media_directory, logger, LOGGING_CONFIG from wrolpi.errors import APIError +from wrolpi.perpetual import PERPETUAL_LOOPS, PER_WORKER_TASKS, PerpetualLoop, run_loop from wrolpi.vars import PYTEST logger = logger.getChild(__name__) @@ -163,254 +162,51 @@ async def cleanup_session(request: Request, response_: HTTPResponse): return response_ -PERPETUAL_WORKERS = list() +def perpetual_signal(sleep: int | float = 1, run_while_testing: bool = False): + """Register a singleton background loop. The wrapped function is called forever, `sleep` seconds after each + call returns, in the ONE perpetual process that Sanic's Worker Manager owns (see wrolpi/perpetual.py). Errors + are logged and the loop continues; a long call is not re-entered until it returns. -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 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: - return False - if not pid: - return False - from wrolpi.cmd import pid_is_running - 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: - """Return True if this process should start the single-process perpetual loops. - - ``perpetual_tasks_started`` is a shared Event. If the worker that set it - 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 - 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: - return _claim_perpetual_tasks_unlocked(app) - with lock: - return _claim_perpetual_tasks_unlocked(app) - - -def _claim_perpetual_tasks_unlocked(app) -> bool: - if app.shared_ctx.perpetual_tasks_started.is_set() and _perpetual_owner_is_alive(app): - return False - app.shared_ctx.perpetual_tasks_started.set() - try: - app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid() - except Exception: - 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 (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): - 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: - """True when Sanic is shutting down (perpetual workers should not reschedule). - - Only the ``before_server_stop`` flag (and Sanic's ``is_stopping`` if it is - ever set) count. ``state.is_running`` stays False in some serving modes - (including this project's docker Sanic workers), so treating - ``is_started and not is_running`` as shutdown killed every perpetual - worker after its first tick. - """ - try: - if getattr(api_app.ctx, 'perpetual_shutdown', False): - return True - state = getattr(api_app, 'state', None) - return bool(state is not None and getattr(state, 'is_stopping', False)) - except Exception: - return False - - -def _uncancel_current_task(): - """Allow a perpetual worker to continue after an unexpected CancelledError. - - Python 3.11+ re-raises CancelledError at the next await unless uncancel() - is called. Without this, catching CancelledError cannot reschedule. - """ - task = asyncio.current_task() - if task is not None and hasattr(task, 'uncancel'): - task.uncancel() - - -@api_app.listener('before_server_start') -async def _clear_perpetual_shutdown(app): - app.ctx.perpetual_shutdown = False - - -@api_app.listener('before_server_stop') -async def _set_perpetual_shutdown(app): - app.ctx.perpetual_shutdown = True - - -async def _dispatch_perpetual(event_: str): - """Dispatch a perpetual-signal event. Isolated so tests can stub it.""" - await api_app.dispatch(event_) - - -@api_app.after_server_start -async def start_perpetual_tasks(app: Sanic): - events = _perpetual_events_for_this_process(app) - logger.info(f'start_perpetual_tasks started pid={os.getpid()} events={len(events)}') - logger.debug(f'start_perpetual_tasks: {events}') - - try: - for event_ in events: - logger.debug(f'start_perpetual_tasks {event_}') - await app.dispatch(event_) - except Exception as e: - logger.error('Failed to start perpetual tasks', exc_info=e) - raise - - logger.debug('start_perpetual_tasks completed') - - -async def _run_perpetual_iteration(func: callable, event_: str, sleep: int | float): - """Run one perpetual-worker iteration and reschedule unless shutting down. - - An ordinary Exception is logged and the loop continues. CancelledError is - terminal only during shutdown; any other cancellation is logged and the - worker is rescheduled. Previously any CancelledError skipped reschedule - with no log, silently killing file processing for the life of the process. - """ - logger.trace(f'perpetual_signal {event_}') - start = time() - try: - await func() - except CancelledError: - if _app_is_stopping(): - logger.info(f'Perpetual worker {event_} cancelled during shutdown') - raise - logger.warning( - f'Perpetual worker {event_} cancelled unexpectedly; will reschedule', - exc_info=True, - ) - _uncancel_current_task() - except Exception as e: - logger.error(f'Perpetual worker {event_} had error', exc_info=e) - finally: - if __debug__ and logger.isEnabledFor(TRACE_LEVEL): - elapsed = int(time() - start) - logger.trace(f'perpetual_signal {event_} took {elapsed} seconds') - - if PYTEST: - return - if _app_is_stopping(): - logger.info(f'Perpetual worker {event_} not rescheduling because the app is stopping') - return - - try: - await asyncio.sleep(sleep) - except CancelledError: - if _app_is_stopping(): - logger.info(f'Perpetual worker {event_} stopped during shutdown') - raise - logger.warning( - f'Perpetual worker {event_} cancelled while sleeping; rescheduling immediately' - ) - _uncancel_current_task() - - try: - await _dispatch_perpetual(event_) - except CancelledError: - logger.info(f'Perpetual worker {event_} cancelled while dispatching next run') - raise - - -def perpetual_signal(event: str = None, sleep: int | float = 1, run_while_testing: bool = False): - """Use Sanic signals to continually call the wrapped function. The wrapped function will continually be called, - even if it has errors. If the function is long-running, it will only be called again after it has finished.""" + The function is returned unchanged so tests can call it directly. Under pytest nothing is registered unless + `run_while_testing` is set.""" def wrapper(func: callable): if PYTEST and not run_while_testing: - # Do not run perpetual signal worker while testing, unless explicitly required. return func - - # Create a Sanic "signal" for the provided function. - event_ = event or f'wrolpi.perpetual.{func.__name__}' - - # Wrap the function in a worker that will call it perpetually. - @api_app.signal(event_) - async def worker(*args, **kwargs): - await _run_perpetual_iteration(lambda: func(*args, **kwargs), event_, sleep) - - # Add this new signal to the global list so that a task will be started after server startup. - PERPETUAL_WORKERS.append(event_) + PERPETUAL_LOOPS.append(PerpetualLoop(func.__name__, func, sleep)) return func return wrapper -async def perpetual_owner_watchdog(): - """Keep the owner heartbeat fresh, or reclaim the loops if the owner is gone. +def per_worker_task(sleep: int | float = 1): + """Register a loop that EVERY Sanic server worker runs for itself (e.g. syncing its own log level). Anything + that must happen exactly once belongs in `perpetual_signal` instead.""" - 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_) + def wrapper(func: callable): + if PYTEST: + return func + PER_WORKER_TASKS.append(PerpetualLoop(func.__name__, func, sleep)) + return func + + return wrapper -# 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) +@api_app.after_server_start +async def start_per_worker_tasks(app: Sanic): + stop = asyncio.Event() + app.ctx.per_worker_stop = stop + app.ctx.per_worker_tasks = [asyncio.create_task(run_loop(loop, stop), name=loop.name) for loop in PER_WORKER_TASKS] + logger.info(f'start_per_worker_tasks pid={os.getpid()} tasks={[i.name for i in PER_WORKER_TASKS]}') + + +@api_app.listener('before_server_stop') +async def stop_per_worker_tasks(app: Sanic): + stop = getattr(app.ctx, 'per_worker_stop', None) + if stop is not None: + stop.set() + tasks = getattr(app.ctx, 'per_worker_tasks', None) or [] + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) diff --git a/wrolpi/contexts.py b/wrolpi/contexts.py index 5aece80a..24e0e050 100644 --- a/wrolpi/contexts.py +++ b/wrolpi/contexts.py @@ -59,13 +59,6 @@ def attach_shared_contexts(app: Sanic): app.shared_ctx.single_tasks_started = multiprocessing.Event() app.shared_ctx.flags_initialized = multiprocessing.Event() - app.shared_ctx.perpetual_tasks_started = multiprocessing.Event() - # PID of the worker that started the single-process perpetual loops. - # 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() # Transcode # Transcoding saturates every core; only one ffmpeg transcode may run machine-wide, @@ -106,7 +99,7 @@ def attach_shared_contexts(app: Sanic): app.shared_ctx.file_worker_jobs = manager.dict() # Jobs (wrolpi/jobs.py): FIFO queue of registered functions run one at a time by the - # perpetual-tasks owner. Never persisted; an API restart loses them. + # perpetual process. Never persisted; an API restart loses them. app.shared_ctx.jobs = manager.dict() app.shared_ctx.jobs_queue = manager.Queue() app.shared_ctx.jobs_lock = multiprocessing.Lock() @@ -233,11 +226,6 @@ 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() - 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/downloader.py b/wrolpi/downloader.py index b6ffef01..a0fae8de 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -1122,10 +1122,10 @@ class DownloadManager: # * `_add_processing_domain` — it mutates shared, non-transactional state, so adding it # before commit would leak the domain forever if the transaction fails (dispatch is then # skipped and no signal handler ever runs to release it); - # * `api_app.dispatch` — dispatching inside the transaction deadlocks, because the awaited - # `signal_download_download` opens its own write session (also `BEGIN IMMEDIATE`, - # inheriting `_immediate_txn`) and blocks on the write lock this transaction still holds, - # timing out after `busy_timeout` with "database is locked" so no download ever starts. + # * starting `signal_download_download` — starting it inside the transaction deadlocks, because it + # opens its own write session (also `BEGIN IMMEDIATE`, inheriting `_immediate_txn`) and blocks on + # the write lock this transaction still holds, timing out after `busy_timeout` with "database is + # locked" so no download ever starts. # `claimed` tracks the domains taken this cycle so the loop's own dedup / concurrency cap # still holds even though `processing_domains` is not mutated until after commit. to_dispatch = [] @@ -1237,12 +1237,13 @@ class DownloadManager: pass # The immediate (write) transaction has now committed and released the write lock. Claim the - # shared processing_domain and dispatch each signal: doing both here (not inside the txn) - # means a failed transaction leaks nothing, and the handler's write session cannot deadlock - # against a lock we no longer hold. + # shared processing_domain and start each download as its own task: doing both here (not inside + # the txn) means a failed transaction leaks nothing, and the download's write session cannot + # deadlock against a lock we no longer hold. (Plain tasks, not Sanic signals: the perpetual + # process has no Sanic server, so it has no finalized signal router.) for context in to_dispatch: self._add_processing_domain(context.pop('domain')) - await api_app.dispatch('wrolpi.download.download', context=context) + background_task(signal_download_download(**context)) async def do_downloads(self): """Schedule any downloads that are new. @@ -1822,9 +1823,10 @@ class DownloadManager: download_manager = DownloadManager() -@api_app.signal('wrolpi.download.download') async def signal_download_download(download_id: int, download_url: str): - """Calls Downloaders based on the download information provided, as well as what is in the DB.""" + """Runs one Download: calls its Downloader based on the download information provided and what is in the DB. + + Started as a background task by `DownloadManager._dispatch_new_downloads` in the perpetual process.""" from wrolpi.db import get_db_session with timer('signal_download_download', 'trace'): diff --git a/wrolpi/jobs.py b/wrolpi/jobs.py index a2508240..bc39c6e7 100644 --- a/wrolpi/jobs.py +++ b/wrolpi/jobs.py @@ -1,7 +1,7 @@ """In-memory FIFO queue of background Jobs. A Job is a call of a *registered* function with JSON-able kwargs. Any Sanic worker may enqueue a -Job; the perpetual-tasks owner process runs them one at a time, in order. Python logging emitted +Job; the perpetual process runs them one at a time, in order. Python logging emitted while a Job runs, and the output of subprocesses started through `JobContext.run_command`, are captured into the Job's log. A caller may wait for a Job, poll it, or cancel it. @@ -406,7 +406,7 @@ async def process_job_queue() -> int: @perpetual_signal(sleep=0.5) async def job_worker(): - """The single consumer of the Job queue; runs in the perpetual-tasks owner process only.""" + """The single consumer of the Job queue; runs in the perpetual process only.""" try: job_id = _queue().get_nowait() except queue.Empty: diff --git a/wrolpi/perpetual.py b/wrolpi/perpetual.py new file mode 100644 index 00000000..fa5760c4 --- /dev/null +++ b/wrolpi/perpetual.py @@ -0,0 +1,192 @@ +"""WROLPi's background loops. + +Singleton loops (file worker, downloads, switches, jobs, bulk tag, flag checks, update check) +run in ONE process that Sanic's Worker Manager owns: `run_perpetual_process` is handed to +`app.manager.manage()` from `main_process_ready`. The manager guarantees a single instance and +sends it SIGTERM on shutdown; a Sanic server worker never runs these loops, so there is nothing +for a worker to claim, hand off, or take over. + +Per-worker loops (things every Sanic server process must do for itself, like syncing its log +level) run as asyncio tasks inside each server worker. + +Both kinds are registered at import time with the decorators in `wrolpi.api_utils` +(`perpetual_signal` and `per_worker_task`) and are no-ops under pytest; tests call the loop +bodies directly. +""" +import asyncio +import os +import signal +import time +from asyncio import CancelledError +from dataclasses import dataclass +from typing import Callable, Awaitable, Iterable + +from wrolpi.common import logger + +logger = logger.getChild(__name__) + +PERPETUAL_PROCESS_NAME = 'Perpetual' +# Sanic names a managed process "-".join(["Sanic", name, index]) (sanic/worker/process.py). +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 +# How long the runner waits for cancelled loops before abandoning them at shutdown. +PERPETUAL_CANCEL_TIMEOUT = 15 + + +@dataclass +class PerpetualLoop: + name: str + func: Callable[[], Awaitable] + sleep: float + + +# Filled at import time by the decorators in wrolpi.api_utils. +PERPETUAL_LOOPS: list[PerpetualLoop] = [] +PER_WORKER_TASKS: list[PerpetualLoop] = [] + + +async def run_loop(loop: PerpetualLoop, stop: asyncio.Event): + """Call `loop.func` forever, `loop.sleep` seconds after each call returns, until `stop` is set. + + An ordinary Exception is logged and the loop continues. A long call is never re-entered: the next call + starts only after the previous one returns. Cancellation ends the loop; the runner is the only canceller. + """ + while not stop.is_set(): + start = time.time() + try: + await loop.func() + except CancelledError: + raise + except Exception as e: + logger.error(f'Perpetual loop {loop.name} had error', exc_info=e) + finally: + logger.trace(f'Perpetual loop {loop.name} took {time.time() - start:.3f} seconds') + + try: + await asyncio.wait_for(stop.wait(), timeout=loop.sleep) + except asyncio.TimeoutError: + pass + + +async def run_loops(loops: Iterable[PerpetualLoop], stop: asyncio.Event, cancel_timeout: float = PERPETUAL_CANCEL_TIMEOUT): + """Run every loop as its own task until `stop` is set, then cancel them all and wait (bounded) for them to end.""" + tasks = [asyncio.create_task(run_loop(loop, stop), name=loop.name) for loop in loops] + if not tasks: + await stop.wait() + return + + await stop.wait() + for task in tasks: + task.cancel() + done, pending = await asyncio.wait(tasks, timeout=cancel_timeout) + for task in pending: + logger.warning(f'Perpetual loop {task.get_name()} did not stop within {cancel_timeout}s; abandoning it') + for task in done: + if not task.cancelled() and task.exception(): + logger.error(f'Perpetual loop {task.get_name()} ended with error', exc_info=task.exception()) + + +def install_stop_signal_handlers(loop: asyncio.AbstractEventLoop, stop: asyncio.Event): + """SIGTERM (the manager's shutdown) and SIGINT set `stop`; a second signal is ignored rather than killing us + mid-cleanup.""" + for sig in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(sig, stop.set) + + +async def _shutdown_singletons(): + """Stop the singleton services that live in this process before its loops are cancelled.""" + from wrolpi.common import cancel_refresh_tasks, cancel_background_tasks + from wrolpi.downloader import download_manager + + try: + download_manager.stop() + except Exception as e: + logger.error('download_manager.stop failed during perpetual shutdown', exc_info=e) + try: + await cancel_refresh_tasks() + except Exception as e: + logger.error('cancel_refresh_tasks failed during perpetual shutdown', exc_info=e) + try: + await cancel_background_tasks() + except Exception as e: + logger.error('cancel_background_tasks failed during perpetual shutdown', exc_info=e) + + +def _initialize_process(): + """What Sanic's `after_server_start` listeners do for a server worker, done here for this process: each + process owns its FileConfig objects (they share `shared_ctx.*config`).""" + from wrolpi.api_utils import api_app + from wrolpi.contexts import initialize_configs_contexts + + initialize_configs_contexts(api_app) + + +async def serve_perpetual_loops(loops: Iterable[PerpetualLoop], stop: asyncio.Event | None = None, + initialize: Callable[[], None] = _initialize_process): + """The perpetual process's main coroutine.""" + stop = stop or asyncio.Event() + install_stop_signal_handlers(asyncio.get_running_loop(), stop) + initialize() + loops = list(loops) + logger.info(f'Perpetual process started pid={os.getpid()} loops={[i.name for i in loops]}') + runner = asyncio.create_task(run_loops(loops, stop), name='perpetual-runner') + await stop.wait() + logger.warning('Perpetual process stopping') + await _shutdown_singletons() + await runner + logger.info('Perpetual process stopped') + + +def run_perpetual_process(): + """Entry point handed to `app.manager.manage()`. Runs in its own (forked) process: it inherits `api_app`, + 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. + signal.signal(signal.SIGTERM, signal.SIG_DFL) + signal.signal(signal.SIGINT, signal.SIG_DFL) + asyncio.run(serve_perpetual_loops(PERPETUAL_LOOPS)) + + +def perpetual_process_needs_restart(workers: dict, now: float, last_request: float) -> bool: + """True if the manager reports the perpetual process as gone and no restart was requested recently. + + `workers` is `app.m.workers` (a dict keyed by process name). A missing entry is not "dead": the manager + has not registered the process yet. + """ + entry = workers.get(PERPETUAL_PROCESS_IDENT) + if not entry or entry.get('state') not in PERPETUAL_PROCESS_DEAD_STATES: + return False + return (now - last_request) >= PERPETUAL_RESTART_COOLDOWN + + +_last_restart_request = 0.0 + + +async def perpetual_process_health_check(): + """Runs in every server worker: ask the manager to restart the perpetual process if it has died. + + Sanic only marks a dead managed process FAILED; it does not restart it. This is a restart request to the + manager, not a takeover: the loops still run in exactly one process. + """ + global _last_restart_request + from wrolpi.api_utils import api_app + + multiplexer = getattr(api_app, 'multiplexer', None) + if multiplexer is None: + return + try: + workers = multiplexer.workers + except Exception as e: + logger.debug('Could not read worker states', exc_info=e) + return + now = time.time() + if not perpetual_process_needs_restart(workers, now, _last_restart_request): + return + _last_restart_request = now + state = workers[PERPETUAL_PROCESS_IDENT].get('state') + logger.warning(f'Perpetual process is {state}; requesting restart pid={os.getpid()}') + multiplexer.restart(PERPETUAL_PROCESS_IDENT) diff --git a/wrolpi/root_api.py b/wrolpi/root_api.py index 5bcab2b9..01d459c2 100644 --- a/wrolpi/root_api.py +++ b/wrolpi/root_api.py @@ -572,7 +572,7 @@ async def get_status(request: Request): sanic_workers = { i: {'pid': j['pid'], 'state': j['state']} for i, j in request.app.multiplexer.workers.items() - if i.startswith('Sanic-Server') + if i.startswith(('Sanic-Server', 'Sanic-Perpetual')) } wrolpi_config = get_wrolpi_config() @@ -732,17 +732,11 @@ async def get_upgrade_check(request: Request): Query params: force: If 'true', force a fresh git fetch before checking. """ - from wrolpi.upgrade import check_for_update + from wrolpi.upgrade import refresh_update_status force = request.args.get('force', 'false').lower() == 'true' - result = check_for_update(fetch=force) - - # Update shared_ctx.status so /api/status returns the latest info. - # Using shared_ctx.status (a manager.dict) ensures info is shared across all workers. - api_app.shared_ctx.status['update_available'] = result.get('update_available', False) - api_app.shared_ctx.status['latest_commit'] = result.get('latest_commit') - api_app.shared_ctx.status['current_commit'] = result.get('current_commit') - api_app.shared_ctx.status['commits_behind'] = result.get('commits_behind', 0) + # Runs git off the event loop and updates shared_ctx.status so /api/status returns the latest info. + result = await refresh_update_status(api_app.shared_ctx.status, fetch=force) return json_response(result) diff --git a/wrolpi/test/test_api_utils.py b/wrolpi/test/test_api_utils.py index 43847207..03452d2d 100644 --- a/wrolpi/test/test_api_utils.py +++ b/wrolpi/test/test_api_utils.py @@ -1,222 +1,71 @@ -"""Tests for perpetual_signal cancellation / reschedule behavior.""" +"""Tests for the perpetual / per-worker loop registrars and the per-worker task listeners.""" import asyncio from unittest import mock import pytest -from wrolpi.api_utils import ( - _app_is_stopping, - _run_perpetual_iteration, - api_app, -) +from wrolpi import perpetual +from wrolpi.api_utils import perpetual_signal, per_worker_task, start_per_worker_tasks, stop_per_worker_tasks +from wrolpi.perpetual import PerpetualLoop -def test_app_is_stopping_false_by_default(): - api_app.ctx.perpetual_shutdown = False - api_app.state.is_stopping = False - # is_started/is_running are not a shutdown signal: docker Sanic workers - # often have is_started=True and is_running=False while serving. - api_app.state.is_started = True - api_app.state.is_running = False - assert _app_is_stopping() is False +def test_registrars_are_noops_under_pytest(): + async def f(): + pass + + assert perpetual_signal(sleep=3)(f) is f + assert per_worker_task(sleep=3)(f) is f + assert all(i.func is not f for i in perpetual.PERPETUAL_LOOPS) + assert all(i.func is not f for i in perpetual.PER_WORKER_TASKS) -def test_app_is_stopping_when_shutdown_flag_set(): - api_app.ctx.perpetual_shutdown = True - try: - assert _app_is_stopping() is True - finally: - api_app.ctx.perpetual_shutdown = False - - -def test_app_is_stopping_when_sanic_is_stopping(): - api_app.ctx.perpetual_shutdown = False - api_app.state.is_stopping = True - try: - assert _app_is_stopping() is True - finally: - api_app.state.is_stopping = False - - -@pytest.mark.asyncio -async def test_perpetual_iteration_reschedules_after_unexpected_cancel(monkeypatch): - """CancelledError must not silently kill the perpetual loop. - - Regression: perpetual_signal caught CancelledError, set cancelled=True, - and skipped re-dispatch with no log. The file-worker pump died for the - life of the process. - """ - dispatched = [] - - async def boom(): - raise asyncio.CancelledError() - - async def fake_dispatch(event): - dispatched.append(event) - +def test_registrars_record_one_loop_each(monkeypatch): + """Outside pytest, each decorator registers exactly one loop in its own registry and returns the function.""" + singletons, everywhere_ = [], [] monkeypatch.setattr('wrolpi.api_utils.PYTEST', False) - monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: False) - monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch) - monkeypatch.setattr('wrolpi.api_utils.asyncio.sleep', mock.AsyncMock()) + monkeypatch.setattr('wrolpi.api_utils.PERPETUAL_LOOPS', singletons) + monkeypatch.setattr('wrolpi.api_utils.PER_WORKER_TASKS', everywhere_) - await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0) - assert dispatched == ['wrolpi.perpetual.test_worker'] + async def singleton(): + pass + + async def everywhere(): + pass + + assert perpetual_signal(sleep=7)(singleton) is singleton + assert per_worker_task(sleep=2)(everywhere) is everywhere + + assert [(i.name, i.sleep) for i in singletons] == [('singleton', 7)] + assert [(i.name, i.sleep) for i in everywhere_] == [('everywhere', 2)] @pytest.mark.asyncio -async def test_perpetual_iteration_stops_on_shutdown_cancel(monkeypatch): - """CancelledError during shutdown must stay terminal (do not reschedule).""" - dispatched = [] +async def test_per_worker_tasks_start_and_stop_with_the_server(monkeypatch): + """after_server_start runs every per-worker loop as a task; before_server_stop ends them.""" + calls = [] - async def boom(): - raise asyncio.CancelledError() + async def tick(): + calls.append(1) - async def fake_dispatch(event): - dispatched.append(event) + monkeypatch.setattr('wrolpi.api_utils.PER_WORKER_TASKS', [PerpetualLoop('tick', tick, 0.01)]) + app = mock.Mock() + app.ctx = mock.Mock() - monkeypatch.setattr('wrolpi.api_utils.PYTEST', False) - monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: True) - monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch) + await start_per_worker_tasks(app) + await asyncio.sleep(0.05) + assert calls, 'the per-worker loop should be running' + assert len(app.ctx.per_worker_tasks) == 1 - with pytest.raises(asyncio.CancelledError): - await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0) - assert dispatched == [] - - -def test_claim_perpetual_tasks_when_never_started(async_client): - """The first worker to start must claim the perpetual loops.""" - from wrolpi.api_utils import _claim_perpetual_tasks, api_app - - api_app.shared_ctx.perpetual_tasks_started.clear() - assert _claim_perpetual_tasks(api_app) is True - - -def test_claim_perpetual_tasks_skips_when_owner_is_alive(async_client): - """A live owner keeps the single-process loops from starting twice.""" - import os - from wrolpi.api_utils import _claim_perpetual_tasks, api_app - - api_app.shared_ctx.perpetual_tasks_started.set() - api_app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid() - assert _claim_perpetual_tasks(api_app) is False - - -def test_claim_perpetual_tasks_when_owner_process_is_dead(async_client): - """A dead owner must not block a replacement worker from restarting the loops. - - ``perpetual_tasks_started`` is a shared Event. Sanic docker runs with - auto_reload, and a worker crash/recycle leaves that Event set. The - replacement process's after_server_start then no-ops, and file processing - (plus downloads, switches, …) is dead until a full API restart. - - This is the unconfirmed wedge from the 2026-08-18 incident: last file-worker - activity from one PID, then silence across every process, recovered only by - restarting the API. - """ - from wrolpi.api_utils import _claim_perpetual_tasks, api_app - - api_app.shared_ctx.perpetual_tasks_started.set() - # A PID that cannot be running on this host. - api_app.shared_ctx.perpetual_tasks_owner_pid.value = 2**31 - 1 - assert _claim_perpetual_tasks(api_app) is True - assert api_app.shared_ctx.perpetual_tasks_owner_pid.value != 2**31 - 1, \ - 'claimer must record this process as the new owner' - - -def test_claim_perpetual_tasks_when_heartbeat_is_stale(async_client): - """A live PID with a stale heartbeat is treated as a dead owner. - - 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 == [OWNER_WATCHDOG_EVENT] - assert FILE_WORKER_PERPETUAL_EVENT not in events + await stop_per_worker_tasks(app) + assert all(t.done() for t in app.ctx.per_worker_tasks) + count = len(calls) + await asyncio.sleep(0.03) + assert len(calls) == count, 'the loop must not run after stop' @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) - - # A sibling test's async_client teardown runs the real before_server_stop listener, - # which leaves perpetual_shutdown=True on the module-global app; the watchdog would - # then return before reclaiming. This test is about a dead owner, not shutdown. - monkeypatch.setattr(api_app.ctx, 'perpetual_shutdown', False, raising=False) - monkeypatch.setattr(api_app.state, 'is_stopping', False, raising=False) - - 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 -async def test_perpetual_iteration_reschedules_after_error(monkeypatch): - """Ordinary exceptions are logged and the loop continues.""" - dispatched = [] - - async def boom(): - raise RuntimeError('worker exploded') - - async def fake_dispatch(event): - dispatched.append(event) - - monkeypatch.setattr('wrolpi.api_utils.PYTEST', False) - monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: False) - monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch) - monkeypatch.setattr('wrolpi.api_utils.asyncio.sleep', mock.AsyncMock()) - - await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0) - assert dispatched == ['wrolpi.perpetual.test_worker'] +async def test_stop_per_worker_tasks_without_start(): + """Stopping an app that never started per-worker tasks (tests, or a startup failure) is harmless.""" + app = mock.Mock() + app.ctx = mock.Mock(spec=[]) # no per_worker_* attributes + await stop_per_worker_tasks(app) diff --git a/wrolpi/test/test_downloader.py b/wrolpi/test/test_downloader.py index d1196fb6..9abecc50 100644 --- a/wrolpi/test/test_downloader.py +++ b/wrolpi/test/test_downloader.py @@ -1,4 +1,5 @@ import asyncio +import contextlib import json import pathlib import sqlite3 @@ -1769,8 +1770,28 @@ def _make_processed_download(session, downloader_name, url, *, when=None, freque return download -def _dispatched_urls(dispatch_mock) -> set: - return {c.kwargs['context']['download_url'] for c in dispatch_mock.call_args_list} +@contextlib.contextmanager +def _capture_started_downloads(): + """Record the downloads `_dispatch_new_downloads` starts (as background tasks) without running them.""" + started = mock.Mock() + + async def _never_runs(): + pass + + def fake_signal_download_download(**kwargs): + started(**kwargs) + return _never_runs() + + def fake_background_task(coro): + coro.close() + + with mock.patch('wrolpi.downloader.signal_download_download', fake_signal_download_download), \ + mock.patch('wrolpi.downloader.background_task', fake_background_task): + yield started + + +def _dispatched_urls(started_mock) -> set: + return {c.kwargs['download_url'] for c in started_mock.call_args_list} @pytest.mark.asyncio @@ -1807,7 +1828,7 @@ async def test_daily_limit_per_domain_blocks(test_session, test_download_manager test_session.add_all([new_blocked, new_allowed]) test_session.commit() - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: + with _capture_started_downloads() as dispatch: await test_download_manager.dispatch_downloads() dispatched = _dispatched_urls(dispatch) @@ -1830,7 +1851,7 @@ async def test_daily_limit_per_domain_allows_under_limit(test_session, test_down test_session.add(new) test_session.commit() - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: + with _capture_started_downloads() as dispatch: await test_download_manager.dispatch_downloads() assert 'https://example.com/new' in _dispatched_urls(dispatch) @@ -1840,14 +1861,14 @@ async def test_daily_limit_per_domain_allows_under_limit(test_session, test_down @pytest.mark.asyncio async def test_download_dispatched_outside_immediate_transaction(test_session, test_download_manager, test_downloader): - """The download signal must be dispatched AFTER the claiming write transaction commits. + """The download task must be started AFTER the claiming write transaction commits. Regression test for a production deadlock (downloads silently stopped on a test device): - `_dispatch_new_downloads` claimed downloads inside its write transaction and `await`ed - `api_app.dispatch(...)` while that transaction was still open. Under NullPool (production) the - dispatched `signal_download_download` opens its OWN connection and also issues BEGIN IMMEDIATE, - which blocks on the write lock the outer transaction still holds until `busy_timeout` (30s) - expires: `database is locked`, and no download ever starts. + `_dispatch_new_downloads` claimed downloads inside its write transaction and started + `signal_download_download` while that transaction was still open. Under NullPool (production) + `signal_download_download` opens its OWN connection and also issues BEGIN IMMEDIATE, which blocks + on the write lock the outer transaction still holds until `busy_timeout` (30s) expires: + `database is locked`, and no download ever starts. `production_like_sessions` gives the dispatcher its own connection, so the write lock is real and a competing connection can probe for it at dispatch time. @@ -1860,18 +1881,17 @@ async def test_download_dispatched_outside_immediate_transaction(test_session, t write_lock_held_at_dispatch = [] - async def record(event, *args, **kwargs): - if event == 'wrolpi.download.download': - write_lock_held_at_dispatch.append(probe_write_lock_is_held(db_file)) + def record(coro): + write_lock_held_at_dispatch.append(probe_write_lock_is_held(db_file)) + coro.close() # never run the download itself with production_like_sessions(test_session): - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: - dispatch.side_effect = record + with mock.patch('wrolpi.downloader.background_task', record): await test_download_manager.dispatch_downloads() - assert write_lock_held_at_dispatch, 'expected the download to be dispatched' + assert write_lock_held_at_dispatch, 'expected the download to be started' assert not any(write_lock_held_at_dispatch), \ - 'download signal was dispatched while the claim still held the write lock (deadlocks under NullPool)' + 'download was started while the claim still held the write lock (deadlocks under NullPool)' @pytest.mark.asyncio @@ -1915,7 +1935,7 @@ async def test_dispatch_releases_claimed_domains_when_transaction_fails(test_ses # Fail the claim transaction's commit (e.g. an I/O error while the write lock is held). locked = OperationalError('COMMIT', {}, sqlite3.OperationalError('database is locked')) - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock), \ + with _capture_started_downloads(), \ mock.patch.object(test_session, 'commit', side_effect=locked): # A transient lock is swallowed by dispatch_downloads; it must not leave the claim leaked. await test_download_manager.dispatch_downloads() @@ -1964,7 +1984,7 @@ async def test_daily_limit_resets_next_day(test_session, test_download_manager, test_session.add(new) test_session.commit() - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: + with _capture_started_downloads() as dispatch: await test_download_manager.dispatch_downloads() assert 'https://example.com/new' in _dispatched_urls(dispatch) @@ -1983,7 +2003,7 @@ async def test_daily_limit_counts_failed_attempts(test_session, test_download_ma test_session.add(new) test_session.commit() - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: + with _capture_started_downloads() as dispatch: await test_download_manager.dispatch_downloads() assert 'https://example.com/new' not in _dispatched_urls(dispatch) @@ -2049,7 +2069,7 @@ async def test_daily_limit_global_blocks(test_session, test_download_manager, te test_session.add(new) test_session.commit() - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: + with _capture_started_downloads() as dispatch: await test_download_manager.dispatch_downloads() assert _dispatched_urls(dispatch) == set() @@ -2069,7 +2089,7 @@ async def test_daily_limit_global_allows_under_limit(test_session, test_download test_session.add(new) test_session.commit() - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: + with _capture_started_downloads() as dispatch: await test_download_manager.dispatch_downloads() assert 'https://wikipedia.org/new' in _dispatched_urls(dispatch) @@ -2092,7 +2112,7 @@ async def test_recurring_downloads_bypass_daily_limits(test_session, test_downlo test_session.add_all([recurring, non_recurring]) test_session.commit() - with mock.patch.object(type(api_app), 'dispatch', new_callable=mock.AsyncMock) as dispatch: + with _capture_started_downloads() as dispatch: await test_download_manager.dispatch_downloads() dispatched = _dispatched_urls(dispatch) diff --git a/wrolpi/test/test_perpetual.py b/wrolpi/test/test_perpetual.py new file mode 100644 index 00000000..7c63ff09 --- /dev/null +++ b/wrolpi/test/test_perpetual.py @@ -0,0 +1,156 @@ +"""Tests for the perpetual process runner and the per-worker health check (wrolpi/perpetual.py).""" +import asyncio +import os +import signal +import threading +import time +from unittest import mock + +import pytest + +from wrolpi.perpetual import ( + PERPETUAL_PROCESS_IDENT, + PERPETUAL_RESTART_COOLDOWN, + PerpetualLoop, + perpetual_process_needs_restart, + run_loop, + run_loops, + serve_perpetual_loops, +) + + +@pytest.mark.asyncio +async def test_run_loop_calls_sleeps_and_survives_errors(): + """The loop keeps calling its function after an exception, and stops promptly when told to.""" + calls = [] + stop = asyncio.Event() + + async def flaky(): + calls.append(time.time()) + if len(calls) == 2: + raise RuntimeError('second call fails') + if len(calls) == 4: + stop.set() + + await asyncio.wait_for(run_loop(PerpetualLoop('flaky', flaky, sleep=0.01), stop), timeout=2) + + assert len(calls) == 4, 'the loop must continue after an error and stop when asked' + + +@pytest.mark.asyncio +async def test_run_loop_does_not_reenter_a_long_call(): + """A slow iteration is awaited; the next call starts only after it returns.""" + active = 0 + overlaps = [] + stop = asyncio.Event() + calls = 0 + + async def slow(): + nonlocal active, calls + calls += 1 + active += 1 + overlaps.append(active) + await asyncio.sleep(0.02) + active -= 1 + if calls == 3: + stop.set() + + await asyncio.wait_for(run_loop(PerpetualLoop('slow', slow, sleep=0), stop), timeout=2) + assert max(overlaps) == 1 + + +@pytest.mark.asyncio +async def test_run_loops_cancels_everything_on_stop(): + """Setting `stop` cancels every loop task; a loop that ignores cancellation is abandoned, not waited for forever.""" + stop = asyncio.Event() + cancelled = [] + + async def sleeper(): + try: + await asyncio.sleep(3600) + except asyncio.CancelledError: + cancelled.append('sleeper') + raise + + async def stubborn(): + try: + await asyncio.sleep(3600) + except asyncio.CancelledError: + cancelled.append('stubborn') + await asyncio.sleep(3600) # refuses to die + + loops = [PerpetualLoop('sleeper', sleeper, 1), PerpetualLoop('stubborn', stubborn, 1)] + runner = asyncio.ensure_future(run_loops(loops, stop, cancel_timeout=0.1)) + await asyncio.sleep(0.01) + stop.set() + await asyncio.wait_for(runner, timeout=2) + + assert sorted(cancelled) == ['sleeper', 'stubborn'] + + +@pytest.mark.asyncio +async def test_serve_perpetual_loops_stops_on_sigterm(): + """The perpetual process exits when the manager sends SIGTERM.""" + ticks = [] + + async def tick(): + ticks.append(1) + + initialized = [] + with mock.patch('wrolpi.perpetual._shutdown_singletons', mock.AsyncMock()) as shutdown: + serving = asyncio.ensure_future( + serve_perpetual_loops([PerpetualLoop('tick', tick, 0.01)], initialize=lambda: initialized.append(1))) + await asyncio.sleep(0.05) + assert ticks, 'loop should be running before the signal' + # From another thread, as the manager (another process) would. + threading.Timer(0.01, lambda: os.kill(os.getpid(), signal.SIGTERM)).start() + await asyncio.wait_for(serving, timeout=2) + + shutdown.assert_awaited_once() + assert initialized == [1], 'per-process initialization must run once before the loops' + + +def test_perpetual_process_needs_restart(): + """Only a process the manager reports as gone triggers a restart request, at most once per cooldown.""" + now = 1_000_000.0 + dead = {PERPETUAL_PROCESS_IDENT: {'pid': 123, 'state': 'FAILED'}} + alive = {PERPETUAL_PROCESS_IDENT: {'pid': 123, 'state': 'STARTED'}} + + assert perpetual_process_needs_restart(dead, now, last_request=0) is True + assert perpetual_process_needs_restart(dead, now, last_request=now - 1) is False, 'within cooldown' + assert perpetual_process_needs_restart(dead, now, last_request=now - PERPETUAL_RESTART_COOLDOWN) is True + for state in ('STARTING', 'STARTED', 'RESTARTING', 'NONE'): + alive[PERPETUAL_PROCESS_IDENT]['state'] = state + assert perpetual_process_needs_restart(alive, now, last_request=0) is False, state + assert perpetual_process_needs_restart({}, now, last_request=0) is False, 'not registered yet is not dead' + assert perpetual_process_needs_restart({'Sanic-Server-0': {'state': 'FAILED'}}, now, 0) is False + + +@pytest.mark.asyncio +async def test_health_check_requests_restart_once(async_client, monkeypatch): + """Every server worker runs this; a dead perpetual process gets one restart request per cooldown.""" + from wrolpi import perpetual + from wrolpi.api_utils import api_app + + multiplexer = mock.Mock() + multiplexer.workers = {PERPETUAL_PROCESS_IDENT: {'pid': 1, 'state': 'FAILED'}} + monkeypatch.setattr(api_app, 'multiplexer', multiplexer, raising=False) + monkeypatch.setattr(perpetual, '_last_restart_request', 0.0) + + await perpetual.perpetual_process_health_check() + await perpetual.perpetual_process_health_check() + multiplexer.restart.assert_called_once_with(PERPETUAL_PROCESS_IDENT) + + multiplexer.workers = {PERPETUAL_PROCESS_IDENT: {'pid': 2, 'state': 'STARTED'}} + monkeypatch.setattr(perpetual, '_last_restart_request', 0.0) + await perpetual.perpetual_process_health_check() + multiplexer.restart.assert_called_once() + + +@pytest.mark.asyncio +async def test_health_check_without_multiplexer_is_a_noop(async_client, monkeypatch): + from wrolpi import perpetual + from wrolpi.api_utils import api_app + + monkeypatch.delattr(api_app, 'multiplexer', raising=False) + await perpetual.perpetual_process_health_check() diff --git a/wrolpi/test/test_root_api.py b/wrolpi/test/test_root_api.py index 73cfb9d4..62550906 100644 --- a/wrolpi/test/test_root_api.py +++ b/wrolpi/test/test_root_api.py @@ -421,6 +421,29 @@ async def test_get_status(async_client, test_session): assert isinstance(response.json.get('local_time'), str), 'local_time should be a str' +@pytest.mark.asyncio +async def test_status_reports_perpetual_process(async_client, monkeypatch): + """/api/status lists the perpetual process next to the server workers, and nothing else the manager tracks.""" + from unittest import mock + from wrolpi.api_utils import api_app + + multiplexer = mock.Mock() + multiplexer.workers = { + 'Sanic-Main': {'pid': 1}, + 'Sanic-Server-0': {'pid': 2, 'state': 'ACKED'}, + 'Sanic-Perpetual-0': {'pid': 3, 'state': 'STARTED'}, + 'Sanic-Inspector-0': {'pid': 4, 'state': 'STARTED'}, + } + monkeypatch.setattr(api_app, 'multiplexer', multiplexer, raising=False) + + request, response = await async_client.get('/api/status') + assert response.status_code == HTTPStatus.OK + assert response.json['sanic_workers'] == { + 'Sanic-Server-0': {'pid': 2, 'state': 'ACKED'}, + 'Sanic-Perpetual-0': {'pid': 3, 'state': 'STARTED'}, + } + + @pytest.mark.asyncio async def test_status_local_time_with_timezone(async_client, test_session, test_wrolpi_config): """Status endpoint includes local_time using configured timezone.""" diff --git a/wrolpi/test/test_upgrade.py b/wrolpi/test/test_upgrade.py index 21ef1ba7..2c52993c 100644 --- a/wrolpi/test/test_upgrade.py +++ b/wrolpi/test/test_upgrade.py @@ -13,6 +13,7 @@ from wrolpi.upgrade import ( get_remote_commit, get_commits_behind, git_fetch, + refresh_update_status, start_upgrade, ) @@ -357,3 +358,58 @@ async def test_status_endpoint_includes_git_branch(async_client): finally: # Clean up api_app.shared_ctx.status.pop('git_branch', None) + + +@pytest.mark.asyncio +async def test_refresh_update_status_does_not_block_event_loop(): + """`git fetch` must run off the event loop. + + A synchronous fetch (up to its 60 s timeout) stalls every request and the + perpetual-owner heartbeat on that worker; a stale heartbeat triggers a + takeover by another worker. + """ + import asyncio + import threading + import time + + seen_threads = [] + + def slow_check(fetch=True): + seen_threads.append(threading.current_thread()) + time.sleep(0.3) + return {'update_available': True, 'commits_behind': 3, 'branch': 'master'} + + ticks = 0 + + async def ticker(): + nonlocal ticks + while True: + ticks += 1 + await asyncio.sleep(0.01) + + status = {} + with patch('wrolpi.upgrade.check_for_update', slow_check): + ticker_task = asyncio.ensure_future(ticker()) + await refresh_update_status(status) + ticker_task.cancel() + + assert seen_threads and seen_threads[0] is not threading.main_thread(), \ + 'check_for_update ran on the event loop thread' + assert ticks >= 10, f'event loop was starved during check_for_update (only {ticks} ticks)' + assert status['update_available'] is True and status['commits_behind'] == 3 and status['git_branch'] == 'master' + + + +@pytest.mark.asyncio +async def test_refresh_update_status_can_skip_fetch(): + """The upgrade-check endpoint passes fetch=False unless the user forces a fetch.""" + seen = [] + + def fake_check(fetch=True): + seen.append(fetch) + return {'update_available': False, 'branch': 'master'} + + with patch('wrolpi.upgrade.check_for_update', fake_check): + await refresh_update_status({}, fetch=False) + await refresh_update_status({}) + assert seen == [False, True] diff --git a/wrolpi/upgrade.py b/wrolpi/upgrade.py index 81950c92..4d7f06c9 100644 --- a/wrolpi/upgrade.py +++ b/wrolpi/upgrade.py @@ -4,6 +4,7 @@ WROLPi upgrade system. This module provides functionality to check for available updates by comparing the local git HEAD with the remote origin branch, and to trigger the upgrade process. """ +import asyncio import pathlib import subprocess @@ -212,6 +213,21 @@ def check_for_update(fetch: bool = True) -> dict: return result +async def refresh_update_status(status, fetch: bool = True) -> dict: + """Run check_for_update (which shells out to git; `git fetch` may take up to 60 s) off the event loop + and record the result in `status` (the shared_ctx.status dict read by /api/status). + + Nothing that shells out may block the event loop of the process that runs it. + """ + result = await asyncio.to_thread(check_for_update, fetch=fetch) + status['update_available'] = result.get('update_available', False) + status['latest_commit'] = result.get('latest_commit') + status['current_commit'] = result.get('current_commit') + status['commits_behind'] = result.get('commits_behind', 0) + status['git_branch'] = result.get('branch') + return result + + async def start_upgrade(): """ Start the WROLPi upgrade process.