mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-25 18:40:06 -04:00
api: run the perpetual loops in one Sanic-managed process
The singleton background loops (file worker, downloads, switches, jobs, bulk tag, flag checks, update check) now run in a dedicated process that Sanic's Worker Manager owns: `run_perpetual_process` is registered with `app.manager.manage()` from `main_process_ready`. The manager guarantees a single instance and SIGTERMs it on shutdown. No server worker runs the loops, so the owner-pid / heartbeat / watchdog claim scheme is deleted rather than hardened. `perpetual_signal` keeps its name and is now a plain registrar. A new `per_worker_task` registrar covers loops every server worker must run for itself: syncing its log level (previously only the owner did, so the other workers never picked up level changes) and a health check that asks the manager to restart the perpetual process when it is FAILED (Sanic marks a dead managed process but does not restart it). Downloads are started as background tasks instead of Sanic signals; the perpetual process has no server, so no finalized signal router. The update check and the upgrade-check endpoint run git in a thread via `refresh_update_status`. /api/status lists the perpetual process next to the servers. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Roland
parent
ca80f9428f
commit
8d395eaf40
14 changed files with 612 additions and 515 deletions
33
main.py
33
main.py
|
|
@ -12,7 +12,7 @@ from sanic.signals import Event
|
|||
from modules.videos.lib import get_videos_downloader_config
|
||||
from wrolpi import flags
|
||||
from wrolpi import root_api # noqa
|
||||
from wrolpi.api_utils import api_app, perpetual_signal
|
||||
from wrolpi.api_utils import api_app, perpetual_signal, per_worker_task
|
||||
from wrolpi.common import logger, check_media_directory, set_log_level, limit_concurrent, \
|
||||
cancel_refresh_tasks, cancel_background_tasks, get_wrolpi_config, can_connect_to_server, wrol_mode_enabled, \
|
||||
create_empty_config_files, TRACE_LEVEL
|
||||
|
|
@ -22,6 +22,7 @@ from modules.videos.cookies import lock_cookies
|
|||
from wrolpi.downloader import download_manager, get_download_manager_config
|
||||
from wrolpi.errors import WROLModeEnabled
|
||||
from wrolpi.files.worker import file_worker
|
||||
from wrolpi.perpetual import run_perpetual_process, PERPETUAL_PROCESS_NAME, perpetual_process_health_check
|
||||
from wrolpi.vars import PROJECT_DIR, DOCKERIZED, INTERNET_SERVER
|
||||
from wrolpi.version import get_version_string
|
||||
|
||||
|
|
@ -168,11 +169,12 @@ def main():
|
|||
@api_app.main_process_ready
|
||||
async def main_process_startup(app: Sanic):
|
||||
"""
|
||||
Initializes multiprocessing tools, flags, etc.
|
||||
Initializes multiprocessing tools, flags, etc., then registers the perpetual process.
|
||||
|
||||
Performed only once when the server starts, this is done before server processes are forked.
|
||||
|
||||
@warning: This is NOT run after auto-reload! You must stop and start Sanic.
|
||||
@warning: This is NOT run after auto-reload! You must stop and start Sanic. (The perpetual process is
|
||||
`transient`, so the manager does restart it on auto-reload.)
|
||||
"""
|
||||
logger.debug('main_process_startup')
|
||||
|
||||
|
|
@ -199,6 +201,10 @@ async def main_process_startup(app: Sanic):
|
|||
except Exception as e:
|
||||
logger.error('Failed to create/migrate the database', exc_info=e)
|
||||
|
||||
# The singleton background loops run in ONE process the Worker Manager owns (not in a server worker), so no
|
||||
# worker can claim or take them over. `shared_ctx` is populated above; the fork inherits it.
|
||||
app.manager.manage(PERPETUAL_PROCESS_NAME, run_perpetual_process, {}, transient=True)
|
||||
|
||||
|
||||
@api_app.listener('after_server_start') # FileConfigs need to be initialized first.
|
||||
async def initialize_configs(app: Sanic):
|
||||
|
|
@ -220,7 +226,9 @@ async def initialize_configs(app: Sanic):
|
|||
@api_app.listener('reload_process_stop')
|
||||
@limit_concurrent(1)
|
||||
async def handle_server_shutdown(*args, **kwargs):
|
||||
"""Stop downloads and securely clear sensitive data when server is shutting down."""
|
||||
"""Stop downloads and securely clear sensitive data when server is shutting down.
|
||||
|
||||
Runs in each server worker. The perpetual process does the same for itself (wrolpi/perpetual.py)."""
|
||||
logger.warning('Shutting down')
|
||||
try:
|
||||
download_manager.stop()
|
||||
|
|
@ -324,7 +332,7 @@ async def perpetual_file_worker_queue():
|
|||
file_worker.check_queue_stall(processed=processed)
|
||||
|
||||
|
||||
@perpetual_signal(sleep=1)
|
||||
@per_worker_task(sleep=1)
|
||||
async def perpetual_check_log_level():
|
||||
"""Copies global log level into this Sanic worker's logger."""
|
||||
log_level = api_app.shared_ctx.log_level.value
|
||||
|
|
@ -333,6 +341,10 @@ async def perpetual_check_log_level():
|
|||
set_log_level(log_level, warn_level=False)
|
||||
|
||||
|
||||
# Every server worker watches the perpetual process and asks the manager to restart it if it dies.
|
||||
per_worker_task(sleep=10)(perpetual_process_health_check)
|
||||
|
||||
|
||||
@perpetual_signal(sleep=10)
|
||||
async def perpetual_check_db_is_up_worker():
|
||||
try:
|
||||
|
|
@ -416,7 +428,7 @@ async def perpetual_check_for_updates():
|
|||
Only runs on native installs (not Docker).
|
||||
"""
|
||||
# Local import to avoid loading upgrade module at startup
|
||||
from wrolpi.upgrade import check_for_update
|
||||
from wrolpi.upgrade import refresh_update_status
|
||||
|
||||
# Don't check for updates in Docker environments
|
||||
if DOCKERIZED:
|
||||
|
|
@ -431,13 +443,8 @@ async def perpetual_check_for_updates():
|
|||
return
|
||||
|
||||
try:
|
||||
result = check_for_update(fetch=True)
|
||||
# Store in shared_ctx.status (a manager.dict) to share across all workers.
|
||||
api_app.shared_ctx.status['update_available'] = result.get('update_available', False)
|
||||
api_app.shared_ctx.status['latest_commit'] = result.get('latest_commit')
|
||||
api_app.shared_ctx.status['current_commit'] = result.get('current_commit')
|
||||
api_app.shared_ctx.status['commits_behind'] = result.get('commits_behind', 0)
|
||||
api_app.shared_ctx.status['git_branch'] = result.get('branch')
|
||||
# Runs `git fetch` in a thread and stores the result in shared_ctx.status for all workers.
|
||||
result = await refresh_update_status(api_app.shared_ctx.status)
|
||||
|
||||
if result.get('update_available'):
|
||||
logger.info(f"Update available: {result['commits_behind']} commits behind on {result['branch']}")
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -59,13 +59,6 @@ def attach_shared_contexts(app: Sanic):
|
|||
|
||||
app.shared_ctx.single_tasks_started = multiprocessing.Event()
|
||||
app.shared_ctx.flags_initialized = multiprocessing.Event()
|
||||
app.shared_ctx.perpetual_tasks_started = multiprocessing.Event()
|
||||
# PID of the worker that started the single-process perpetual loops.
|
||||
# The Event alone survives that worker dying (Sanic auto_reload / crash),
|
||||
# so replacement workers check this PID before skipping startup.
|
||||
app.shared_ctx.perpetual_tasks_owner_pid = multiprocessing.Value(ctypes.c_int, 0)
|
||||
app.shared_ctx.perpetual_tasks_heartbeat = multiprocessing.Value(ctypes.c_double, 0.0)
|
||||
app.shared_ctx.perpetual_tasks_lock = multiprocessing.Lock()
|
||||
|
||||
# Transcode
|
||||
# Transcoding saturates every core; only one ffmpeg transcode may run machine-wide,
|
||||
|
|
@ -106,7 +99,7 @@ def attach_shared_contexts(app: Sanic):
|
|||
app.shared_ctx.file_worker_jobs = manager.dict()
|
||||
|
||||
# Jobs (wrolpi/jobs.py): FIFO queue of registered functions run one at a time by the
|
||||
# perpetual-tasks owner. Never persisted; an API restart loses them.
|
||||
# perpetual process. Never persisted; an API restart loses them.
|
||||
app.shared_ctx.jobs = manager.dict()
|
||||
app.shared_ctx.jobs_queue = manager.Queue()
|
||||
app.shared_ctx.jobs_lock = multiprocessing.Lock()
|
||||
|
|
@ -233,11 +226,6 @@ def reset_shared_contexts(app: Sanic):
|
|||
del app.shared_ctx.events_history[:]
|
||||
app.shared_ctx.single_tasks_started.clear()
|
||||
app.shared_ctx.flags_initialized.clear()
|
||||
app.shared_ctx.perpetual_tasks_started.clear()
|
||||
if hasattr(app.shared_ctx, 'perpetual_tasks_owner_pid'):
|
||||
app.shared_ctx.perpetual_tasks_owner_pid.value = 0
|
||||
if hasattr(app.shared_ctx, 'perpetual_tasks_heartbeat'):
|
||||
app.shared_ctx.perpetual_tasks_heartbeat.value = 0.0
|
||||
|
||||
# Do not start downloads when reloading.
|
||||
app.shared_ctx.download_manager_stopped.set()
|
||||
|
|
|
|||
|
|
@ -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'):
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
"""In-memory FIFO queue of background Jobs.
|
||||
|
||||
A Job is a call of a *registered* function with JSON-able kwargs. Any Sanic worker may enqueue a
|
||||
Job; the perpetual-tasks owner process runs them one at a time, in order. Python logging emitted
|
||||
Job; the perpetual process runs them one at a time, in order. Python logging emitted
|
||||
while a Job runs, and the output of subprocesses started through `JobContext.run_command`, are
|
||||
captured into the Job's log. A caller may wait for a Job, poll it, or cancel it.
|
||||
|
||||
|
|
@ -406,7 +406,7 @@ async def process_job_queue() -> int:
|
|||
|
||||
@perpetual_signal(sleep=0.5)
|
||||
async def job_worker():
|
||||
"""The single consumer of the Job queue; runs in the perpetual-tasks owner process only."""
|
||||
"""The single consumer of the Job queue; runs in the perpetual process only."""
|
||||
try:
|
||||
job_id = _queue().get_nowait()
|
||||
except queue.Empty:
|
||||
|
|
|
|||
192
wrolpi/perpetual.py
Normal file
192
wrolpi/perpetual.py
Normal file
|
|
@ -0,0 +1,192 @@
|
|||
"""WROLPi's background loops.
|
||||
|
||||
Singleton loops (file worker, downloads, switches, jobs, bulk tag, flag checks, update check)
|
||||
run in ONE process that Sanic's Worker Manager owns: `run_perpetual_process` is handed to
|
||||
`app.manager.manage()` from `main_process_ready`. The manager guarantees a single instance and
|
||||
sends it SIGTERM on shutdown; a Sanic server worker never runs these loops, so there is nothing
|
||||
for a worker to claim, hand off, or take over.
|
||||
|
||||
Per-worker loops (things every Sanic server process must do for itself, like syncing its log
|
||||
level) run as asyncio tasks inside each server worker.
|
||||
|
||||
Both kinds are registered at import time with the decorators in `wrolpi.api_utils`
|
||||
(`perpetual_signal` and `per_worker_task`) and are no-ops under pytest; tests call the loop
|
||||
bodies directly.
|
||||
"""
|
||||
import asyncio
|
||||
import os
|
||||
import signal
|
||||
import time
|
||||
from asyncio import CancelledError
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable, Awaitable, Iterable
|
||||
|
||||
from wrolpi.common import logger
|
||||
|
||||
logger = logger.getChild(__name__)
|
||||
|
||||
PERPETUAL_PROCESS_NAME = 'Perpetual'
|
||||
# Sanic names a managed process "-".join(["Sanic", name, index]) (sanic/worker/process.py).
|
||||
PERPETUAL_PROCESS_IDENT = f'Sanic-{PERPETUAL_PROCESS_NAME}-0'
|
||||
# A managed (non-server) process never reaches ACKED; these are the states the manager reports for one that
|
||||
# is gone. See sanic/worker/constants.py ProcessState.
|
||||
PERPETUAL_PROCESS_DEAD_STATES = frozenset({'FAILED', 'COMPLETED', 'TERMINATED'})
|
||||
# A restart request is idempotent to the manager, but five workers noticing one death should not log five times.
|
||||
PERPETUAL_RESTART_COOLDOWN = 60
|
||||
# How long the runner waits for cancelled loops before abandoning them at shutdown.
|
||||
PERPETUAL_CANCEL_TIMEOUT = 15
|
||||
|
||||
|
||||
@dataclass
|
||||
class PerpetualLoop:
|
||||
name: str
|
||||
func: Callable[[], Awaitable]
|
||||
sleep: float
|
||||
|
||||
|
||||
# Filled at import time by the decorators in wrolpi.api_utils.
|
||||
PERPETUAL_LOOPS: list[PerpetualLoop] = []
|
||||
PER_WORKER_TASKS: list[PerpetualLoop] = []
|
||||
|
||||
|
||||
async def run_loop(loop: PerpetualLoop, stop: asyncio.Event):
|
||||
"""Call `loop.func` forever, `loop.sleep` seconds after each call returns, until `stop` is set.
|
||||
|
||||
An ordinary Exception is logged and the loop continues. A long call is never re-entered: the next call
|
||||
starts only after the previous one returns. Cancellation ends the loop; the runner is the only canceller.
|
||||
"""
|
||||
while not stop.is_set():
|
||||
start = time.time()
|
||||
try:
|
||||
await loop.func()
|
||||
except CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f'Perpetual loop {loop.name} had error', exc_info=e)
|
||||
finally:
|
||||
logger.trace(f'Perpetual loop {loop.name} took {time.time() - start:.3f} seconds')
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(stop.wait(), timeout=loop.sleep)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
|
||||
async def run_loops(loops: Iterable[PerpetualLoop], stop: asyncio.Event, cancel_timeout: float = PERPETUAL_CANCEL_TIMEOUT):
|
||||
"""Run every loop as its own task until `stop` is set, then cancel them all and wait (bounded) for them to end."""
|
||||
tasks = [asyncio.create_task(run_loop(loop, stop), name=loop.name) for loop in loops]
|
||||
if not tasks:
|
||||
await stop.wait()
|
||||
return
|
||||
|
||||
await stop.wait()
|
||||
for task in tasks:
|
||||
task.cancel()
|
||||
done, pending = await asyncio.wait(tasks, timeout=cancel_timeout)
|
||||
for task in pending:
|
||||
logger.warning(f'Perpetual loop {task.get_name()} did not stop within {cancel_timeout}s; abandoning it')
|
||||
for task in done:
|
||||
if not task.cancelled() and task.exception():
|
||||
logger.error(f'Perpetual loop {task.get_name()} ended with error', exc_info=task.exception())
|
||||
|
||||
|
||||
def install_stop_signal_handlers(loop: asyncio.AbstractEventLoop, stop: asyncio.Event):
|
||||
"""SIGTERM (the manager's shutdown) and SIGINT set `stop`; a second signal is ignored rather than killing us
|
||||
mid-cleanup."""
|
||||
for sig in (signal.SIGTERM, signal.SIGINT):
|
||||
loop.add_signal_handler(sig, stop.set)
|
||||
|
||||
|
||||
async def _shutdown_singletons():
|
||||
"""Stop the singleton services that live in this process before its loops are cancelled."""
|
||||
from wrolpi.common import cancel_refresh_tasks, cancel_background_tasks
|
||||
from wrolpi.downloader import download_manager
|
||||
|
||||
try:
|
||||
download_manager.stop()
|
||||
except Exception as e:
|
||||
logger.error('download_manager.stop failed during perpetual shutdown', exc_info=e)
|
||||
try:
|
||||
await cancel_refresh_tasks()
|
||||
except Exception as e:
|
||||
logger.error('cancel_refresh_tasks failed during perpetual shutdown', exc_info=e)
|
||||
try:
|
||||
await cancel_background_tasks()
|
||||
except Exception as e:
|
||||
logger.error('cancel_background_tasks failed during perpetual shutdown', exc_info=e)
|
||||
|
||||
|
||||
def _initialize_process():
|
||||
"""What Sanic's `after_server_start` listeners do for a server worker, done here for this process: each
|
||||
process owns its FileConfig objects (they share `shared_ctx.*config`)."""
|
||||
from wrolpi.api_utils import api_app
|
||||
from wrolpi.contexts import initialize_configs_contexts
|
||||
|
||||
initialize_configs_contexts(api_app)
|
||||
|
||||
|
||||
async def serve_perpetual_loops(loops: Iterable[PerpetualLoop], stop: asyncio.Event | None = None,
|
||||
initialize: Callable[[], None] = _initialize_process):
|
||||
"""The perpetual process's main coroutine."""
|
||||
stop = stop or asyncio.Event()
|
||||
install_stop_signal_handlers(asyncio.get_running_loop(), stop)
|
||||
initialize()
|
||||
loops = list(loops)
|
||||
logger.info(f'Perpetual process started pid={os.getpid()} loops={[i.name for i in loops]}')
|
||||
runner = asyncio.create_task(run_loops(loops, stop), name='perpetual-runner')
|
||||
await stop.wait()
|
||||
logger.warning('Perpetual process stopping')
|
||||
await _shutdown_singletons()
|
||||
await runner
|
||||
logger.info('Perpetual process stopped')
|
||||
|
||||
|
||||
def run_perpetual_process():
|
||||
"""Entry point handed to `app.manager.manage()`. Runs in its own (forked) process: it inherits `api_app`,
|
||||
the populated `shared_ctx`, and the module singletons, and needs no arguments."""
|
||||
# The fork inherited the manager's Python signal handlers; until the event loop installs ours, a signal
|
||||
# must do the default thing (exit) rather than run the manager's handler inside this process.
|
||||
signal.signal(signal.SIGTERM, signal.SIG_DFL)
|
||||
signal.signal(signal.SIGINT, signal.SIG_DFL)
|
||||
asyncio.run(serve_perpetual_loops(PERPETUAL_LOOPS))
|
||||
|
||||
|
||||
def perpetual_process_needs_restart(workers: dict, now: float, last_request: float) -> bool:
|
||||
"""True if the manager reports the perpetual process as gone and no restart was requested recently.
|
||||
|
||||
`workers` is `app.m.workers` (a dict keyed by process name). A missing entry is not "dead": the manager
|
||||
has not registered the process yet.
|
||||
"""
|
||||
entry = workers.get(PERPETUAL_PROCESS_IDENT)
|
||||
if not entry or entry.get('state') not in PERPETUAL_PROCESS_DEAD_STATES:
|
||||
return False
|
||||
return (now - last_request) >= PERPETUAL_RESTART_COOLDOWN
|
||||
|
||||
|
||||
_last_restart_request = 0.0
|
||||
|
||||
|
||||
async def perpetual_process_health_check():
|
||||
"""Runs in every server worker: ask the manager to restart the perpetual process if it has died.
|
||||
|
||||
Sanic only marks a dead managed process FAILED; it does not restart it. This is a restart request to the
|
||||
manager, not a takeover: the loops still run in exactly one process.
|
||||
"""
|
||||
global _last_restart_request
|
||||
from wrolpi.api_utils import api_app
|
||||
|
||||
multiplexer = getattr(api_app, 'multiplexer', None)
|
||||
if multiplexer is None:
|
||||
return
|
||||
try:
|
||||
workers = multiplexer.workers
|
||||
except Exception as e:
|
||||
logger.debug('Could not read worker states', exc_info=e)
|
||||
return
|
||||
now = time.time()
|
||||
if not perpetual_process_needs_restart(workers, now, _last_restart_request):
|
||||
return
|
||||
_last_restart_request = now
|
||||
state = workers[PERPETUAL_PROCESS_IDENT].get('state')
|
||||
logger.warning(f'Perpetual process is {state}; requesting restart pid={os.getpid()}')
|
||||
multiplexer.restart(PERPETUAL_PROCESS_IDENT)
|
||||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,222 +1,71 @@
|
|||
"""Tests for perpetual_signal cancellation / reschedule behavior."""
|
||||
"""Tests for the perpetual / per-worker loop registrars and the per-worker task listeners."""
|
||||
import asyncio
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from wrolpi.api_utils import (
|
||||
_app_is_stopping,
|
||||
_run_perpetual_iteration,
|
||||
api_app,
|
||||
)
|
||||
from wrolpi import perpetual
|
||||
from wrolpi.api_utils import perpetual_signal, per_worker_task, start_per_worker_tasks, stop_per_worker_tasks
|
||||
from wrolpi.perpetual import PerpetualLoop
|
||||
|
||||
|
||||
def test_app_is_stopping_false_by_default():
|
||||
api_app.ctx.perpetual_shutdown = False
|
||||
api_app.state.is_stopping = False
|
||||
# is_started/is_running are not a shutdown signal: docker Sanic workers
|
||||
# often have is_started=True and is_running=False while serving.
|
||||
api_app.state.is_started = True
|
||||
api_app.state.is_running = False
|
||||
assert _app_is_stopping() is False
|
||||
def test_registrars_are_noops_under_pytest():
|
||||
async def f():
|
||||
pass
|
||||
|
||||
assert perpetual_signal(sleep=3)(f) is f
|
||||
assert per_worker_task(sleep=3)(f) is f
|
||||
assert all(i.func is not f for i in perpetual.PERPETUAL_LOOPS)
|
||||
assert all(i.func is not f for i in perpetual.PER_WORKER_TASKS)
|
||||
|
||||
|
||||
def test_app_is_stopping_when_shutdown_flag_set():
|
||||
api_app.ctx.perpetual_shutdown = True
|
||||
try:
|
||||
assert _app_is_stopping() is True
|
||||
finally:
|
||||
api_app.ctx.perpetual_shutdown = False
|
||||
|
||||
|
||||
def test_app_is_stopping_when_sanic_is_stopping():
|
||||
api_app.ctx.perpetual_shutdown = False
|
||||
api_app.state.is_stopping = True
|
||||
try:
|
||||
assert _app_is_stopping() is True
|
||||
finally:
|
||||
api_app.state.is_stopping = False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_perpetual_iteration_reschedules_after_unexpected_cancel(monkeypatch):
|
||||
"""CancelledError must not silently kill the perpetual loop.
|
||||
|
||||
Regression: perpetual_signal caught CancelledError, set cancelled=True,
|
||||
and skipped re-dispatch with no log. The file-worker pump died for the
|
||||
life of the process.
|
||||
"""
|
||||
dispatched = []
|
||||
|
||||
async def boom():
|
||||
raise asyncio.CancelledError()
|
||||
|
||||
async def fake_dispatch(event):
|
||||
dispatched.append(event)
|
||||
|
||||
def test_registrars_record_one_loop_each(monkeypatch):
|
||||
"""Outside pytest, each decorator registers exactly one loop in its own registry and returns the function."""
|
||||
singletons, everywhere_ = [], []
|
||||
monkeypatch.setattr('wrolpi.api_utils.PYTEST', False)
|
||||
monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: False)
|
||||
monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch)
|
||||
monkeypatch.setattr('wrolpi.api_utils.asyncio.sleep', mock.AsyncMock())
|
||||
monkeypatch.setattr('wrolpi.api_utils.PERPETUAL_LOOPS', singletons)
|
||||
monkeypatch.setattr('wrolpi.api_utils.PER_WORKER_TASKS', everywhere_)
|
||||
|
||||
await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0)
|
||||
assert dispatched == ['wrolpi.perpetual.test_worker']
|
||||
async def singleton():
|
||||
pass
|
||||
|
||||
async def everywhere():
|
||||
pass
|
||||
|
||||
assert perpetual_signal(sleep=7)(singleton) is singleton
|
||||
assert per_worker_task(sleep=2)(everywhere) is everywhere
|
||||
|
||||
assert [(i.name, i.sleep) for i in singletons] == [('singleton', 7)]
|
||||
assert [(i.name, i.sleep) for i in everywhere_] == [('everywhere', 2)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_perpetual_iteration_stops_on_shutdown_cancel(monkeypatch):
|
||||
"""CancelledError during shutdown must stay terminal (do not reschedule)."""
|
||||
dispatched = []
|
||||
async def test_per_worker_tasks_start_and_stop_with_the_server(monkeypatch):
|
||||
"""after_server_start runs every per-worker loop as a task; before_server_stop ends them."""
|
||||
calls = []
|
||||
|
||||
async def boom():
|
||||
raise asyncio.CancelledError()
|
||||
async def tick():
|
||||
calls.append(1)
|
||||
|
||||
async def fake_dispatch(event):
|
||||
dispatched.append(event)
|
||||
monkeypatch.setattr('wrolpi.api_utils.PER_WORKER_TASKS', [PerpetualLoop('tick', tick, 0.01)])
|
||||
app = mock.Mock()
|
||||
app.ctx = mock.Mock()
|
||||
|
||||
monkeypatch.setattr('wrolpi.api_utils.PYTEST', False)
|
||||
monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: True)
|
||||
monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch)
|
||||
await start_per_worker_tasks(app)
|
||||
await asyncio.sleep(0.05)
|
||||
assert calls, 'the per-worker loop should be running'
|
||||
assert len(app.ctx.per_worker_tasks) == 1
|
||||
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0)
|
||||
assert dispatched == []
|
||||
|
||||
|
||||
def test_claim_perpetual_tasks_when_never_started(async_client):
|
||||
"""The first worker to start must claim the perpetual loops."""
|
||||
from wrolpi.api_utils import _claim_perpetual_tasks, api_app
|
||||
|
||||
api_app.shared_ctx.perpetual_tasks_started.clear()
|
||||
assert _claim_perpetual_tasks(api_app) is True
|
||||
|
||||
|
||||
def test_claim_perpetual_tasks_skips_when_owner_is_alive(async_client):
|
||||
"""A live owner keeps the single-process loops from starting twice."""
|
||||
import os
|
||||
from wrolpi.api_utils import _claim_perpetual_tasks, api_app
|
||||
|
||||
api_app.shared_ctx.perpetual_tasks_started.set()
|
||||
api_app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid()
|
||||
assert _claim_perpetual_tasks(api_app) is False
|
||||
|
||||
|
||||
def test_claim_perpetual_tasks_when_owner_process_is_dead(async_client):
|
||||
"""A dead owner must not block a replacement worker from restarting the loops.
|
||||
|
||||
``perpetual_tasks_started`` is a shared Event. Sanic docker runs with
|
||||
auto_reload, and a worker crash/recycle leaves that Event set. The
|
||||
replacement process's after_server_start then no-ops, and file processing
|
||||
(plus downloads, switches, …) is dead until a full API restart.
|
||||
|
||||
This is the unconfirmed wedge from the 2026-08-18 incident: last file-worker
|
||||
activity from one PID, then silence across every process, recovered only by
|
||||
restarting the API.
|
||||
"""
|
||||
from wrolpi.api_utils import _claim_perpetual_tasks, api_app
|
||||
|
||||
api_app.shared_ctx.perpetual_tasks_started.set()
|
||||
# A PID that cannot be running on this host.
|
||||
api_app.shared_ctx.perpetual_tasks_owner_pid.value = 2**31 - 1
|
||||
assert _claim_perpetual_tasks(api_app) is True
|
||||
assert api_app.shared_ctx.perpetual_tasks_owner_pid.value != 2**31 - 1, \
|
||||
'claimer must record this process as the new owner'
|
||||
|
||||
|
||||
def test_claim_perpetual_tasks_when_heartbeat_is_stale(async_client):
|
||||
"""A live PID with a stale heartbeat is treated as a dead owner.
|
||||
|
||||
After a crash the kernel can reuse the owner PID for an unrelated
|
||||
process, so pid_is_running alone is not enough. The owner must keep
|
||||
a heartbeat; if it goes quiet, another worker reclaims.
|
||||
"""
|
||||
import os
|
||||
import time
|
||||
from wrolpi.api_utils import OWNER_HEARTBEAT_STALE_SECONDS, _claim_perpetual_tasks, api_app
|
||||
|
||||
api_app.shared_ctx.perpetual_tasks_started.set()
|
||||
api_app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid()
|
||||
api_app.shared_ctx.perpetual_tasks_heartbeat.value = time.time() - OWNER_HEARTBEAT_STALE_SECONDS - 1
|
||||
assert _claim_perpetual_tasks(api_app) is True
|
||||
|
||||
|
||||
def test_non_owner_starts_watchdog_not_file_worker(async_client):
|
||||
"""Non-owner workers run a reclaim watchdog, not a second file-worker pump.
|
||||
|
||||
Five concurrent pumps would race on the shared status dict, the busy
|
||||
flag, and the DB — overlapping refreshes/moves are how stale
|
||||
primary_path rows appear. File jobs stay single-consumer; other
|
||||
workers only notice a dead owner and take over.
|
||||
"""
|
||||
import os
|
||||
import time
|
||||
from wrolpi.api_utils import (
|
||||
FILE_WORKER_PERPETUAL_EVENT,
|
||||
OWNER_WATCHDOG_EVENT,
|
||||
_perpetual_events_for_this_process,
|
||||
api_app,
|
||||
)
|
||||
|
||||
api_app.shared_ctx.perpetual_tasks_started.set()
|
||||
api_app.shared_ctx.perpetual_tasks_owner_pid.value = os.getpid()
|
||||
api_app.shared_ctx.perpetual_tasks_heartbeat.value = time.time()
|
||||
events = _perpetual_events_for_this_process(api_app)
|
||||
assert events == [OWNER_WATCHDOG_EVENT]
|
||||
assert FILE_WORKER_PERPETUAL_EVENT not in events
|
||||
await stop_per_worker_tasks(app)
|
||||
assert all(t.done() for t in app.ctx.per_worker_tasks)
|
||||
count = len(calls)
|
||||
await asyncio.sleep(0.03)
|
||||
assert len(calls) == count, 'the loop must not run after stop'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_watchdog_reclaims_and_starts_pumps_when_owner_is_dead(async_client, monkeypatch):
|
||||
"""The watchdog, not after_server_start, is what heals a dead owner mid-life.
|
||||
|
||||
Existing workers have already passed start_perpetual_tasks. If the
|
||||
owner dies later, a periodic tick must claim and dispatch the pumps.
|
||||
"""
|
||||
import os
|
||||
from wrolpi.api_utils import perpetual_owner_watchdog, api_app
|
||||
|
||||
dispatched = []
|
||||
|
||||
async def fake_dispatch(event):
|
||||
dispatched.append(event)
|
||||
|
||||
# A sibling test's async_client teardown runs the real before_server_stop listener,
|
||||
# which leaves perpetual_shutdown=True on the module-global app; the watchdog would
|
||||
# then return before reclaiming. This test is about a dead owner, not shutdown.
|
||||
monkeypatch.setattr(api_app.ctx, 'perpetual_shutdown', False, raising=False)
|
||||
monkeypatch.setattr(api_app.state, 'is_stopping', False, raising=False)
|
||||
|
||||
api_app.shared_ctx.perpetual_tasks_started.set()
|
||||
api_app.shared_ctx.perpetual_tasks_owner_pid.value = 2**31 - 1
|
||||
monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch)
|
||||
monkeypatch.setattr('wrolpi.api_utils.PERPETUAL_WORKERS', [
|
||||
'wrolpi.perpetual.perpetual_file_worker_queue',
|
||||
'wrolpi.perpetual.perpetual_owner_watchdog',
|
||||
])
|
||||
|
||||
await perpetual_owner_watchdog()
|
||||
|
||||
assert api_app.shared_ctx.perpetual_tasks_owner_pid.value == os.getpid()
|
||||
assert api_app.shared_ctx.perpetual_tasks_heartbeat.value > 0
|
||||
assert 'wrolpi.perpetual.perpetual_file_worker_queue' in dispatched
|
||||
assert 'wrolpi.perpetual.perpetual_owner_watchdog' not in dispatched
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_perpetual_iteration_reschedules_after_error(monkeypatch):
|
||||
"""Ordinary exceptions are logged and the loop continues."""
|
||||
dispatched = []
|
||||
|
||||
async def boom():
|
||||
raise RuntimeError('worker exploded')
|
||||
|
||||
async def fake_dispatch(event):
|
||||
dispatched.append(event)
|
||||
|
||||
monkeypatch.setattr('wrolpi.api_utils.PYTEST', False)
|
||||
monkeypatch.setattr('wrolpi.api_utils._app_is_stopping', lambda: False)
|
||||
monkeypatch.setattr('wrolpi.api_utils._dispatch_perpetual', fake_dispatch)
|
||||
monkeypatch.setattr('wrolpi.api_utils.asyncio.sleep', mock.AsyncMock())
|
||||
|
||||
await _run_perpetual_iteration(boom, 'wrolpi.perpetual.test_worker', sleep=0)
|
||||
assert dispatched == ['wrolpi.perpetual.test_worker']
|
||||
async def test_stop_per_worker_tasks_without_start():
|
||||
"""Stopping an app that never started per-worker tasks (tests, or a startup failure) is harmless."""
|
||||
app = mock.Mock()
|
||||
app.ctx = mock.Mock(spec=[]) # no per_worker_* attributes
|
||||
await stop_per_worker_tasks(app)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
156
wrolpi/test/test_perpetual.py
Normal file
156
wrolpi/test/test_perpetual.py
Normal file
|
|
@ -0,0 +1,156 @@
|
|||
"""Tests for the perpetual process runner and the per-worker health check (wrolpi/perpetual.py)."""
|
||||
import asyncio
|
||||
import os
|
||||
import signal
|
||||
import threading
|
||||
import time
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from wrolpi.perpetual import (
|
||||
PERPETUAL_PROCESS_IDENT,
|
||||
PERPETUAL_RESTART_COOLDOWN,
|
||||
PerpetualLoop,
|
||||
perpetual_process_needs_restart,
|
||||
run_loop,
|
||||
run_loops,
|
||||
serve_perpetual_loops,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_loop_calls_sleeps_and_survives_errors():
|
||||
"""The loop keeps calling its function after an exception, and stops promptly when told to."""
|
||||
calls = []
|
||||
stop = asyncio.Event()
|
||||
|
||||
async def flaky():
|
||||
calls.append(time.time())
|
||||
if len(calls) == 2:
|
||||
raise RuntimeError('second call fails')
|
||||
if len(calls) == 4:
|
||||
stop.set()
|
||||
|
||||
await asyncio.wait_for(run_loop(PerpetualLoop('flaky', flaky, sleep=0.01), stop), timeout=2)
|
||||
|
||||
assert len(calls) == 4, 'the loop must continue after an error and stop when asked'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_loop_does_not_reenter_a_long_call():
|
||||
"""A slow iteration is awaited; the next call starts only after it returns."""
|
||||
active = 0
|
||||
overlaps = []
|
||||
stop = asyncio.Event()
|
||||
calls = 0
|
||||
|
||||
async def slow():
|
||||
nonlocal active, calls
|
||||
calls += 1
|
||||
active += 1
|
||||
overlaps.append(active)
|
||||
await asyncio.sleep(0.02)
|
||||
active -= 1
|
||||
if calls == 3:
|
||||
stop.set()
|
||||
|
||||
await asyncio.wait_for(run_loop(PerpetualLoop('slow', slow, sleep=0), stop), timeout=2)
|
||||
assert max(overlaps) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_loops_cancels_everything_on_stop():
|
||||
"""Setting `stop` cancels every loop task; a loop that ignores cancellation is abandoned, not waited for forever."""
|
||||
stop = asyncio.Event()
|
||||
cancelled = []
|
||||
|
||||
async def sleeper():
|
||||
try:
|
||||
await asyncio.sleep(3600)
|
||||
except asyncio.CancelledError:
|
||||
cancelled.append('sleeper')
|
||||
raise
|
||||
|
||||
async def stubborn():
|
||||
try:
|
||||
await asyncio.sleep(3600)
|
||||
except asyncio.CancelledError:
|
||||
cancelled.append('stubborn')
|
||||
await asyncio.sleep(3600) # refuses to die
|
||||
|
||||
loops = [PerpetualLoop('sleeper', sleeper, 1), PerpetualLoop('stubborn', stubborn, 1)]
|
||||
runner = asyncio.ensure_future(run_loops(loops, stop, cancel_timeout=0.1))
|
||||
await asyncio.sleep(0.01)
|
||||
stop.set()
|
||||
await asyncio.wait_for(runner, timeout=2)
|
||||
|
||||
assert sorted(cancelled) == ['sleeper', 'stubborn']
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_serve_perpetual_loops_stops_on_sigterm():
|
||||
"""The perpetual process exits when the manager sends SIGTERM."""
|
||||
ticks = []
|
||||
|
||||
async def tick():
|
||||
ticks.append(1)
|
||||
|
||||
initialized = []
|
||||
with mock.patch('wrolpi.perpetual._shutdown_singletons', mock.AsyncMock()) as shutdown:
|
||||
serving = asyncio.ensure_future(
|
||||
serve_perpetual_loops([PerpetualLoop('tick', tick, 0.01)], initialize=lambda: initialized.append(1)))
|
||||
await asyncio.sleep(0.05)
|
||||
assert ticks, 'loop should be running before the signal'
|
||||
# From another thread, as the manager (another process) would.
|
||||
threading.Timer(0.01, lambda: os.kill(os.getpid(), signal.SIGTERM)).start()
|
||||
await asyncio.wait_for(serving, timeout=2)
|
||||
|
||||
shutdown.assert_awaited_once()
|
||||
assert initialized == [1], 'per-process initialization must run once before the loops'
|
||||
|
||||
|
||||
def test_perpetual_process_needs_restart():
|
||||
"""Only a process the manager reports as gone triggers a restart request, at most once per cooldown."""
|
||||
now = 1_000_000.0
|
||||
dead = {PERPETUAL_PROCESS_IDENT: {'pid': 123, 'state': 'FAILED'}}
|
||||
alive = {PERPETUAL_PROCESS_IDENT: {'pid': 123, 'state': 'STARTED'}}
|
||||
|
||||
assert perpetual_process_needs_restart(dead, now, last_request=0) is True
|
||||
assert perpetual_process_needs_restart(dead, now, last_request=now - 1) is False, 'within cooldown'
|
||||
assert perpetual_process_needs_restart(dead, now, last_request=now - PERPETUAL_RESTART_COOLDOWN) is True
|
||||
for state in ('STARTING', 'STARTED', 'RESTARTING', 'NONE'):
|
||||
alive[PERPETUAL_PROCESS_IDENT]['state'] = state
|
||||
assert perpetual_process_needs_restart(alive, now, last_request=0) is False, state
|
||||
assert perpetual_process_needs_restart({}, now, last_request=0) is False, 'not registered yet is not dead'
|
||||
assert perpetual_process_needs_restart({'Sanic-Server-0': {'state': 'FAILED'}}, now, 0) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_health_check_requests_restart_once(async_client, monkeypatch):
|
||||
"""Every server worker runs this; a dead perpetual process gets one restart request per cooldown."""
|
||||
from wrolpi import perpetual
|
||||
from wrolpi.api_utils import api_app
|
||||
|
||||
multiplexer = mock.Mock()
|
||||
multiplexer.workers = {PERPETUAL_PROCESS_IDENT: {'pid': 1, 'state': 'FAILED'}}
|
||||
monkeypatch.setattr(api_app, 'multiplexer', multiplexer, raising=False)
|
||||
monkeypatch.setattr(perpetual, '_last_restart_request', 0.0)
|
||||
|
||||
await perpetual.perpetual_process_health_check()
|
||||
await perpetual.perpetual_process_health_check()
|
||||
multiplexer.restart.assert_called_once_with(PERPETUAL_PROCESS_IDENT)
|
||||
|
||||
multiplexer.workers = {PERPETUAL_PROCESS_IDENT: {'pid': 2, 'state': 'STARTED'}}
|
||||
monkeypatch.setattr(perpetual, '_last_restart_request', 0.0)
|
||||
await perpetual.perpetual_process_health_check()
|
||||
multiplexer.restart.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_health_check_without_multiplexer_is_a_noop(async_client, monkeypatch):
|
||||
from wrolpi import perpetual
|
||||
from wrolpi.api_utils import api_app
|
||||
|
||||
monkeypatch.delattr(api_app, 'multiplexer', raising=False)
|
||||
await perpetual.perpetual_process_health_check()
|
||||
|
|
@ -421,6 +421,29 @@ async def test_get_status(async_client, test_session):
|
|||
assert isinstance(response.json.get('local_time'), str), 'local_time should be a str'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_status_reports_perpetual_process(async_client, monkeypatch):
|
||||
"""/api/status lists the perpetual process next to the server workers, and nothing else the manager tracks."""
|
||||
from unittest import mock
|
||||
from wrolpi.api_utils import api_app
|
||||
|
||||
multiplexer = mock.Mock()
|
||||
multiplexer.workers = {
|
||||
'Sanic-Main': {'pid': 1},
|
||||
'Sanic-Server-0': {'pid': 2, 'state': 'ACKED'},
|
||||
'Sanic-Perpetual-0': {'pid': 3, 'state': 'STARTED'},
|
||||
'Sanic-Inspector-0': {'pid': 4, 'state': 'STARTED'},
|
||||
}
|
||||
monkeypatch.setattr(api_app, 'multiplexer', multiplexer, raising=False)
|
||||
|
||||
request, response = await async_client.get('/api/status')
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json['sanic_workers'] == {
|
||||
'Sanic-Server-0': {'pid': 2, 'state': 'ACKED'},
|
||||
'Sanic-Perpetual-0': {'pid': 3, 'state': 'STARTED'},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_status_local_time_with_timezone(async_client, test_session, test_wrolpi_config):
|
||||
"""Status endpoint includes local_time using configured timezone."""
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ from wrolpi.upgrade import (
|
|||
get_remote_commit,
|
||||
get_commits_behind,
|
||||
git_fetch,
|
||||
refresh_update_status,
|
||||
start_upgrade,
|
||||
)
|
||||
|
||||
|
|
@ -357,3 +358,58 @@ async def test_status_endpoint_includes_git_branch(async_client):
|
|||
finally:
|
||||
# Clean up
|
||||
api_app.shared_ctx.status.pop('git_branch', None)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_update_status_does_not_block_event_loop():
|
||||
"""`git fetch` must run off the event loop.
|
||||
|
||||
A synchronous fetch (up to its 60 s timeout) stalls every request and the
|
||||
perpetual-owner heartbeat on that worker; a stale heartbeat triggers a
|
||||
takeover by another worker.
|
||||
"""
|
||||
import asyncio
|
||||
import threading
|
||||
import time
|
||||
|
||||
seen_threads = []
|
||||
|
||||
def slow_check(fetch=True):
|
||||
seen_threads.append(threading.current_thread())
|
||||
time.sleep(0.3)
|
||||
return {'update_available': True, 'commits_behind': 3, 'branch': 'master'}
|
||||
|
||||
ticks = 0
|
||||
|
||||
async def ticker():
|
||||
nonlocal ticks
|
||||
while True:
|
||||
ticks += 1
|
||||
await asyncio.sleep(0.01)
|
||||
|
||||
status = {}
|
||||
with patch('wrolpi.upgrade.check_for_update', slow_check):
|
||||
ticker_task = asyncio.ensure_future(ticker())
|
||||
await refresh_update_status(status)
|
||||
ticker_task.cancel()
|
||||
|
||||
assert seen_threads and seen_threads[0] is not threading.main_thread(), \
|
||||
'check_for_update ran on the event loop thread'
|
||||
assert ticks >= 10, f'event loop was starved during check_for_update (only {ticks} ticks)'
|
||||
assert status['update_available'] is True and status['commits_behind'] == 3 and status['git_branch'] == 'master'
|
||||
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_update_status_can_skip_fetch():
|
||||
"""The upgrade-check endpoint passes fetch=False unless the user forces a fetch."""
|
||||
seen = []
|
||||
|
||||
def fake_check(fetch=True):
|
||||
seen.append(fetch)
|
||||
return {'update_available': False, 'branch': 'master'}
|
||||
|
||||
with patch('wrolpi.upgrade.check_for_update', fake_check):
|
||||
await refresh_update_status({}, fetch=False)
|
||||
await refresh_update_status({})
|
||||
assert seen == [False, True]
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue