diff --git a/wrolpi/downloader.py b/wrolpi/downloader.py index b6e937a6..36937f17 100644 --- a/wrolpi/downloader.py +++ b/wrolpi/downloader.py @@ -998,8 +998,12 @@ class DownloadManager: session.flush(downloads) try: - # Start downloading ASAP. - background_task(self.dispatch_downloads()) + from wrolpi import perpetual + if perpetual.IN_PERPETUAL_PROCESS or PYTEST: + # Start downloading ASAP. Only the perpetual process runs downloads; a request worker leaves + # them for perpetual_download_worker's next cycle (a few seconds) rather than running a download + # inside the API worker. + background_task(self.dispatch_downloads()) # Save the config now that new Downloads exist. save_downloads_config.activate_switch() except RuntimeError: diff --git a/wrolpi/perpetual.py b/wrolpi/perpetual.py index d5cfbaeb..213c8fd9 100644 --- a/wrolpi/perpetual.py +++ b/wrolpi/perpetual.py @@ -36,8 +36,10 @@ 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 +# One restart request per death across all workers. The cooldown only has to outlast the manager acting on the +# request (the replacement appears within a second or two); a longer one leaves the loops dead that much longer +# when the replacement dies too. +PERPETUAL_RESTART_COOLDOWN = 15 # 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 @@ -45,6 +47,15 @@ PERPETUAL_CANCEL_TIMEOUT = 15 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 +# Shared flags that mean "the perpetual process is in the middle of this". They are set and cleared by context +# managers in this process only; a SIGKILL inside one leaves it set, and the next attempt raises "flag is already +# set" forever (every file refresh, for instance). +PERPETUAL_PROGRESS_FLAGS = ('file_worker_busy', 'file_worker_counting', 'file_worker_discovery', + 'file_worker_modeling', 'file_worker_indexing', 'file_worker_cleanup', + 'global_refresh_active') + +# True only inside the process started by `run_perpetual_process`. +IN_PERPETUAL_PROCESS = False @dataclass @@ -157,14 +168,26 @@ def reconcile_after_predecessor(app) -> dict: 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 import flags from wrolpi.downloader import download_manager, Download, DownloadStatus from wrolpi.files.worker import file_worker from wrolpi.db import get_db_session - report = dict(locks=[], jobs=0, downloads=0, processing_domains=0, file_worker_jobs=0) + report = dict(locks=[], flags=[], jobs=0, downloads=0, processing_domains=0, file_worker_jobs=0, + file_worker_status=False) report['locks'] = heal_shared_locks(app) + for name in PERPETUAL_PROGRESS_FLAGS: + try: + flag = getattr(flags, name) + if flag.is_set(): + logger.warning(f'reconcile: clearing in-progress flag {name} left set by a dead process') + flag.clear() + report['flags'].append(name) + except Exception as e: + logger.error(f'reconcile: failed to clear flag {name}', exc_info=e) + try: from wrolpi.jobs import fail_orphaned_jobs report['jobs'] = fail_orphaned_jobs() @@ -200,6 +223,7 @@ def reconcile_after_predecessor(app) -> dict: 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() + report['file_worker_status'] = True except Exception as e: logger.error('reconcile: failed to reset the file worker status', exc_info=e) @@ -239,6 +263,8 @@ def run_perpetual_process(): 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. + global IN_PERPETUAL_PROCESS + IN_PERPETUAL_PROCESS = True signal.signal(signal.SIGTERM, signal.SIG_DFL) signal.signal(signal.SIGINT, signal.SIG_DFL) if not PERPETUAL_LOOPS: diff --git a/wrolpi/test/test_downloader.py b/wrolpi/test/test_downloader.py index 9abecc50..00810417 100644 --- a/wrolpi/test/test_downloader.py +++ b/wrolpi/test/test_downloader.py @@ -2158,3 +2158,22 @@ async def test_renew_recurring_downloads_scans_without_the_write_lock(test_sessi assert lock_held_during_scan, 'the recurring downloads were never scanned' assert not any(lock_held_during_scan), \ 'the write lock was held while scanning recurring downloads' + + +@pytest.mark.asyncio +async def test_create_downloads_does_not_dispatch_outside_the_perpetual_process(test_session, test_download_manager, + test_downloader, monkeypatch): + """Only the perpetual process runs downloads. A request worker creating Downloads must not start one itself; + perpetual_download_worker picks them up on its next cycle.""" + from wrolpi import perpetual + monkeypatch.setattr('wrolpi.downloader.PYTEST', False) + monkeypatch.setattr(perpetual, 'IN_PERPETUAL_PROCESS', False) + with mock.patch.object(test_download_manager, 'dispatch_downloads') as dispatch: + test_download_manager.create_downloads(test_session, ['https://example.com/worker'], test_downloader.name) + dispatch.assert_not_called() + + monkeypatch.setattr(perpetual, 'IN_PERPETUAL_PROCESS', True) + with mock.patch.object(test_download_manager, 'dispatch_downloads', new_callable=mock.AsyncMock) as dispatch: + test_download_manager.create_downloads(test_session, ['https://example.com/perpetual'], test_downloader.name) + await asyncio.sleep(0) + dispatch.assert_called_once() diff --git a/wrolpi/test/test_perpetual.py b/wrolpi/test/test_perpetual.py index 423e9ce9..95469d03 100644 --- a/wrolpi/test/test_perpetual.py +++ b/wrolpi/test/test_perpetual.py @@ -188,9 +188,9 @@ async def test_health_check_without_multiplexer_is_a_noop(async_client, monkeypa await perpetual.perpetual_process_health_check() -def _wait_for_byte(fd: int, timeout: float) -> bool: +def _wait_for_byte(fd: int, timeout: float) -> bytes: ready, _, _ = select.select([fd], [], [], timeout) - return bool(ready) and bool(os.read(fd, 1)) + return os.read(fd, 1) if ready else b'' @pytest.mark.parametrize('sig', [signal.SIGINT, signal.SIGTERM], ids=['SIGINT (manager shutdown)', 'SIGTERM (restart)']) @@ -202,16 +202,18 @@ def test_run_perpetual_process_exits_cleanly_on_signal(async_client, monkeypatch async def loop_with_live_background_task(): from wrolpi.common import background_task background_task(asyncio.sleep(3600)) - os.write(ready_w, b'x') + # The byte doubles as the assertion that the entry point marked this process as the perpetual one. + os.write(ready_w, b'x' if perpetual.IN_PERPETUAL_PROCESS else b'n') monkeypatch.setattr(perpetual, 'PERPETUAL_LOOPS', [PerpetualLoop('bg', loop_with_live_background_task, 3600)]) + assert perpetual.IN_PERPETUAL_PROCESS is False, 'the test process is not the perpetual process' # 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' + assert _wait_for_byte(ready_r, 10) == b'x', 'the perpetual process did not start its loop as the perpetual process' os.kill(process.pid, sig) process.join(15) assert process.exitcode == 0, f'perpetual process exited {process.exitcode}' @@ -293,12 +295,12 @@ def test_sanic_manager_marks_failed_and_restart_revives(test_directory): @pytest.mark.asyncio -async def test_reconcile_after_predecessor(async_client, test_session, monkeypatch): +async def test_reconcile_after_predecessor(async_client, test_session, monkeypatch, flags_lock): """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 + from wrolpi import jobs, flags # 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. @@ -312,6 +314,9 @@ async def test_reconcile_after_predecessor(async_client, test_session, monkeypat created_at='', started_at='', finished_at=None, error=None) file_worker._set_job_status('fj', 'running') file_worker.update_status(status='refreshing') + flags.file_worker_counting.set() + flags.global_refresh_active.set() + flags.db_up.set() # not an in-progress flag; must survive assert api_app.shared_ctx.jobs_lock.acquire(timeout=1) monkeypatch.setattr(perpetual, 'SHARED_LOCK_HEAL_TIMEOUT', 0.1) @@ -329,7 +334,12 @@ async def test_reconcile_after_predecessor(async_client, test_session, monkeypat 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' + assert file_worker.status['status'] == 'idle' and report['file_worker_status'] is True + assert report['flags'] == ['file_worker_counting', 'global_refresh_active'] + assert not flags.file_worker_counting.is_set() and not flags.global_refresh_active.is_set() + assert flags.db_up.is_set(), 'only in-progress flags are cleared' + with flags.file_worker_counting: # usable again; this raised "flag is already set" before + pass # Idempotent: a clean state reports nothing. assert not any(reconcile_after_predecessor(api_app).values())