rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

perpetual: survive a predecessor's death, exit cleanly, coordinate restarts

Browse source 
Review fixes for the managed perpetual process.

Shutdown: cancel_background_tasks / cancel_refresh_tasks gather with
return_exceptions, so a cancelled child's CancelledError no longer aborts
the shutdown that called them (the process exited 1 with a download in
flight).

Restart after a crash: reconcile_after_predecessor heals shared locks a
dead process left held, fails RUNNING Jobs owned by a dead pid, clears
processing_domains, renews pending Downloads, and resets the file-worker
status before the loops start.  N server workers issue one restart
request per cooldown through a shared timestamp in shared_ctx.

The log-level sync now also runs in the perpetual process.  Manage() is
registered through register_perpetual_process so it is testable; the
entry point refuses to run with no loops registered (spawn).

Tests exercise the real entry point in a forked process with a live
background task under SIGINT and SIGTERM, Sanic's real WorkerManager
marking a crashed managed process FAILED and reviving it on restart,
Sanic's process naming and ProcessState enum, the listener registration,
and the reconcile step.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-09-22 16:55:55 -06:00
parent 8d395eaf40
commit 587bd8d2bf
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
10 changed files with 404 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

8
main.py
View file

@ -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}')

24
wrolpi/common.py
View file

@ -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):

6
wrolpi/contexts.py
View file

@ -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()

18
wrolpi/jobs.py
View file

@ -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'))

138
wrolpi/perpetual.py
View file

@ -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)

10
wrolpi/test/test_api_utils.py
View file

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

16
wrolpi/test/test_common.py
View file

@ -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)

201
wrolpi/test/test_perpetual.py
View file

@ -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())

19
wrolpi/test/test_root_api.py
View file

@ -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'},
}

5
wrolpi/test/test_upgrade.py
View file

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