mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
startup: run database config imports off the worker event loop (#723)
* startup: run database config imports off the worker event loop import_all_db_configs runs in a Sanic worker's after_server_start listener, after that worker's socket is already accepting connections. Every import in it was a synchronous call on the event loop, so a large tags/channels/domains/downloads config froze that worker for the whole import and every request the kernel routed to it hung. With the default worker count equal to CPU count, a one- or two-core host lost all or half of the API for minutes on startup, and the frontend's 60s request limit turned that into repeated silent retries. Each synchronous import now runs in asyncio.to_thread. The SQLite engine already allows cross-thread connections (check_same_thread off, NullPool) and other startup work already uses to_thread. Imports still run one at a time in the documented order; only the loop is no longer blocked. Test: a ticker task must keep ticking while two deliberately slow synchronous imports run. It scored zero ticks before this change. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> * startup: thread the downloads import too; test covers every import import_downloads_config is async but never awaited anything: its body called the synchronous DownloadManagerConfig.import_config on the event loop, so a large download_manager.yaml still froze the worker. The db_up check stays on the loop and only the import moves to a thread. The responsiveness test now patches every import the startup path performs, including that sync downloads import, with the same blocking sleep, and requires ~90% of the total sleep budget in ticks so any one import returning to the loop fails it. The ticker is awaited after cancel so the task does not outlive the loop. Docstring trimmed to the one constraint not visible at the call site. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
parent
92fa3f785f
commit
bcd12c9b96
3 changed files with 69 additions and 8 deletions
|
|
@ -1183,6 +1183,9 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
Returns a dict mapping config name to success status.
|
||||
Used by main.py startup and disaster recovery tests.
|
||||
|
||||
This runs from a worker's after_server_start while the worker is already accepting requests,
|
||||
so no import may block the event loop.
|
||||
|
||||
Import order matters:
|
||||
1. Tags - no dependencies
|
||||
2. Downloads - no dependencies
|
||||
|
|
@ -1200,7 +1203,7 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
|
||||
# Tags first (channels/domains depend on tags existing)
|
||||
try:
|
||||
tags.import_tags_config()
|
||||
await asyncio.to_thread(tags.import_tags_config)
|
||||
results['tags'] = True
|
||||
logger.debug('tags config imported')
|
||||
except Exception as e:
|
||||
|
|
@ -1218,7 +1221,7 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
|
||||
# Channels (uses tags, links to downloads)
|
||||
try:
|
||||
import_channels_config()
|
||||
await asyncio.to_thread(import_channels_config)
|
||||
results['channels'] = True
|
||||
logger.debug('channels config imported')
|
||||
except Exception as e:
|
||||
|
|
@ -1227,7 +1230,7 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
|
||||
# Domains (uses tags)
|
||||
try:
|
||||
import_domains_config()
|
||||
await asyncio.to_thread(import_domains_config)
|
||||
results['domains'] = True
|
||||
logger.debug('domains config imported')
|
||||
except Exception as e:
|
||||
|
|
@ -1236,7 +1239,7 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
|
||||
# Inventories (config-only, no DB). Migrates any legacy inventories.yaml, then loads per-inventory files.
|
||||
try:
|
||||
import_inventories_config()
|
||||
await asyncio.to_thread(import_inventories_config)
|
||||
results['inventories'] = True
|
||||
logger.debug('inventories config imported')
|
||||
except Exception as e:
|
||||
|
|
@ -1249,7 +1252,7 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
# user-curated with no other config home, so they own playlists.yaml.
|
||||
try:
|
||||
from wrolpi.collections.config import playlists_config
|
||||
playlists_config.import_config()
|
||||
await asyncio.to_thread(playlists_config.import_config)
|
||||
results['playlists'] = True
|
||||
logger.debug('playlists config imported')
|
||||
except Exception as e:
|
||||
|
|
@ -1259,7 +1262,7 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
# Map pins (YAML-only, no DB)
|
||||
try:
|
||||
from modules.map.pins import get_map_pins_config
|
||||
get_map_pins_config().import_config()
|
||||
await asyncio.to_thread(get_map_pins_config().import_config)
|
||||
results['map_pins'] = True
|
||||
logger.debug('map pins config imported')
|
||||
except Exception as e:
|
||||
|
|
@ -1269,7 +1272,7 @@ async def import_all_db_configs() -> dict[str, bool]:
|
|||
# Flasher saved firmware configurations (YAML-only, no DB)
|
||||
try:
|
||||
from modules.flasher.config import get_flasher_config
|
||||
get_flasher_config().import_config()
|
||||
await asyncio.to_thread(get_flasher_config().import_config)
|
||||
results['flasher'] = True
|
||||
logger.debug('flasher config imported')
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -2381,7 +2381,9 @@ async def import_downloads_config():
|
|||
logger.warning(f'Refusing to import downloads config when DB is not up.')
|
||||
return
|
||||
|
||||
get_download_manager_config().import_config()
|
||||
# The import upserts every download on the calling thread; keep it off the event loop (this runs
|
||||
# from a worker's after_server_start while the worker is already accepting requests).
|
||||
await asyncio.to_thread(get_download_manager_config().import_config)
|
||||
|
||||
|
||||
def parse_feed(url: str) -> FeedParserDict:
|
||||
|
|
|
|||
|
|
@ -19,11 +19,14 @@ Import Order (from import_all_db_configs):
|
|||
4. Domains
|
||||
5. Inventories
|
||||
"""
|
||||
import asyncio
|
||||
import contextlib
|
||||
import pathlib
|
||||
import shutil
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
from unittest import mock
|
||||
|
||||
from modules.archive.lib import get_domains_config
|
||||
from modules.videos.lib import get_channels_config
|
||||
|
|
@ -576,3 +579,56 @@ class TestMissingTagsConfig:
|
|||
assert domains[0].name == 'tagged-domain.com'
|
||||
# Tag reference should resolve to None
|
||||
assert domains[0].tag is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_import_all_db_configs_does_not_block_event_loop(test_session, test_directory):
|
||||
"""A slow config import must not freeze the API worker that runs startup.
|
||||
|
||||
Large tags/channels/downloads configs can take minutes to import. They run in a Sanic worker's
|
||||
after_server_start listener while that worker is already accepting connections, so a blocking
|
||||
import makes every request routed to that worker hang. The imports must run off the loop."""
|
||||
import time
|
||||
|
||||
sleep_seconds = 0.2
|
||||
|
||||
def slow_import(*args, **kwargs):
|
||||
time.sleep(sleep_seconds) # deliberately synchronous, like a real large import
|
||||
|
||||
# Every import import_all_db_configs performs, including the sync import behind the async
|
||||
# downloads wrapper. Each is replaced with the same blocking sleep.
|
||||
patched = [
|
||||
'wrolpi.tags.import_tags_config',
|
||||
'wrolpi.downloader.DownloadManagerConfig.import_config',
|
||||
'modules.videos.lib.import_channels_config',
|
||||
'modules.archive.lib.import_domains_config',
|
||||
'modules.inventory.common.import_inventories_config',
|
||||
'wrolpi.collections.config.PlaylistsConfig.import_config',
|
||||
'modules.map.pins.MapPinsConfig.import_config',
|
||||
'modules.flasher.config.FlasherConfig.import_config',
|
||||
]
|
||||
|
||||
ticks = 0
|
||||
tick_seconds = 0.05
|
||||
|
||||
async def ticker():
|
||||
nonlocal ticks
|
||||
while True:
|
||||
await asyncio.sleep(tick_seconds)
|
||||
ticks += 1
|
||||
|
||||
ticker_task = asyncio.create_task(ticker())
|
||||
try:
|
||||
with contextlib.ExitStack() as stack:
|
||||
for target in patched:
|
||||
stack.enter_context(mock.patch(target, slow_import))
|
||||
await import_all_db_configs()
|
||||
finally:
|
||||
ticker_task.cancel()
|
||||
with contextlib.suppress(asyncio.CancelledError):
|
||||
await ticker_task
|
||||
|
||||
# A responsive loop ticks through the whole sleep budget; require most of it so that any one
|
||||
# import moving back onto the loop (losing its share of ticks) fails.
|
||||
expected = len(patched) * sleep_seconds / tick_seconds
|
||||
assert ticks >= expected * 0.9, f'event loop was blocked during import ({ticks=}, {expected=})'
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue