diff --git a/main.py b/main.py index 2b84c729..92cd3506 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 register_perpetual_process, 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. + register_perpetual_process(app) + @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() @@ -325,14 +333,20 @@ async def perpetual_file_worker_queue(): @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.""" + """Copies the global log level into this process's logger. Registered for every server worker AND the + perpetual process (which does most of the interesting logging).""" log_level = api_app.shared_ctx.log_level.value if log_level != logger.getEffectiveLevel(): logger.info(f'changing log level from {logger.getEffectiveLevel()} to {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 +430,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 +445,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/common.py b/wrolpi/common.py index acfe2999..c9695d15 100644 --- a/wrolpi/common.py +++ b/wrolpi/common.py @@ -2560,12 +2560,16 @@ def background_task(coro) -> Task: async def cancel_background_tasks(): - """Cancels any background async tasks, if any.""" - if BACKGROUND_TASKS: - logger_.warning(f'Canceling {len(BACKGROUND_TASKS)} background tasks') - for task in BACKGROUND_TASKS: + """Cancels any background async tasks, if any, and waits for them to finish. + + Never raises for a task's own outcome: a cancelled child would otherwise surface as CancelledError here (a + BaseException) and abort the shutdown that called us.""" + tasks = [task for task in BACKGROUND_TASKS if not task.done()] + if tasks: + logger_.warning(f'Canceling {len(tasks)} background tasks') + for task in tasks: task.cancel() - await asyncio.gather(*BACKGROUND_TASKS) + await asyncio.gather(*tasks, return_exceptions=True) async def await_background_tasks(timeout: int = 10): @@ -2669,11 +2673,13 @@ REFRESH_TASKS: List[Task] = [] async def cancel_refresh_tasks(): """Cancel all refresh tasks, if any.""" - if REFRESH_TASKS: - logger_.warning(f'Canceling {len(REFRESH_TASKS)} refreshes') - for task in REFRESH_TASKS: + tasks = [task for task in REFRESH_TASKS if not task.done()] + if tasks: + logger_.warning(f'Canceling {len(tasks)} refreshes') + for task in tasks: task.cancel() - await asyncio.gather(*REFRESH_TASKS) + # See cancel_background_tasks: the children's CancelledError must not abort the caller. + await asyncio.gather(*tasks, return_exceptions=True) def cancelable_wrapper(func: callable): diff --git a/wrolpi/contexts.py b/wrolpi/contexts.py index 5aece80a..d961cdf2 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, @@ -76,6 +69,10 @@ def attach_shared_contexts(app: Sanic): # Speed tests in flight across all workers (wrolpi.speedtest). Informational only, no lock. app.shared_ctx.speedtest_active = multiprocessing.Value(ctypes.c_int, 0) + # When any server worker last asked the manager to restart the perpetual process (wrolpi/perpetual.py). + # Shared so the N workers issue one request per cooldown, not N. + app.shared_ctx.perpetual_restart_requested_at = multiprocessing.Value(ctypes.c_double, 0.0) + # Switches app.shared_ctx.switches = manager.dict() app.shared_ctx.switches_lock = multiprocessing.Lock() @@ -106,7 +103,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 +230,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() - 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 + if hasattr(app.shared_ctx, 'perpetual_restart_requested_at'): + app.shared_ctx.perpetual_restart_requested_at.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..3591760c 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. @@ -211,6 +211,24 @@ def _claim_job(job_id: str) -> Optional[dict]: return record +def fail_orphaned_jobs() -> int: + """Mark RUNNING Jobs owned by another (dead) process as FAILED. Called by the perpetual process when it + starts: only it runs Jobs, so any RUNNING record not carrying its pid belongs to a predecessor that died + mid-Job. Returns how many were failed.""" + count = 0 + with _lock(): + for job_id, record in list(_jobs().items()): + if record['status'] == RUNNING and record.get('pid') != os.getpid(): + record = dict(record) + record.update(status=FAILED, finished_at=now().isoformat(), + error='The process running this Job exited before it finished') + _jobs()[job_id] = record + count += 1 + if count: + logger.warning(f'fail_orphaned_jobs: failed {count} Job(s) left running by a previous process') + return count + + def _cancel_requested(job_id: str) -> bool: record = _jobs().get(job_id) return bool(record and record.get('cancel_requested')) @@ -406,7 +424,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..d5cfbaeb --- /dev/null +++ b/wrolpi/perpetual.py @@ -0,0 +1,312 @@ +"""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 +signals it to stop (SIGINT on shutdown, SIGTERM when it is restarted); a Sanic server worker never +runs these loops, so there is nothing for a worker to claim, hand off, or take over. + +If the process dies (an OOM kill on a small Pi is the realistic case), the manager only marks it +FAILED. Every server worker runs `perpetual_process_health_check`, which asks the manager to +restart it; the new process reconciles shared state left behind by its predecessor +(`reconcile_after_predecessor`) before starting its loops. + +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 +# Shared multiprocessing locks whose critical sections are short. A predecessor SIGKILLed inside one leaves it +# held forever and wedges every process that touches it; a live holder releases within this many seconds. +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 + + +@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): + """SIGINT (the manager's shutdown) and SIGTERM (the manager restarting us) 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 heal_shared_locks(app, timeout: float = SHARED_LOCK_HEAL_TIMEOUT) -> list[str]: + """Release any shared lock still held after `timeout` seconds; returns the names released. + + Only this process's predecessor can have held one for that long (the sections are short), and it is dead. + `multiprocessing.Lock` is a semaphore, so any process may release it.""" + healed = [] + for name in SHARED_LOCK_NAMES: + lock = getattr(app.shared_ctx, name, None) + if lock is None: + continue + if lock.acquire(timeout=timeout): + lock.release() + continue + logger.warning(f'Shared lock {name} was held by a dead process; releasing it') + try: + lock.release() + except ValueError: + # Released by its holder between our two calls. + pass + healed.append(name) + return healed + + +def reconcile_after_predecessor(app) -> dict: + """Undo what a perpetual process that died mid-work left in shared state and the DB. + + Shared state outlives the process: a domain still in `processing_domains` would never download again, a + `pending` Download or a RUNNING Job would stay that way forever, and the file-worker status would show a + 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.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['locks'] = heal_shared_locks(app) + + try: + from wrolpi.jobs import fail_orphaned_jobs + report['jobs'] = fail_orphaned_jobs() + except Exception as e: + logger.error('reconcile: failed to fail orphaned Jobs', exc_info=e) + + try: + domains = list(download_manager.processing_domains) if 'processing_domains' in \ + app.shared_ctx.download_manager_data else [] + if domains: + logger.warning(f'reconcile: clearing processing_domains left by a dead process: {domains}') + download_manager.processing_domains = [] + report['processing_domains'] = len(domains) + except Exception as e: + logger.error('reconcile: failed to clear processing_domains', exc_info=e) + + try: + with get_db_session(commit=True) as session: + report['downloads'] = session.query(Download) \ + .filter(Download.status == DownloadStatus.pending) \ + .update(dict(status=DownloadStatus.new), synchronize_session=False) + if report['downloads']: + logger.warning(f'reconcile: renewed {report["downloads"]} pending Download(s) left by a dead process') + except Exception as e: + # No database yet (media not mounted) is normal at boot; perpetual_check_db_is_up_worker handles it. + logger.debug('reconcile: could not renew pending Downloads', exc_info=e) + + try: + for job_id, record in list(file_worker._jobs.items()): + status = record['status'] if isinstance(record, dict) else record + if status == 'running': + file_worker._set_job_status(job_id, 'failed', 'The file worker process exited before it finished') + 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() + except Exception as e: + logger.error('reconcile: failed to reset the file worker status', exc_info=e) + + if any(report.values()): + logger.warning(f'reconcile_after_predecessor: {report}') + return report + + +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`). Then clean up after a predecessor.""" + from wrolpi.api_utils import api_app + from wrolpi.contexts import initialize_configs_contexts + + initialize_configs_contexts(api_app) + reconcile_after_predecessor(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) + if not PERPETUAL_LOOPS: + # Only `fork` carries the import-time registrations into this process; under `spawn` nothing would run. + raise RuntimeError('No perpetual loops are registered in this process; is Sanic.start_method "fork"?') + asyncio.run(serve_perpetual_loops(PERPETUAL_LOOPS)) + + +def register_perpetual_process(app) -> None: + """Called from `main_process_ready` after `shared_ctx` is attached. `transient=True` is load-bearing: it makes + the manager restart the process on auto-reload and marks it restartable, which the health check's restart + request requires.""" + app.manager.manage(PERPETUAL_PROCESS_NAME, run_perpetual_process, {}, transient=True) + + +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 + + +def _claim_restart_request(app, workers: dict, now: float) -> bool: + """Atomically decide, across all server workers, that this one issues the restart request. + + Two workers can read FAILED within the same monitor tick; a second `restart` would SIGTERM the process the + first one just started. The timestamp lives in `shared_ctx` under the Value's own lock.""" + requested_at = getattr(app.shared_ctx, 'perpetual_restart_requested_at', None) + if requested_at is None: + # Not attached (tests); fall back to a per-process cooldown. + global _last_restart_request + if not perpetual_process_needs_restart(workers, now, _last_restart_request): + return False + _last_restart_request = now + return True + with requested_at.get_lock(): + if not perpetual_process_needs_restart(workers, now, requested_at.value): + return False + requested_at.value = now + return True + + +_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. + """ + 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 + if not _claim_restart_request(api_app, workers, time.time()): + return + 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..ac093a86 100644 --- a/wrolpi/test/test_api_utils.py +++ b/wrolpi/test/test_api_utils.py @@ -1,222 +1,79 @@ -"""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 api_app, 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_per_worker_listeners_are_registered_on_the_app(): + """A typo in the listener event name would leave per-worker tasks never started or never stopped.""" + registered = {(i.event, i.listener) for i in api_app._future_listeners} + assert ('after_server_start', start_per_worker_tasks) in registered + assert ('before_server_stop', stop_per_worker_tasks) in registered -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_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_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_common.py b/wrolpi/test/test_common.py index 23f75197..48aab779 100644 --- a/wrolpi/test/test_common.py +++ b/wrolpi/test/test_common.py @@ -1423,3 +1423,19 @@ async def test_verify_gpg_signature_tampered(tmp_path, gpg_test_key): with mock.patch('wrolpi.common.GPG_PUBLIC_KEY', test_pubkey): assert await verify_gpg_signature(data_file, sig_file) is False + + +@pytest.mark.asyncio +async def test_cancel_background_tasks_does_not_raise_for_cancelled_children(): + """Cancelling children raises CancelledError (a BaseException) from `gather` unless exceptions are collected; + that aborted the shutdown of the process calling this.""" + from wrolpi.common import background_task, cancel_background_tasks, BACKGROUND_TASKS + + tasks = [background_task(asyncio.sleep(3600)) for _ in range(3)] + await asyncio.sleep(0) + assert all(t in BACKGROUND_TASKS for t in tasks) + + await cancel_background_tasks() + + assert all(t.cancelled() for t in tasks) + assert not any(t in BACKGROUND_TASKS for t in tasks) 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..423e9ce9 --- /dev/null +++ b/wrolpi/test/test_perpetual.py @@ -0,0 +1,335 @@ +"""Tests for the perpetual process runner and the per-worker health check (wrolpi/perpetual.py).""" +import asyncio +import multiprocessing +import os +import pathlib +import select +import signal +import sys +import threading +import time +from unittest import mock + +import pytest +from sanic.worker.constants import ProcessState +from sanic.worker.process import Worker + +from wrolpi import perpetual +from wrolpi.perpetual import ( + PERPETUAL_PROCESS_DEAD_STATES, + PERPETUAL_PROCESS_IDENT, + PERPETUAL_PROCESS_NAME, + PERPETUAL_RESTART_COOLDOWN, + PerpetualLoop, + perpetual_process_needs_restart, + reconcile_after_predecessor, + register_perpetual_process, + run_loop, + run_loops, + run_perpetual_process, + 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. + + The dead/alive partition is checked against Sanic's own ProcessState enum, so a rename fails here.""" + now = 1_000_000.0 + names = {state.name for state in ProcessState} + assert PERPETUAL_PROCESS_DEAD_STATES <= names, 'a dead state is not a Sanic ProcessState' + for state in ProcessState: + workers = {PERPETUAL_PROCESS_IDENT: {'pid': 123, 'state': state.name}} + expected = state.name in PERPETUAL_PROCESS_DEAD_STATES + assert perpetual_process_needs_restart(workers, now, last_request=0) is expected, state.name + + dead = {PERPETUAL_PROCESS_IDENT: {'pid': 123, 'state': ProcessState.FAILED.name}} + 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 + 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-0': {'state': 'FAILED'}}, now, 0) is False + + +def test_process_ident_matches_sanic_naming(): + """`PERPETUAL_PROCESS_IDENT` is what Sanic's Worker actually names the managed process.""" + worker = Worker(PERPETUAL_PROCESS_NAME, PERPETUAL_PROCESS_NAME, lambda: None, {}, multiprocessing.get_context(), + {}, 1) + assert [p.name for p in worker.processes] == [PERPETUAL_PROCESS_IDENT] + + +def test_register_perpetual_process_is_transient(): + """`transient=True` is load-bearing: it restarts the process on auto-reload and makes it restartable.""" + app = mock.Mock() + register_perpetual_process(app) + app.manager.manage.assert_called_once_with(PERPETUAL_PROCESS_NAME, run_perpetual_process, {}, transient=True) + + +@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) + api_app.shared_ctx.perpetual_restart_requested_at.value = 0.0 + + await perpetual.perpetual_process_health_check() + # A second worker (different process, so a different module-level state) sees the same FAILED tick. + monkeypatch.setattr(perpetual, '_last_restart_request', 0.0) + await perpetual.perpetual_process_health_check() + multiplexer.restart.assert_called_once_with(PERPETUAL_PROCESS_IDENT) + assert api_app.shared_ctx.perpetual_restart_requested_at.value > 0, 'the claim is recorded in shared_ctx' + + multiplexer.workers = {PERPETUAL_PROCESS_IDENT: {'pid': 2, 'state': 'STARTED'}} + api_app.shared_ctx.perpetual_restart_requested_at.value = 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() + + +def _wait_for_byte(fd: int, timeout: float) -> bool: + ready, _, _ = select.select([fd], [], [], timeout) + return bool(ready) and bool(os.read(fd, 1)) + + +@pytest.mark.parametrize('sig', [signal.SIGINT, signal.SIGTERM], ids=['SIGINT (manager shutdown)', 'SIGTERM (restart)']) +def test_run_perpetual_process_exits_cleanly_on_signal(async_client, monkeypatch, sig): + """The real entry point, in a real forked process, with a real background task alive, exits 0 on the signals + the manager sends. (This is a sync test so the fork does not inherit a running event loop.)""" + ready_r, ready_w = os.pipe() + + async def loop_with_live_background_task(): + from wrolpi.common import background_task + background_task(asyncio.sleep(3600)) + os.write(ready_w, b'x') + + monkeypatch.setattr(perpetual, 'PERPETUAL_LOOPS', [PerpetualLoop('bg', loop_with_live_background_task, 3600)]) + # 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' + os.kill(process.pid, sig) + process.join(15) + assert process.exitcode == 0, f'perpetual process exited {process.exitcode}' + finally: + if process.is_alive(): + process.kill() + process.join(5) + os.close(ready_r) + os.close(ready_w) + + +def test_run_perpetual_process_refuses_to_run_no_loops(monkeypatch): + """Under `spawn` nothing would be registered; that must fail loudly instead of running zero loops.""" + monkeypatch.setattr(perpetual, 'PERPETUAL_LOOPS', []) + with pytest.raises(RuntimeError, match='fork'): + run_perpetual_process() + + +def _crash_once_then_sleep(marker: str): + if not pathlib.Path(marker).exists(): + pathlib.Path(marker).touch() + sys.exit(1) + while True: + time.sleep(1) + + +def test_sanic_manager_marks_failed_and_restart_revives(test_directory): + """Against Sanic's real WorkerManager: a crashed managed process becomes FAILED (which the health check + keys on) and `restart(process_names=[ident])` starts a replacement with a new pid.""" + from multiprocessing import Pipe, Manager + from sanic.worker.manager import WorkerManager + + saved = {s: signal.getsignal(s) for s in (signal.SIGINT, signal.SIGTERM)} + ctx = multiprocessing.get_context('fork') + state_manager = Manager() + worker_state = state_manager.dict() + manager = None + try: + manager = WorkerManager(1, _crash_once_then_sleep, {'marker': str(test_directory / 'server-never')}, + ctx, Pipe(), worker_state) + # The fake "server" must not crash: give it an existing marker. + (test_directory / 'server-never').touch() + manager.manage(PERPETUAL_PROCESS_NAME, _crash_once_then_sleep, {'marker': str(test_directory / 'marker')}, + transient=True) + manager.start() + + deadline = time.time() + 15 + while time.time() < deadline: + manager._sync_states() + if worker_state.get(PERPETUAL_PROCESS_IDENT, {}).get('state') == ProcessState.FAILED.name: + break + time.sleep(0.1) + else: + pytest.fail(f'never FAILED: {dict(worker_state)}') + dead_pid = worker_state[PERPETUAL_PROCESS_IDENT]['pid'] + + manager.restart(process_names=[PERPETUAL_PROCESS_IDENT]) + + deadline = time.time() + 15 + while time.time() < deadline: + manager._sync_states() + entry = worker_state.get(PERPETUAL_PROCESS_IDENT, {}) + if entry.get('pid') not in (None, dead_pid) and entry.get('state') not in PERPETUAL_PROCESS_DEAD_STATES: + break + time.sleep(0.1) + else: + pytest.fail(f'not revived: {dict(worker_state)}') + assert pathlib.Path(test_directory / 'marker').exists() + finally: + if manager is not None: + for process in list(manager.processes): + try: + process._current_process.kill() + except Exception: + pass + state_manager.shutdown() + for sig, handler in saved.items(): + signal.signal(sig, handler) + + +@pytest.mark.asyncio +async def test_reconcile_after_predecessor(async_client, test_session, monkeypatch): + """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 + + # 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. + download_manager.processing_domains = ['example.com'] + test_session.add(Download(url='https://example.com/a', downloader='video', status='pending')) + test_session.add(Download(url='https://example.com/b', downloader='video', status='new')) + test_session.commit() + api_app.shared_ctx.jobs['dead'] = dict(id='dead', name='x', status=jobs.RUNNING, pid=2 ** 31 - 1, kwargs={}, + created_at='', started_at='', finished_at=None, error=None) + api_app.shared_ctx.jobs['mine'] = dict(id='mine', name='x', status=jobs.RUNNING, pid=os.getpid(), kwargs={}, + created_at='', started_at='', finished_at=None, error=None) + file_worker._set_job_status('fj', 'running') + file_worker.update_status(status='refreshing') + assert api_app.shared_ctx.jobs_lock.acquire(timeout=1) + monkeypatch.setattr(perpetual, 'SHARED_LOCK_HEAL_TIMEOUT', 0.1) + + report = reconcile_after_predecessor(api_app) + + assert report['locks'] == ['jobs_lock'] + assert api_app.shared_ctx.jobs_lock.acquire(timeout=1), 'the orphaned lock is usable again' + api_app.shared_ctx.jobs_lock.release() + assert report['jobs'] == 1 + assert api_app.shared_ctx.jobs['dead']['status'] == jobs.FAILED + assert api_app.shared_ctx.jobs['mine']['status'] == jobs.RUNNING, 'this process\'s own Job is untouched' + assert report['processing_domains'] == 1 and list(download_manager.processing_domains) == [] + assert report['downloads'] == 1 + test_session.expire_all() + 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' + + # Idempotent: a clean state reports nothing. + assert not any(reconcile_after_predecessor(api_app).values()) diff --git a/wrolpi/test/test_root_api.py b/wrolpi/test/test_root_api.py index 73cfb9d4..055e7364 100644 --- a/wrolpi/test/test_root_api.py +++ b/wrolpi/test/test_root_api.py @@ -421,6 +421,38 @@ 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.""" + import multiprocessing + from unittest import mock + from sanic.worker.process import Worker, WorkerProcess + from wrolpi.api_utils import api_app + from wrolpi.perpetual import PERPETUAL_PROCESS_NAME + + def sanic_name(name): + worker = Worker(name, name, lambda: None, {}, multiprocessing.get_context(), {}, 1) + return next(iter(worker.processes)).name + + server = sanic_name(f'{WorkerProcess.SERVER_LABEL}-0') + perpetual_name = sanic_name(PERPETUAL_PROCESS_NAME) + multiplexer = mock.Mock() + multiplexer.workers = { + 'Sanic-Main': {'pid': 1}, + server: {'pid': 2, 'state': 'ACKED'}, + perpetual_name: {'pid': 3, 'state': 'STARTED'}, + sanic_name('Inspector'): {'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'] == { + server: {'pid': 2, 'state': 'ACKED'}, + perpetual_name: {'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..52e8c2da 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,57 @@ 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) would stall the event loop of the process that + also runs downloads and the file 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.