diff --git a/main.py b/main.py index 3360886d..92cd3506 100755 --- a/main.py +++ b/main.py @@ -22,7 +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.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 @@ -203,7 +203,7 @@ async def main_process_startup(app: Sanic): # 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) + register_perpetual_process(app) @api_app.listener('after_server_start') # FileConfigs need to be initialized first. @@ -332,9 +332,11 @@ 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.""" + """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}') 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 24e0e050..d961cdf2 100644 --- a/wrolpi/contexts.py +++ b/wrolpi/contexts.py @@ -69,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() @@ -226,6 +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() + 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/jobs.py b/wrolpi/jobs.py index bc39c6e7..3591760c 100644 --- a/wrolpi/jobs.py +++ b/wrolpi/jobs.py @@ -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')) diff --git a/wrolpi/perpetual.py b/wrolpi/perpetual.py index fa5760c4..236821b9 100644 --- a/wrolpi/perpetual.py +++ b/wrolpi/perpetual.py @@ -3,8 +3,13 @@ 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. +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. @@ -35,6 +40,11 @@ PERPETUAL_PROCESS_DEAD_STATES = frozenset({'FAILED', 'COMPLETED', 'TERMINATED'}) 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 @@ -91,8 +101,8 @@ async def run_loops(loops: Iterable[PerpetualLoop], stop: asyncio.Event, cancel_ 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.""" + """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) @@ -116,13 +126,96 @@ async def _shutdown_singletons(): 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.jobs import fail_orphaned_jobs + 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: + 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`).""" + 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, @@ -148,9 +241,19 @@ def run_perpetual_process(): # 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. @@ -163,6 +266,26 @@ def perpetual_process_needs_restart(workers: dict, now: float, last_request: flo 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 @@ -172,7 +295,6 @@ async def perpetual_process_health_check(): 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) @@ -183,10 +305,8 @@ async def perpetual_process_health_check(): 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): + if not _claim_restart_request(api_app, workers, time.time()): 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/test/test_api_utils.py b/wrolpi/test/test_api_utils.py index 03452d2d..ac093a86 100644 --- a/wrolpi/test/test_api_utils.py +++ b/wrolpi/test/test_api_utils.py @@ -5,10 +5,18 @@ from unittest import mock import pytest from wrolpi import perpetual -from wrolpi.api_utils import perpetual_signal, per_worker_task, start_per_worker_tasks, stop_per_worker_tasks +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_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_registrars_are_noops_under_pytest(): async def f(): pass 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_perpetual.py b/wrolpi/test/test_perpetual.py index 7c63ff09..423e9ce9 100644 --- a/wrolpi/test/test_perpetual.py +++ b/wrolpi/test/test_perpetual.py @@ -1,20 +1,32 @@ """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, ) @@ -111,19 +123,36 @@ async def test_serve_perpetual_loops_stops_on_sigterm(): 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'}} + """Only a process the manager reports as gone triggers a restart request, at most once per cooldown. - assert perpetual_process_needs_restart(dead, now, last_request=0) is True + 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 - 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 + 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 @@ -135,14 +164,17 @@ async def test_health_check_requests_restart_once(async_client, monkeypatch): 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) + 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'}} - monkeypatch.setattr(perpetual, '_last_restart_request', 0.0) + api_app.shared_ctx.perpetual_restart_requested_at.value = 0.0 await perpetual.perpetual_process_health_check() multiplexer.restart.assert_called_once() @@ -154,3 +186,150 @@ async def test_health_check_without_multiplexer_is_a_noop(async_client, monkeypa 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 62550906..055e7364 100644 --- a/wrolpi/test/test_root_api.py +++ b/wrolpi/test/test_root_api.py @@ -424,23 +424,32 @@ async def test_get_status(async_client, test_session): @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}, - 'Sanic-Server-0': {'pid': 2, 'state': 'ACKED'}, - 'Sanic-Perpetual-0': {'pid': 3, 'state': 'STARTED'}, - 'Sanic-Inspector-0': {'pid': 4, 'state': 'STARTED'}, + 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'] == { - 'Sanic-Server-0': {'pid': 2, 'state': 'ACKED'}, - 'Sanic-Perpetual-0': {'pid': 3, 'state': 'STARTED'}, + server: {'pid': 2, 'state': 'ACKED'}, + perpetual_name: {'pid': 3, 'state': 'STARTED'}, } diff --git a/wrolpi/test/test_upgrade.py b/wrolpi/test/test_upgrade.py index 2c52993c..52e8c2da 100644 --- a/wrolpi/test/test_upgrade.py +++ b/wrolpi/test/test_upgrade.py @@ -364,9 +364,8 @@ async def test_status_endpoint_includes_git_branch(async_client): 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. + 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