diff --git a/CHANGELOG.md b/CHANGELOG.md index de0d1a2..ea6bbc7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Changelog +## 1.5.0 (2025-12-17) + +### Feat + +- **modules**: enhance repository CSV handling with caching and background refresh + +### Fix + +- **loop**: improve task handling in LoopManager for better coroutine management + +### Refactor + +- **worker**: simplify UUID generation logic in Worker class +- **tests**: streamline test_get_available_modules_does_not_block_on_remote + ## 1.4.0 (2025-12-05) ### Feat diff --git a/pyproject.toml b/pyproject.toml index 0526c94..03a54d9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "funcnodes-worker" -version = "1.4.1" +version = "1.5.0" description = "Worker package for FuncNodes" readme = "README.md" authors = [{name = "Julian Kimmig", email = "julian.kimmig@linkdlab.de"}] @@ -8,7 +8,7 @@ authors = [{name = "Julian Kimmig", email = "julian.kimmig@linkdlab.de"}] requires-python = ">=3.11" dependencies = [ "asynctoolkit>=0.1.1", - "funcnodes-core>=2.0.0", + "funcnodes-core>=2.2.0", "packaging>=24.2", "pip>=25.0.1", "pydantic>=2.0.0", diff --git a/src/funcnodes_worker/loop.py b/src/funcnodes_worker/loop.py index 2d9ae09..21ffa04 100644 --- a/src/funcnodes_worker/loop.py +++ b/src/funcnodes_worker/loop.py @@ -152,13 +152,20 @@ def remove_loop(self, loop: CustomLoop): self._loop.is_running() ) # and asyncio.get_event_loop() == self._loop + try: + running_loop = asyncio.get_running_loop() + except RuntimeError: + running_loop = None + if loop.running and not self._loop.is_closed(): loop._running = False # set this to false prevent recursion try: - if not is_running: - self._loop.run_until_complete(loop.stop()) - else: + if is_running: _ = self.async_call(loop.stop()) + elif running_loop is not None: + running_loop.create_task(loop.stop()) + else: + self._loop.run_until_complete(loop.stop()) except Exception as e: worker = self._worker() if worker is not None: @@ -198,18 +205,36 @@ async def _wait_cancel(t: asyncio.Task): task.cancel() waiter = _wait_cancel(task) - if not is_running and not self._loop.is_closed(): - try: - self._loop.run_until_complete(waiter) - except Exception as exc: # pragma: no cover - defensive - worker = self._worker() - if worker is not None: - worker.logger.exception(exc) + + try: + running_loop = asyncio.get_running_loop() + except RuntimeError: + running_loop = None + + task_loop = task.get_loop() + + if task_loop.is_running(): + if running_loop is task_loop: + task_loop.create_task(waiter) + else: + # task loop runs in another thread + asyncio.run_coroutine_threadsafe(waiter, task_loop) + return + + if not task_loop.is_closed(): + if running_loop is not None and running_loop is not task_loop: + # Avoid run_until_complete when another loop is already running + running_loop.create_task(waiter) + else: + try: + task_loop.run_until_complete(waiter) + except Exception as exc: # pragma: no cover - defensive + worker = self._worker() + if worker is not None: + worker.logger.exception(exc) else: - scheduled = self.async_call(waiter) - if scheduled is None: - # no loop to schedule on; close coroutine to avoid "never awaited" - waiter.close() + # loop is closed; close coroutine to silence warnings + waiter.close() def async_call(self, croutine: asyncio.Coroutine): # Check if the loop is closed or not running diff --git a/src/funcnodes_worker/utils/modules.py b/src/funcnodes_worker/utils/modules.py index b6a6720..328d9aa 100644 --- a/src/funcnodes_worker/utils/modules.py +++ b/src/funcnodes_worker/utils/modules.py @@ -2,11 +2,18 @@ import csv import importlib import importlib.metadata +from datetime import datetime, timezone +from pathlib import Path from typing import List, Optional, Dict from funcnodes_core import AVAILABLE_MODULES, setup, FUNCNODES_LOGGER from funcnodes_core._setup import setup_module from funcnodes_core.utils.plugins import InstalledModule +from funcnodes_core.utils.cache import ( + get_cache_path, + set_cache_meta_for, + write_cache_text, +) from dataclasses import dataclass, field import logging from asynctoolkit.defaults.http import HTTPTool @@ -89,41 +96,141 @@ def version_to_range(version: str) -> str: AVAILABLE_REPOS: Dict[str, AvailableRepo] = {} -async def load_repo_csv(): - """ - Load repository metadata from a remote CSV file and update AVAILABLE_REPOS. - """ +REPO_CSV_URL = "https://raw.githubusercontent.com/Linkdlab/funcnodes_repositories/refs/heads/main/funcnodes_modules.csv" - url = "https://raw.githubusercontent.com/Linkdlab/funcnodes_repositories/refs/heads/main/funcnodes_modules.csv" - try: - async with await HTTPTool().run(url=url) as resp: - text = await resp.text() - except Exception as e: - FUNCNODES_LOGGER.exception(e) - return +def get_cached_csv_path() -> Path: + return get_cache_path("funcnodes_modules.csv") + - # Read the CSV data into a dictionary reader +def _parse_repo_csv(text: str): + """Parse CSV text and update AVAILABLE_REPOS.""" reader = csv.DictReader(text.splitlines(), delimiter=",") for line in reader: try: - # Create an AvailableRepo instance from the CSV row data = AvailableRepo.from_dict(line) if data.package_name in AVAILABLE_REPOS: - # If there is existing module data, carry it over. moddata = AVAILABLE_REPOS[data.package_name].moduledata data.moduledata = moddata if data.moduledata: - # Mark as installed if module data is present. data.installed = True - # Update the global dictionary with this repository info AVAILABLE_REPOS[data.package_name] = data - except Exception as e: - # Log any exceptions that occur during parsing FUNCNODES_LOGGER.exception(e) +def load_cached_repo_csv() -> bool: + """ + Load repository data from local cache. + Returns True if cache was loaded successfully, False otherwise. + """ + cache_path = get_cached_csv_path() + if not cache_path.exists(): + return False + + try: + text = cache_path.read_text(encoding="utf-8") + _parse_repo_csv(text) + FUNCNODES_LOGGER.debug("Loaded repository data from cache") + return True + except Exception as e: + FUNCNODES_LOGGER.warning(f"Failed to load cached repo CSV: {e}") + return False + + +def save_repo_csv_to_cache(text: str): + """Save repository CSV data to local cache.""" + cache_path = get_cached_csv_path() + try: + write_cache_text(cache_path, text) + meta = { + "last_updated": datetime.now(timezone.utc) + .isoformat() + .replace("+00:00", "Z"), + "source_url": REPO_CSV_URL, + } + set_cache_meta_for(cache_path, meta) + FUNCNODES_LOGGER.debug("Saved repository data to cache") + except Exception as e: + FUNCNODES_LOGGER.warning(f"Failed to save repo CSV to cache: {e}") + + +_background_refresh_task: Optional[asyncio.Task] = None +_background_refresh_callbacks: List = [] + + +async def refresh_repo_csv_background(): + """ + Refresh repository CSV in the background. + + Any callbacks registered via start_background_repo_refresh will be invoked on success. + """ + global _background_refresh_task + global _background_refresh_callbacks + + callbacks: List = _background_refresh_callbacks + try: + async with await HTTPTool().run(url=REPO_CSV_URL) as resp: + text = await resp.text() + + _parse_repo_csv(text) + save_repo_csv_to_cache(text) + + except Exception as e: + FUNCNODES_LOGGER.debug(f"Background repo refresh failed: {e}") + callbacks = [] + finally: + _background_refresh_task = None + _background_refresh_callbacks = [] + + for cb in callbacks: + try: + res = cb(AVAILABLE_REPOS) + if asyncio.iscoroutine(res): + await res + except Exception as e: + FUNCNODES_LOGGER.debug(f"Repo refresh callback failed: {e}") + + +def start_background_repo_refresh(callback=None): + """Start a background task to refresh the repo CSV.""" + global _background_refresh_task + + if callback is not None: + _background_refresh_callbacks.append(callback) + + if _background_refresh_task is not None: + return _background_refresh_task + + _background_refresh_task = asyncio.create_task(refresh_repo_csv_background()) + return _background_refresh_task + + +async def load_repo_csv(use_cache: bool = True, update_cache: bool = True): + """ + Load repository metadata from a remote CSV file and update AVAILABLE_REPOS. + """ + cache_loaded = False + if use_cache: + cache_loaded = load_cached_repo_csv() + if cache_loaded and not update_cache: + return + + try: + async with await HTTPTool().run(url=REPO_CSV_URL) as resp: + text = await resp.text() + + _parse_repo_csv(text) + + if update_cache: + save_repo_csv_to_cache(text) + + except Exception as e: + FUNCNODES_LOGGER.exception(e) + if use_cache and not cache_loaded: + load_cached_repo_csv() + + async def install_package( package_name, version=None, @@ -328,7 +435,13 @@ def try_import_repo(name: str) -> Optional[AvailableRepo]: return None -async def reload_base(with_repos=True, retries=2, retries_delay=1): +async def reload_base( + with_repos: bool = True, + retries: int = 2, + retries_delay: float = 1, + background_repo_refresh: bool = False, + repo_refresh_callback=None, +): """ Reload the core setup and update repository/module information. @@ -349,11 +462,12 @@ async def reload_base(with_repos=True, retries=2, retries_delay=1): - Exception: If the core setup fails after all retries. """ # Initialize or refresh the core setup - retries = min(retries, 0) + retries = max(retries, 0) while retries >= 0: retries -= 1 try: setup() + break except Exception as e: if retries < 0: raise e @@ -361,10 +475,14 @@ async def reload_base(with_repos=True, retries=2, retries_delay=1): continue if with_repos: - try: - await load_repo_csv() - except Exception: - pass + if background_repo_refresh: + load_cached_repo_csv() + start_background_repo_refresh(callback=repo_refresh_callback) + else: + try: + await load_repo_csv(use_cache=True, update_cache=True) + except Exception: + pass # Update the 'installed' flag for each repo based on the presence of module data for repo in AVAILABLE_REPOS.values(): if repo.moduledata: diff --git a/src/funcnodes_worker/worker.py b/src/funcnodes_worker/worker.py index c704369..1496b81 100644 --- a/src/funcnodes_worker/worker.py +++ b/src/funcnodes_worker/worker.py @@ -621,13 +621,12 @@ def __init__( self.viewdata: ViewState = { "renderoptions": fn.config.FUNCNODES_RENDER_OPTIONS, } - self._uuid = ( - (slugify(name)[:16] + "_" + uuid4().hex[:16]) - if name - else uuid4().hex - if not uuid - else uuid - ) + if uuid: + self._uuid = uuid + elif name: + self._uuid = f"{slugify(name)[:16]}_{uuid4().hex[:16]}" + else: + self._uuid = uuid4().hex self._name = name or None self._WORKERS_DIR: Path = get_workers_dir() self._WORKER_DIR: Path = get_worker_dir(self.uuid()) @@ -867,7 +866,15 @@ async def ini_config(self): async def update_from_config(self, config: dict): """updates the worker from a config dict""" self.logger.debug("Update from config") - await reload_base(with_repos=True) + + async def on_repo_refresh(repos): + await self.worker_event("repos_update", repos=repos) + + await reload_base( + with_repos=True, + background_repo_refresh=True, + repo_refresh_callback=on_repo_refresh, + ) if "package_dependencies" in config: for name, dep in config["package_dependencies"].items(): try: @@ -1810,7 +1817,14 @@ async def remove_worker_dependency(self, src: WorkerDict): @exposed_method() async def get_available_modules(self): - await reload_base() + async def on_repo_refresh(repos): + await self.worker_event("repos_update", repos=repos) + + await reload_base( + with_repos=True, + background_repo_refresh=True, + repo_refresh_callback=on_repo_refresh, + ) ans = { "installed": [], "active": [], diff --git a/tests/test_repo_csv_cache.py b/tests/test_repo_csv_cache.py new file mode 100644 index 0000000..8e8435b --- /dev/null +++ b/tests/test_repo_csv_cache.py @@ -0,0 +1,163 @@ +import asyncio +import json + +import funcnodes_core as fn +from pytest_funcnodes import funcnodes_test + + +CSV_TEXT = """package_name,description,releases +funcnodes-randompackageßnamefortestA,Basic nodes,"0.1.0,0.2.0" +funcnodes-randompackageßnamefortestB,R2 nodes,"1.0.0" +""" + + +def _cache_csv_path(): + return fn.config.get_config_dir() / "cache" / "funcnodes_modules.csv" + + +@funcnodes_test +def test_load_cached_repo_csv_success(): + from funcnodes_worker.utils import modules as mod + + mod.AVAILABLE_REPOS.clear() + cache_path = _cache_csv_path() + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text(CSV_TEXT, encoding="utf-8") + + assert mod.load_cached_repo_csv() is True + assert "funcnodes-randompackageßnamefortestA" in mod.AVAILABLE_REPOS + assert ( + mod.AVAILABLE_REPOS["funcnodes-randompackageßnamefortestA"].description + == "Basic nodes" + ) + + +@funcnodes_test +def test_load_cached_repo_csv_missing_returns_false(): + from funcnodes_worker.utils import modules as mod + + mod.AVAILABLE_REPOS.clear() + assert mod.load_cached_repo_csv() is False + + +@funcnodes_test +async def test_load_repo_csv_falls_back_to_cache_on_remote_failure(monkeypatch): + from funcnodes_worker.utils import modules as mod + + mod.AVAILABLE_REPOS.clear() + cache_path = _cache_csv_path() + cache_path.parent.mkdir(parents=True, exist_ok=True) + cache_path.write_text(CSV_TEXT, encoding="utf-8") + + async def failing_run(self, url=None, **kwargs): + raise RuntimeError("network down") + + monkeypatch.setattr(mod.HTTPTool, "run", failing_run) + + await mod.load_repo_csv(use_cache=True, update_cache=True) + assert "funcnodes-randompackageßnamefortestB" in mod.AVAILABLE_REPOS + + +@funcnodes_test +def test_save_repo_csv_to_cache_writes_meta(): + from funcnodes_worker.utils import modules as mod + + mod.AVAILABLE_REPOS.clear() + mod.save_repo_csv_to_cache(CSV_TEXT) + + cache_path = _cache_csv_path() + meta_path = cache_path.with_suffix(cache_path.suffix + ".meta.json") + + assert cache_path.exists() + assert meta_path.exists() + meta = json.loads(meta_path.read_text(encoding="utf-8")) + assert meta["source_url"].startswith("https://") + assert "last_updated" in meta + + +@funcnodes_test +async def test_background_refresh_accumulates_callbacks(monkeypatch): + from funcnodes_worker.utils import modules as mod + + mod.AVAILABLE_REPOS.clear() + + class _FakeResp: + async def text(self): + return CSV_TEXT + + class _FakeCtx: + async def __aenter__(self): + return _FakeResp() + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def fake_run(self, url=None, **kwargs): + return _FakeCtx() + + monkeypatch.setattr(mod.HTTPTool, "run", fake_run) + + called = [] + + async def cb1(repos): + called.append("cb1") + + async def cb2(repos): + called.append("cb2") + + task1 = mod.start_background_repo_refresh(callback=cb1) + task2 = mod.start_background_repo_refresh(callback=cb2) + assert task1 is task2 + + await task1 + assert set(called) == {"cb1", "cb2"} + + +@funcnodes_test +async def test_get_available_modules_does_not_block_on_remote(monkeypatch): + from funcnodes_worker import Worker + from funcnodes_worker.utils import modules as mod + + class _SlowResp: + async def text(self): + await asyncio.sleep(0.5) + return CSV_TEXT + + class _SlowCtx: + async def __aenter__(self): + return _SlowResp() + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def slow_run(self, url=None, **kwargs): + return _SlowCtx() + + monkeypatch.setattr(mod.HTTPTool, "run", slow_run) + # setup() can be slow/unrelated here; avoid blocking the loop for this timing test + monkeypatch.setattr(mod, "setup", lambda: None) + + class _TestWorker(Worker): + def _on_nodespaceerror(self, error, src): + return None + + def on_nodespaceevent(self, event, **kwargs): + return None + + worker = _TestWorker(uuid="TestWorker_nonblocking") + mod.AVAILABLE_REPOS.clear() + cache_path = fn.config.get_config_dir() / "cache" / "funcnodes_modules.csv" + if cache_path.exists(): + cache_path.unlink() + + res = await worker.get_available_modules() + assert res is not None + assert len(res["available"]) + len(res["active"]) == 0 + assert mod._background_refresh_task is not None + + await asyncio.sleep(1) + assert mod._background_refresh_task is None + assert len(mod._background_refresh_callbacks) == 0 + res = await worker.get_available_modules() + assert res is not None + assert len(res["available"]) + len(res["active"]) == 2 diff --git a/uv.lock b/uv.lock index 2babca6..0e639ac 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.11" resolution-markers = [ "sys_platform != 'emscripten'", @@ -429,7 +429,7 @@ wheels = [ [[package]] name = "funcnodes-core" -version = "2.0.0" +version = "2.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "dill" }, @@ -439,9 +439,8 @@ dependencies = [ { name = "setuptools" }, { name = "tqdm" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/67/44/6ddee77de599ede9fe598106bec950790838103370fe812efe0548587d6b/funcnodes_core-2.0.0.tar.gz", hash = "sha256:73d5a9731b2d30ad98fe9c47d809de8a324dafcdbbc3701c86194c0d1b535b52", size = 190475, upload-time = "2025-11-27T18:51:11.954Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/49/bf/5f6652700e98a856e31bd80b608ca89c8f2b558c2a40c980c3cb39eefc16/funcnodes_core-2.0.0-py3-none-any.whl", hash = "sha256:6e42ef5065162adf09f3ffe2fff0fc5d1397235e98c670bdc8bf031bd0883e8d", size = 91624, upload-time = "2025-11-27T18:51:10.504Z" }, + { url = "https://files.pythonhosted.org/packages/d4/ff/5f76caf509c614dde0f453cad55e4b890888618816a1a1e3b6f2b34e6c44/funcnodes_core-2.2.0-py3-none-any.whl", hash = "sha256:1a1d839dfe70f1c8ff2ba55abf6fc3c7875cd3de430432d2dc33b51e0c7a2ea1", size = 92765, upload-time = "2025-12-12T12:58:28.801Z" }, ] [[package]] @@ -474,7 +473,7 @@ wheels = [ [[package]] name = "funcnodes-worker" -version = "1.4.1" +version = "1.5.0" source = { editable = "." } dependencies = [ { name = "asynctoolkit" }, @@ -524,7 +523,7 @@ requires-dist = [ { name = "aiohttp", marker = "extra == 'http'" }, { name = "aiohttp-cors", marker = "extra == 'http'" }, { name = "asynctoolkit", specifier = ">=0.1.1" }, - { name = "funcnodes-core", specifier = ">=2.0.0" }, + { name = "funcnodes-core", specifier = ">=2.2.0" }, { name = "funcnodes-worker", extras = ["venv", "http", "subprocess-monitor"], marker = "extra == 'all'" }, { name = "packaging", specifier = ">=24.2" }, { name = "pip", specifier = ">=25.0.1" },