diff --git a/.github/workflows/lint-openbb-platform.yml b/.github/workflows/lint-openbb-platform.yml index 8bbd160f6de8..b01ca087baa0 100644 --- a/.github/workflows/lint-openbb-platform.yml +++ b/.github/workflows/lint-openbb-platform.yml @@ -6,15 +6,6 @@ env: on: pull_request: types: [opened, synchronize, edited] - paths: - - 'openbb_platform/**' - - '!openbb_platform/api/**' - - '!openbb_platform/extensions/**' - - '!openbb_platform/providers/**' - - '!openbb_platform/obbject_extensions/**' - - '.github/workflows/lint-openbb-platform.yml' - - '.github/codeql/**' - - '.github/workflows/codeql.yml' merge_group: types: [checks_requested] @@ -23,8 +14,46 @@ concurrency: cancel-in-progress: true jobs: + changes: + name: Detect Changes + runs-on: ubuntu-latest + permissions: + pull-requests: read + outputs: + platform: ${{ steps.filter.outputs.platform }} + steps: + - name: Check for shared platform changes + id: filter + env: + GH_TOKEN: ${{ github.token }} + EVENT_NAME: ${{ github.event_name }} + REPO: ${{ github.repository }} + PR_NUMBER: ${{ github.event.pull_request.number }} + run: | + # Path filtering lives here instead of `on.pull_request.paths`: a workflow + # skipped by a trigger filter never reports its checks, which leaves the + # required "General Code Linting" and "Markdown Linting" checks pending + # forever. Jobs skipped by a job-level `if:` report success instead. + if [ "$EVENT_NAME" != "pull_request" ]; then + echo "platform=true" >> "$GITHUB_OUTPUT" + exit 0 + fi + gh api --paginate "repos/${REPO}/pulls/${PR_NUMBER}/files" --jq '.[].filename' > changed_files.txt + if awk ' + /^\.github\/(workflows\/(lint-openbb-platform|codeql)\.yml$|codeql\/)/ { found = 1 } + /^openbb_platform\// && !/^openbb_platform\/(api|extensions|providers|obbject_extensions)\// { found = 1 } + END { exit !found } + ' changed_files.txt; then + echo "platform=true" >> "$GITHUB_OUTPUT" + else + echo "platform=false" >> "$GITHUB_OUTPUT" + fi + code-linting: name: General Code Linting + needs: changes + # Fail closed: run unless change detection explicitly found nothing relevant. + if: ${{ !cancelled() && needs.changes.outputs.platform != 'false' }} runs-on: ubuntu-latest env: UV_CACHE_DIR: ${{ github.workspace }}/.uv-cache @@ -122,6 +151,8 @@ jobs: markdown-link-check: name: Markdown Linting + needs: changes + if: ${{ !cancelled() && needs.changes.outputs.platform != 'false' }} runs-on: ubuntu-latest permissions: contents: read @@ -136,6 +167,8 @@ jobs: json-yaml-validate: name: JSON Check + needs: changes + if: ${{ !cancelled() && needs.changes.outputs.platform != 'false' }} runs-on: ubuntu-latest permissions: contents: read diff --git a/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py b/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py index ed7f830131a7..f537ce8db687 100644 --- a/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py +++ b/openbb_platform/providers/cboe/openbb_cboe/utils/helpers.py @@ -3,6 +3,10 @@ from __future__ import annotations import logging +import os +import sqlite3 +import threading +from contextlib import asynccontextmanager from datetime import ( date as dateType, datetime, @@ -149,10 +153,17 @@ async def response_callback(response, _) -> Any: return await response.read() -_cache: Any = None +# Every cached request shares one SQLite connection, which lives on an event loop +# of its own (see ``_get_cache_loop``). ``_cache_lock`` guards the state below that +# threads outside that loop also touch; ``_stores`` and ``_swept`` belong to it, and +# ``_cache_guard`` lets one thing at a time open, use or close the connection. +_cache_lock = threading.Lock() _cache_loop: Any = None -_swept = False +_cache_guard: Any = None _cache_users = 0 +_stores: Any = None +_swept = False +_inherited: list = [] def cache_path() -> str: @@ -175,6 +186,30 @@ async def _oldest(backend: Any, count: int) -> set: return keys +async def _pragma(connection: Any, statement: str) -> Any: + """Run a PRAGMA and return its row, leaving no statement open behind it.""" + cursor = await connection.execute(statement) + + try: + return await cursor.fetchone() + finally: + await cursor.close() + + +async def _size(backend: Any) -> int: + """Return how big the cache file is, counting what is still in its log. + + The file on disk cannot say. After a ``VACUUM`` in write-ahead mode the new + pages wait in the log until it is checkpointed, and the file keeps its old + size until then. + """ + async with backend.responses.get_connection() as connection: + (pages,) = await _pragma(connection, "PRAGMA page_count") + (page_size,) = await _pragma(connection, "PRAGMA page_size") + + return pages * page_size + + async def _sweep(backend: Any) -> None: """Drop what has expired, then the oldest of what is left if still too big. @@ -184,18 +219,17 @@ async def _sweep(backend: Any) -> None: of the file that overshoots it, which is one rewrite of the file rather than one per entry. """ - from pathlib import Path - await backend.delete_expired_responses() await _vacuum(backend) - stored = Path(f"{cache_path()}.sqlite") for _ in range(CACHE_SWEEP_PASSES): - if not stored.exists() or stored.stat().st_size <= CACHE_MAX_BYTES: + size = await _size(backend) + + if size <= CACHE_MAX_BYTES: return held = await backend.responses.size() - overshoot = 1 - CACHE_MAX_BYTES / stored.stat().st_size + overshoot = 1 - CACHE_MAX_BYTES / size dropped = await _oldest(backend, max(1, round(held * overshoot))) if not dropped: @@ -206,74 +240,297 @@ async def _sweep(backend: Any) -> None: async def _vacuum(backend: Any) -> None: - """Give back the pages the deleted responses were held in.""" + """Give back the pages the deleted responses were held in, if there are any. + + A ``VACUUM`` rewrites the whole file while holding it, and every process + sweeps as it starts, so one that has nothing to give back must not run it. + """ async with backend.responses.get_connection(commit=True) as connection: + (unused,) = await _pragma(connection, "PRAGMA freelist_count") + + if not unused: + return + await connection.execute("VACUUM") + # In write-ahead mode the vacuumed file reaches the disk, and gives its + # space back, only when the log is checkpointed. + await _pragma(connection, "PRAGMA wal_checkpoint(TRUNCATE)") -async def close_cache_backend(backend: Any) -> None: - """Close the shared cache and forget it, so the next caller builds a fresh one. +async def _use_wal(stores: Any) -> None: + """Put the cache file in write-ahead logging, where readers and a writer coexist. - ``SQLiteBackend.close()`` clears the connection object the backend was - constructed with and never rebuilds it, so a closed backend is permanently - unusable - every later request through it dies on ``NoneType has no - attribute '_connection'``. Dropping the module reference is what makes the - close recoverable. + Otherwise a process writing to the file makes every other process wait to + read it, and a process reading it makes the writer wait to commit; one that + holds the file for longer than SQLite will wait makes the request fail with + ``database is locked``. The mode belongs to the file, so it holds for every + process that opens it from now on. A file that cannot be switched - another + process is part way through a write, or the file system cannot keep the shared + memory the mode needs - keeps working in its default mode. Parameters ---------- - backend : Any - The backend to close; ignored if it is no longer the shared one. + stores : SQLiteBackend + The stores whose file is switched. + """ + try: + async with stores.responses.get_connection() as connection: + row = await _pragma(connection, "PRAGMA journal_mode=WAL") + except Exception: # noqa: BLE001 + _logger.exception("Could not put the Cboe response cache in WAL mode") + + return + + if not row or str(row[0]).lower() != "wal": + _logger.warning("The Cboe response cache is staying in %s mode", row) + + +class _CacheConnection(sqlite3.Connection): + """A connection set up for a cache that many processes share. + + It commits each statement as it runs. By default a write opens a transaction, + takes the file's write lock, and only gives it up when a second call, made + after the event loop has had its turn, commits it. Other processes wait out + that gap, however long a busy machine makes it, and a write cancelled inside + it leaves the lock held until something else commits. + + It does not wait for the disk after each commit either (``synchronous`` is + ``NORMAL``). A commit that waits holds the write lock for as long as the disk + takes, which on a slow one is long enough for every other process to give up + on the lock; what the wait buys is surviving a power cut, and a response that + is lost is fetched again. + + Both have to be set by the thread the connection lives on, and the library + drops ``isolation_level=None`` along with every other ``None`` argument, so + they are set here, where the library hands a connection ``factory`` its + arguments. """ - global _cache, _cache_loop # noqa: PLW0603 - if _cache is backend: - _cache = None - _cache_loop = None + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.isolation_level = None + self.execute("PRAGMA synchronous = NORMAL") + + +def _new_backend() -> Any: + """Build the SQLite stores of the response cache.""" + from aiohttp_client_cache import SQLiteBackend + + return SQLiteBackend(cache_path(), autoclose=False, factory=_CacheConnection) + + +async def _sweep_file() -> None: + """Sweep the cache file on a connection of its own, closed when it is done. + + The sweep can stop part way - on a row that cannot be read back, say - with a + read still open. It runs before the requests' connection exists, and on a + connection that is closed whatever happens, so nothing of it is left holding + the file once it is over. + """ + backend = _new_backend() try: - await backend.close() + await _sweep(backend) except Exception: # noqa: BLE001 - _logger.exception("Closing the Cboe response cache failed") + _logger.exception("Sweeping the Cboe response cache failed") + finally: + try: + await backend.close() + except Exception: # noqa: BLE001 + _logger.exception("Closing the Cboe sweep connection failed") + + +def _get_cache_loop() -> Any: + """Return the event loop the cache's connection lives on, starting it if needed. + + An aiosqlite connection, and the lock beside it, only work on the loop that + first used them. Requests run on whatever loop their caller has - core's + ``run_async`` gives every synchronous call a thread and a loop of its own - + so a connection per loop would be one writer per loop on a single file, and + one connection shared between loops would fail with ``bound to a different + event loop`` or wait on a wake-up that never reaches its loop. One loop, on a + thread of its own, owns the single connection, and requests reach it from + wherever they run. + + Returns + ------- + AbstractEventLoop + The loop, running on a daemon thread, that every cache operation uses. + """ + global _cache_loop, _cache_guard # noqa: PLW0603 + + from asyncio import Lock, new_event_loop + + with _cache_lock: + if _cache_loop is None: + loop = new_event_loop() + threading.Thread( + target=loop.run_forever, name="cboe-cache", daemon=True + ).start() + _cache_guard = Lock() + _cache_loop = loop + + return _cache_loop + + +async def _on_cache_loop(operation: Any) -> Any: + """Run a coroutine on the cache loop and wait for it from the running one.""" + from asyncio import run_coroutine_threadsafe, wrap_future + + return await wrap_future(run_coroutine_threadsafe(operation, _get_cache_loop())) + + +async def _store_call(table: str, method: str, *args: Any) -> Any: + """Run a method of one of the cache's stores where its connection is. + One at a time. A read leaves its statement open until it is collected, so a + connection that starts a write while another request's read is still open + holds the file for reading as it asks to write. Another process committing at + that moment makes SQLite refuse the write at once, as ``database is locked``, + where it would otherwise have waited. A connection that finishes each + operation before it starts the next never asks for both. -async def cache_backend() -> Any: - """Return the shared cache for the running event loop, swept on first use. + Parameters + ---------- + table : str + ``responses`` or ``redirects``. + method : str + The store's method. + *args : Any + The method's arguments. Returns ------- Any - The ``SQLiteBackend`` every cached request on the running loop shares. + What the method returns; a method that yields returns the list it yielded. """ - global _cache, _cache_loop, _swept, _cache_users # noqa: PLW0603 - from asyncio import get_running_loop + async def call() -> Any: + async with _cache_guard: + result = getattr(getattr(_stores, table), method)(*args) - from aiohttp_client_cache import SQLiteBackend + if hasattr(result, "__aiter__"): + return [item async for item in result] + + return await result + + return await _on_cache_loop(call()) + + +async def _open() -> None: + """Open the stores if they are not open, sweeping the file first the first time. + + Runs on the cache loop. The sweep finishes before the stores exist, so no + request is ever in the middle of the file while it is vacuumed. + """ + global _stores, _swept # noqa: PLW0603 + + async with _cache_guard: + if _stores is not None: + return + + if not _swept: + await _sweep_file() + _swept = True + + _stores = _new_backend() + await _use_wal(_stores) + + +async def _close() -> None: + """Close the stores if no request is using them any more. + + ``SQLiteBackend.close()`` clears the connection object the backend was built + with and never rebuilds it, so a closed backend cannot be used again. The next + request opens a fresh one. + """ + global _stores # noqa: PLW0603 + + async with _cache_guard: + with _cache_lock: + idle = _cache_users == 0 + + if _stores is None or not idle: + return + + stores, _stores = _stores, None + + try: + await stores.close() + except Exception: # noqa: BLE001 + _logger.exception("Closing the Cboe response cache failed") + + +@asynccontextmanager +async def cache_backend(): + """Provide the response cache for the length of one request. + + Every request, on every loop and thread, is backed by the same connection. + It is opened for the first request to need it and closed after the last has + finished, so an idle process holds no connection open. + + Yields + ------ + SharedCache + The ``aiohttp_client_cache`` backend to build a ``CachedSession`` on. + """ + global _cache_users # noqa: PLW0603 + + from asyncio import run_coroutine_threadsafe, shield, wrap_future + + from openbb_cboe.utils.shared_cache import SharedCache + + with _cache_lock: + _cache_users += 1 - loop = get_running_loop() + try: + await _on_cache_loop(_open()) - if _cache is None or (_cache_loop is not loop and _cache_users == 0): - _cache = SQLiteBackend( - cache_path(), + yield SharedCache( + _store_call, + cache_name=CACHE_NAME, expire_after=CACHE_TTL["*"], urls_expire_after=CACHE_TTL, autoclose=False, ) - _cache_loop = loop + finally: + with _cache_lock: + _cache_users -= 1 - if not _swept: - _swept = True - _cache_users += 1 + # Handed to the cache loop here and now, not by a task that could be + # cancelled, or whose loop could end, before it ever started: a close + # that is lost leaves the connection, and its thread, open for good. + closing = run_coroutine_threadsafe(_close(), _get_cache_loop()) + await shield(wrap_future(closing)) - try: - await _sweep(_cache) - except Exception: # noqa: BLE001 - _logger.exception("Sweeping the Cboe response cache failed") - finally: - _cache_users -= 1 - return _cache +def _forget_cache_loop() -> None: + """Start over in a forked child, whose cache loop's thread did not survive. + + The child would otherwise wait for ever on a loop that is not running. The + parent's connection is kept, never closed or collected, because closing an + SQLite connection in the process it was not opened in can undo the parent's + work on the file. + """ + global _cache_lock, _cache_loop, _cache_guard, _cache_users, _stores # noqa: PLW0603 + + if _stores is not None: + _inherited.append(_stores) + + _cache_lock = threading.Lock() + _cache_loop = None + _cache_guard = None + _cache_users = 0 + _stores = None + + +def _forget_after_fork() -> None: + """Have a forked child start over, on the platforms that can fork.""" + if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_forget_cache_loop) + + +_forget_after_fork() async def get_cboe_data(url: str, use_cache: bool = True, **kwargs) -> Any: @@ -292,26 +549,15 @@ async def get_cboe_data(url: str, use_cache: bool = True, **kwargs) -> Any: Any The deserialized response. """ - global _cache_users # noqa: PLW0603 - from aiohttp_client_cache.session import CachedSession if use_cache is not True: return await amake_request(url, response_callback=response_callback, **kwargs) - backend = await cache_backend() - _cache_users += 1 - - try: - async with CachedSession(cache=backend) as session: - response = await session.get(url, timeout=10, **kwargs) - - return await response_callback(response, None) - finally: - _cache_users -= 1 + async with cache_backend() as backend, CachedSession(cache=backend) as session: + response = await session.get(url, timeout=10, **kwargs) - if _cache_users == 0: - await close_cache_backend(backend) + return await response_callback(response, None) async def get_company_directory(use_cache: bool = True, **kwargs) -> DataFrame: diff --git a/openbb_platform/providers/cboe/openbb_cboe/utils/options/create_surface.py b/openbb_platform/providers/cboe/openbb_cboe/utils/options/create_surface.py index 3b67bade54f9..2a99496df268 100644 --- a/openbb_platform/providers/cboe/openbb_cboe/utils/options/create_surface.py +++ b/openbb_platform/providers/cboe/openbb_cboe/utils/options/create_surface.py @@ -146,7 +146,7 @@ def _between(frame, col, low, high): "open_interest", "volume", ] - df = df[[c for c in dict.fromkeys(columns) if c in df.columns]] + df = df[[c for c in dict.fromkeys(columns) if c in df.columns]].copy() metric_title = METRIC_TITLES.get(metric, metric.replace("_", " ").title()) label = f"{symbol} {OPTION_TYPE_LABELS[option_type]} {metric_title} Surface" diff --git a/openbb_platform/providers/cboe/openbb_cboe/utils/shared_cache.py b/openbb_platform/providers/cboe/openbb_cboe/utils/shared_cache.py new file mode 100644 index 000000000000..83ea7339d679 --- /dev/null +++ b/openbb_platform/providers/cboe/openbb_cboe/utils/shared_cache.py @@ -0,0 +1,103 @@ +"""Response cache whose storage is reached through another event loop. + +An aiosqlite connection, and the lock that guards it, only work on the event +loop that first used them. Cboe requests run on whichever loop their caller +has - core's ``run_async`` gives every synchronous call a thread and a loop of +its own - so the one connection they all share has to live on a loop of its +own, and each request reaches it from wherever it is running. These classes are +that reach: they implement the storage interface ``aiohttp_client_cache`` asks +of a custom backend and leave the work to a ``call`` function. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator, Awaitable, Callable +from typing import Any + +from aiohttp_client_cache import CacheBackend +from aiohttp_client_cache.backends import BaseCache + +StoreCall = Callable[..., Awaitable[Any]] +"""``call(table, method, *args)`` runs a store's method where its connection is. + +A method that yields, like ``keys``, comes back as the list of what it yielded. +""" + + +class SharedStore(BaseCache): + """One table of the shared cache, usable from any event loop. + + Parameters + ---------- + table : str + The name of the backend's store this stands for, ``responses`` or + ``redirects``. + call : StoreCall + Runs a method of that store where the connection is. + """ + + def __init__(self, table: str, call: StoreCall) -> None: + super().__init__() + self._table = table + self._call = call + + async def contains(self, key: str) -> bool: + """Check if a key is stored in the cache.""" + return await self._call(self._table, "contains", key) + + async def clear(self) -> None: + """Delete all items from the cache.""" + await self._call(self._table, "clear") + + async def close(self) -> None: + """Leave the connection open: other requests are still using it.""" + + async def delete(self, key: str) -> None: + """Delete an item from the cache.""" + await self._call(self._table, "delete", key) + + async def bulk_delete(self, keys: set) -> None: + """Delete items from the cache.""" + await self._call(self._table, "bulk_delete", keys) + + async def keys(self) -> AsyncIterator[str]: # type: ignore[override] + """Yield all keys stored in the cache.""" + for key in await self._call(self._table, "keys"): + yield key + + async def read(self, key: str) -> Any: + """Read an item from the cache; ``None`` if it is missing.""" + return await self._call(self._table, "read", key) + + async def size(self) -> int: + """Get the number of items in the cache.""" + return await self._call(self._table, "size") + + async def values(self) -> AsyncIterator[Any]: # type: ignore[override] + """Yield all values stored in the cache.""" + for value in await self._call(self._table, "values"): + yield value + + async def write(self, key: str, item: Any) -> None: + """Write an item to the cache.""" + await self._call(self._table, "write", key, item) + + +class SharedCache(CacheBackend): + """A response cache that keeps what it stores through ``call``. + + Expiry, keys and redirects are decided here, on the caller's loop, as they + are for any backend; only reading and writing leave it. + + Parameters + ---------- + call : StoreCall + Runs a store's method where the connection is. + **kwargs + The settings of ``aiohttp_client_cache.CacheBackend``. + """ + + def __init__(self, call: StoreCall, **kwargs: Any) -> None: + super().__init__(**kwargs) + self.responses = SharedStore("responses", call) + self.redirects = SharedStore("redirects", call) diff --git a/openbb_platform/providers/cboe/pyproject.toml b/openbb_platform/providers/cboe/pyproject.toml index b3abf4961ab4..4e99e45bb869 100644 --- a/openbb_platform/providers/cboe/pyproject.toml +++ b/openbb_platform/providers/cboe/pyproject.toml @@ -79,6 +79,18 @@ exclude = ["tests", "integration"] [tool.ty.rules] unresolved-import = "warn" +# The models extend the platform's standard models by overriding their fields: +# a query parameter narrowed to a ``Literal`` (core derives the parameter's choices +# from it), or a field the provider may leave empty. ty treats both as unsound +# overrides of a mutable attribute. The check is off for the model modules, where +# that is how a provider is written, and stays on everywhere else. +[[tool.ty.overrides]] +include = ["openbb_cboe/models/**"] + +[tool.ty.overrides.rules] +invalid-attribute-override = "ignore" +invalid-mutable-override = "ignore" + [tool.coverage.report] exclude_lines = ["pragma: no cover", "if TYPE_CHECKING:", "raise NotImplementedError", "@overload"] diff --git a/openbb_platform/providers/cboe/tests/cache_worker.py b/openbb_platform/providers/cboe/tests/cache_worker.py new file mode 100644 index 000000000000..0c1707588f51 --- /dev/null +++ b/openbb_platform/providers/cboe/tests/cache_worker.py @@ -0,0 +1,73 @@ +"""One OS process making cached requests, for the tests that share a cache file. + +Run as ``cache_worker.py URL CACHE_FILE ROUNDS REQUESTS_PER_ROUND THREADS``. Each +thread runs bursts of concurrent requests on an event loop of its own, over a few +dozen URLs the other processes ask for too, and the last line printed is a JSON +summary of how many succeeded and what the failures were. +""" + +import asyncio +import collections +import json +import logging +import sys +import threading +import traceback +from pathlib import Path + +from openbb_cboe.utils import helpers + +URL, CACHE_FILE, ROUNDS, PER_ROUND, THREADS = sys.argv[1:6] +URLS = 40 +# Requests in flight per event loop. A test server can only queue so many +# connections at once (128 on macOS); this keeps the workers' total under it, so +# what the test measures is the cache and not the server's accept queue. +IN_FLIGHT = 2 + +logging.disable(logging.CRITICAL) +helpers.cache_path = lambda: CACHE_FILE +succeeded = 0 +failures: collections.Counter = collections.Counter() + + +async def request(number, turn): + """Request one URL, recording how it went.""" + global succeeded # noqa: PLW0603 + + try: + async with turn: + await asyncio.wait_for( + helpers.get_cboe_data(f"{URL}/{number % URLS}.json"), timeout=30 + ) + + succeeded += 1 + except Exception as error: # noqa: BLE001 + frames = traceback.extract_tb(error.__traceback__)[-4:] + where = " <- ".join( + f"{Path(frame.filename).name}:{frame.lineno} {frame.name}" + for frame in reversed(frames) + ) + failures[f"{type(error).__name__}: {error} (at {where})"] += 1 + + +def work(thread): + """Run the bursts of one thread, each on a new event loop.""" + + async def burst(first): + turn = asyncio.Semaphore(IN_FLIGHT) + + await asyncio.gather(*(request(first + n, turn) for n in range(int(PER_ROUND)))) + + for round_number in range(int(ROUNDS)): + asyncio.run(burst(round_number * int(PER_ROUND) + thread * 7)) + + +threads = [threading.Thread(target=work, args=(n,)) for n in range(int(THREADS))] + +for thread in threads: + thread.start() + +for thread in threads: + thread.join() + +sys.stdout.write(json.dumps({"succeeded": succeeded, "failures": failures}) + "\n") diff --git a/openbb_platform/providers/cboe/tests/test_helpers.py b/openbb_platform/providers/cboe/tests/test_helpers.py index b6e275919345..17d2c79d76ff 100644 --- a/openbb_platform/providers/cboe/tests/test_helpers.py +++ b/openbb_platform/providers/cboe/tests/test_helpers.py @@ -1,10 +1,18 @@ """Tests for openbb_cboe.utils.helpers.""" import asyncio -from contextlib import asynccontextmanager +import json +import os +import sqlite3 +import subprocess +import sys +import threading +import time +from contextlib import asynccontextmanager, closing from datetime import date +from pathlib import Path from types import SimpleNamespace -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from pandas import DataFrame @@ -57,6 +65,73 @@ async def close(self): self.closed = True +@pytest.fixture +def cache_file(tmp_path): + """Keep the response cache in a file of its own, and check nothing is left open.""" + with ( + patch.object(helpers, "cache_path", lambda: str(tmp_path / "cboe")), + patch.object(helpers, "_swept", True), + ): + yield tmp_path + + assert helpers._cache_users == 0 + assert helpers._stores is None + + +@pytest.fixture +def server(): + """A local HTTP server that records the path of every request it is sent.""" + from aiohttp import web + + requests: list = [] + started = threading.Event() + running: dict = {} + + async def handler(request): + requests.append(request.path) + await asyncio.sleep(0.02) + + return web.json_response({"path": request.path}) + + async def serve(): + app = web.Application() + app.router.add_get("/{tail:.*}", handler) + runner = web.AppRunner(app) + await runner.setup() + await web.TCPSite(runner, "127.0.0.1", 0).start() + running["url"] = f"http://127.0.0.1:{runner.addresses[0][1]}" + running["loop"] = asyncio.get_running_loop() + running["stop"] = asyncio.Event() + started.set() + await running["stop"].wait() + await runner.cleanup() + + thread = threading.Thread(target=lambda: asyncio.run(serve()), daemon=True) + thread.start() + assert started.wait(10) + + yield SimpleNamespace(url=running["url"], requests=requests) + + running["loop"].call_soon_threadsafe(running["stop"].set) + thread.join(10) + + +def _on_a_thread(factory): + """Run ``factory()`` on a thread, and an event loop, of its own.""" + outcome: dict = {} + + def target(): + try: + outcome["result"] = asyncio.run(factory()) + except BaseException as error: # noqa: BLE001 + outcome["error"] = error + + thread = threading.Thread(target=target, daemon=True) + thread.start() + + return thread, outcome + + class TestNewYorkClock: """Every session-date default is anchored to America/New_York.""" @@ -103,35 +178,70 @@ class TestGetCboeData: """Cache and no-cache request paths.""" def test_uses_cached_session(self): - """With use_cache, the request goes through the SQLite-backed session.""" - from unittest.mock import AsyncMock + """With use_cache, the request goes through a session on the shared cache.""" + + @asynccontextmanager + async def shared_cache(): + yield "backend" session = FakeCachedSession(FakeResponse("application/json", {"ok": True})) - backend = SimpleNamespace(close=AsyncMock()) with ( patch.object(helpers, "amake_request") as mock_request, - patch("aiohttp_client_cache.session.CachedSession", return_value=session), - patch.object(helpers, "cache_backend", AsyncMock(return_value=backend)), + patch( + "aiohttp_client_cache.session.CachedSession", return_value=session + ) as cached_session, + patch.object(helpers, "cache_backend", shared_cache), ): result = asyncio.run(helpers.get_cboe_data("https://cboe.test/a.json")) assert result == {"ok": True} assert session.closed is True - backend.close.assert_awaited_once() + cached_session.assert_called_once_with(cache="backend") mock_request.assert_not_called() + def test_a_failed_request_raises_and_lets_go_of_the_cache(self): + """The cache is released however the request ends.""" + released = [] + + @asynccontextmanager + async def shared_cache(): + try: + yield "backend" + finally: + released.append(True) + + class BrokenSession(FakeCachedSession): + async def get(self, url, **kwargs): + raise TimeoutError("cboe is down") + + session = BrokenSession(FakeResponse("application/json", {})) + + with ( + patch("aiohttp_client_cache.session.CachedSession", return_value=session), + patch.object(helpers, "cache_backend", shared_cache), + pytest.raises(TimeoutError), + ): + asyncio.run(helpers.get_cboe_data("https://cboe.test/a.json")) + + assert released == [True] + assert session.closed is True + def test_bypasses_cache(self): """Without use_cache, the request goes straight through amake_request.""" - with patch.object( - helpers, "amake_request", new=AsyncMock(return_value={"ok": False}) - ) as mock_request: + with ( + patch.object( + helpers, "amake_request", new=AsyncMock(return_value={"ok": False}) + ) as mock_request, + patch.object(helpers, "cache_backend") as shared_cache, + ): result = asyncio.run( helpers.get_cboe_data("https://cboe.test/a.json", use_cache=False) ) assert result == {"ok": False} mock_request.assert_awaited_once() + shared_cache.assert_not_called() class TestDirectories: @@ -678,15 +788,59 @@ async def _keys(count): yield f"key{index}" -@asynccontextmanager -async def _connection(): - """Stand in for the aiosqlite connection a vacuum runs on.""" +class _Cursor: + """Stand in for the cursor of an aiosqlite statement.""" + + def __init__(self, row): + self._row = row - class Connection: - async def execute(self, statement): - return None + async def fetchone(self): + """Return the one row there is.""" + return self._row - yield Connection() + async def close(self): + """Close the cursor.""" + + +class _Database: + """Stand in for the aiosqlite connection: it reports a size and logs what it is told.""" + + def __init__(self, size=0, page_size=1, unused_pages=0): + self.size = size + self.page_size = page_size + self.unused_pages = unused_pages + self.statements: list = [] + + async def execute(self, statement): + """Record the statement and answer the pragmas the cache asks.""" + self.statements.append(statement) + answers = { + "PRAGMA page_count": (self.size // self.page_size,), + "PRAGMA page_size": (self.page_size,), + "PRAGMA freelist_count": (self.unused_pages,), + "PRAGMA journal_mode=WAL": ("wal",), + } + + return _Cursor(answers.get(statement, (0, -1, -1))) + + +def _stores(database, held=0, keys=_empty, bulk_delete=None): + """Stand in for the SQLite stores of the cache, on a fake database.""" + + @asynccontextmanager + async def get_connection(commit=False): + yield database + + return SimpleNamespace( + close=AsyncMock(), + delete_expired_responses=AsyncMock(), + responses=SimpleNamespace( + size=AsyncMock(return_value=held), + keys=keys, + bulk_delete=bulk_delete or AsyncMock(), + get_connection=get_connection, + ), + ) class TestResponseCache: @@ -711,285 +865,752 @@ def test_each_endpoint_keeps_for_as_long_as_it_stays_good(self): for url, seconds in expiry.items(): assert get_url_expiration(url, helpers.CACHE_TTL) == seconds, url - def test_one_cache_is_shared_by_every_request(self): - """A backend per request opens the file again for each one.""" - from unittest.mock import patch + def test_an_expired_response_is_deleted_not_just_ignored(self): + """An expiry alone leaves the row behind and the file grows forever.""" + backend = _stores(_Database()) - async def twice(): - return await helpers.cache_backend(), await helpers.cache_backend() + asyncio.run(helpers._sweep(backend)) - with ( - patch.object(helpers, "_cache", None), - patch.object(helpers, "_cache_loop", None), - patch.object(helpers, "_cache_users", 0), - patch.object(helpers, "_swept", True), - ): - first, second = asyncio.run(twice()) + backend.delete_expired_responses.assert_awaited_once() - assert first is second + def test_a_file_over_the_limit_drops_its_oldest(self): + """What the sweep leaves over the limit is trimmed by the overshoot.""" + database = _Database(size=1000, unused_pages=1) + dropped: list = [] + + async def bulk_delete(keys): + dropped.append(len(keys)) + database.size = 100 + + backend = _stores( + database, held=10, keys=lambda: _keys(10), bulk_delete=bulk_delete + ) + + with patch.object(helpers, "CACHE_MAX_BYTES", 500): + asyncio.run(helpers._sweep(backend)) + + assert dropped == [5] + + def test_a_sweep_with_nothing_left_to_drop_stops(self): + """An oversized file whose entries are already gone is not re-trimmed.""" + backend = _stores(_Database(size=1000), held=0) + + with patch.object(helpers, "CACHE_MAX_BYTES", 500): + asyncio.run(helpers._sweep(backend)) + + backend.responses.bulk_delete.assert_not_awaited() - def test_a_new_event_loop_gets_a_fresh_backend(self): - """Locks bound to a finished loop must not leak into the next one.""" - from unittest.mock import patch + def test_a_file_under_the_limit_is_left_alone(self): + """Nothing is dropped from a file that already fits.""" + backend = _stores(_Database(size=100), held=10, keys=lambda: _keys(10)) + + with patch.object(helpers, "CACHE_MAX_BYTES", 500): + asyncio.run(helpers._sweep(backend)) + + backend.responses.bulk_delete.assert_not_awaited() + + def test_the_size_is_the_pages_the_file_holds(self): + """The size comes from SQLite, which counts pages still waiting in the log.""" + database = _Database(size=8192, page_size=4096) + + assert asyncio.run(helpers._size(_stores(database))) == 8192 + assert database.statements == ["PRAGMA page_count", "PRAGMA page_size"] + + def test_a_vacuum_checkpoints_the_log(self): + """Vacuumed pages only reach the disk, and free its space, once checkpointed.""" + database = _Database(unused_pages=3) + + asyncio.run(helpers._vacuum(_stores(database))) + + assert database.statements == [ + "PRAGMA freelist_count", + "VACUUM", + "PRAGMA wal_checkpoint(TRUNCATE)", + ] + + def test_a_vacuum_with_nothing_to_give_back_does_not_run(self): + """Every process sweeps as it starts, and a vacuum rewrites the whole file.""" + database = _Database(unused_pages=0) + + asyncio.run(helpers._vacuum(_stores(database))) + + assert database.statements == ["PRAGMA freelist_count"] + + +class TestCacheLifecycle: + """One connection for every request, opened by the first and closed by the last.""" + + @staticmethod + def _stores(built): + """Stand in for the SQLite stores, recording each one that is opened.""" + + def new_backend(): + stores = _stores(_Database()) + built.append(stores) + + return stores + + return patch.object(helpers, "_new_backend", new_backend) + + @staticmethod + async def _use_the_cache(): + async with helpers.cache_backend(): + pass + + def test_requests_on_other_loops_share_one_connection(self, cache_file): + """A request on another loop is served by the connection already open.""" + built: list = [] + holding, release = threading.Event(), threading.Event() + + async def hold(): + async with helpers.cache_backend(): + holding.set() + + while not release.is_set(): + await asyncio.sleep(0.005) + + with self._stores(built): + holder, outcome = _on_a_thread(hold) + + try: + assert holding.wait(10) + asyncio.run(self._use_the_cache()) + + assert len(built) == 1 + built[0].close.assert_not_awaited() + finally: + release.set() + holder.join(10) + + assert "error" not in outcome + built[0].close.assert_awaited_once() + + def test_the_last_request_to_finish_closes_it(self, cache_file): + """Nothing stays open while nothing is running, and a closed one is not reused.""" + built: list = [] + + with self._stores(built): + asyncio.run(self._use_the_cache()) + asyncio.run(self._use_the_cache()) + + assert len(built) == 2 + assert all(stores.close.await_count == 1 for stores in built) + + def test_a_request_that_fails_still_lets_go(self, cache_file): + """An error in the request must not leave the connection open for ever.""" + built: list = [] + + async def fails(): + async with helpers.cache_backend(): + raise ValueError("the request failed") + + with self._stores(built), pytest.raises(ValueError): + asyncio.run(fails()) + + built[0].close.assert_awaited_once() + + def test_a_cancelled_request_still_lets_go(self, cache_file): + """Cancelling the request must not leave the connection open for ever.""" + built: list = [] + + async def scenario(): + inside = asyncio.Event() + + async def hold(): + async with helpers.cache_backend(): + inside.set() + await asyncio.sleep(30) + + task = asyncio.create_task(hold()) + await asyncio.wait_for(inside.wait(), timeout=10) + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + with self._stores(built): + asyncio.run(scenario()) + + built[0].close.assert_awaited_once() + + def test_a_close_is_not_lost_when_the_request_is_cancelled_again(self, cache_file): + """Ending the loop on a request that is closing must not leave the cache open. + + ``asyncio.run`` cancels what is left when its main coroutine returns. A close + handed to a task that had not yet begun was lost with it, and the + connection, and its thread, stayed open. + """ + built: list = [] + + async def scenario(): + inside = asyncio.Event() + + async def hold(): + async with helpers.cache_backend(): + inside.set() + await asyncio.sleep(30) + + task = asyncio.create_task(hold()) + await asyncio.wait_for(inside.wait(), timeout=10) + task.cancel() + # Returns at once: the request has not yet reached its close. + + with self._stores(built): + asyncio.run(scenario()) + + deadline = time.monotonic() + 10 + + while helpers._stores is not None and time.monotonic() < deadline: + time.sleep(0.01) + + assert helpers._stores is None + built[0].close.assert_awaited_once() + + def test_the_file_is_swept_once_and_before_anything_opens_it(self, cache_file): + """The sweep vacuums the file, so no request may be in it while it runs.""" + order: list = [] + + async def sweep(): + order.append("sweep") + + def new_backend(): + order.append("open") + + return _stores(_Database()) with ( - patch.object(helpers, "_cache", None), - patch.object(helpers, "_cache_loop", None), - patch.object(helpers, "_cache_users", 0), - patch.object(helpers, "_swept", True), + patch.object(helpers, "_swept", False), + patch.object(helpers, "_sweep_file", sweep), + patch.object(helpers, "_new_backend", new_backend), ): - first = asyncio.run(helpers.cache_backend()) - second = asyncio.run(helpers.cache_backend()) + asyncio.run(self._use_the_cache()) + asyncio.run(self._use_the_cache()) - assert first is not second + assert order == ["sweep", "open", "open"] - def test_the_backend_outlives_any_one_session(self): - """A session closing on exit must not take the shared connection with it.""" - from unittest.mock import patch + def test_a_failed_sweep_is_not_retried_and_does_not_stop_requests(self, cache_file): + """A sweep that fails is logged, and the requests carry on.""" + built: list = [] + sweep = AsyncMock(side_effect=Exception("unreadable")) with ( - patch.object(helpers, "_cache", None), - patch.object(helpers, "_cache_loop", None), - patch.object(helpers, "_cache_users", 0), - patch.object(helpers, "_swept", True), + patch.object(helpers, "_swept", False), + patch.object(helpers, "_sweep", sweep), + self._stores(built), ): - backend = asyncio.run(helpers.cache_backend()) + asyncio.run(self._use_the_cache()) + asyncio.run(self._use_the_cache()) - assert backend.autoclose is False + sweep.assert_awaited_once() + assert len(built) == 3 # the sweep's own connection, then one per use - def test_a_failing_sweep_still_returns_the_backend(self): - """A sweep failure is logged, never raised into the request.""" - from unittest.mock import patch + def test_the_sweep_connection_is_closed_whatever_the_sweep_does(self, cache_file): + """A sweep that raises, or a close that raises, is logged and goes no further.""" + backend = SimpleNamespace(close=AsyncMock(side_effect=Exception("locked"))) with ( - patch.object(helpers, "_cache", None), - patch.object(helpers, "_cache_loop", None), - patch.object(helpers, "_cache_users", 0), - patch.object(helpers, "_swept", False), + patch.object(helpers, "_new_backend", return_value=backend), patch.object( helpers, "_sweep", new=AsyncMock(side_effect=Exception("boom")) ), ): - backend = asyncio.run(helpers.cache_backend()) + asyncio.run(helpers._sweep_file()) + + backend.close.assert_awaited_once() - assert backend is not None + def test_a_close_that_fails_is_logged_not_raised(self, cache_file): + """A close error must not surface as the request's failure.""" + stores = _stores(_Database()) + stores.close = AsyncMock(side_effect=Exception("locked")) - def test_the_shared_cache_closes_only_after_the_last_request(self): - """A request finishing must not close the connection under another.""" - from asyncio import Event, create_task, sleep + with patch.object(helpers, "_new_backend", return_value=stores): + asyncio.run(self._use_the_cache()) - fake_cache = SimpleNamespace(close=AsyncMock()) - gate = Event() - overlap_closes: list = [] + stores.close.assert_awaited_once() - class GatedSession(FakeCachedSession): - async def get(self, url, **kwargs): - if "slow" in url: - await gate.wait() - return await super().get(url, **kwargs) +class TestCacheLoop: + """The connection lives on one loop of its own, whichever loop asks.""" - async def scenario(): - slow = create_task(helpers.get_cboe_data("https://cboe.test/slow.json")) + def test_stores_are_reached_on_the_cache_thread(self): + """Every operation runs on the one thread the connection belongs to.""" + threads: list = [] - while helpers._cache_users == 0: - await sleep(0) + async def read(key): + threads.append(threading.current_thread().name) - await helpers.get_cboe_data("https://cboe.test/fast.json") - overlap_closes.append(fake_cache.close.await_count) - gate.set() - await slow + return f"value of {key}" - with ( - patch( - "aiohttp_client_cache.session.CachedSession", - side_effect=lambda **kwargs: GatedSession( - FakeResponse("application/json", {"ok": True}) - ), - ), - patch.object(helpers, "cache_backend", AsyncMock(return_value=fake_cache)), - patch.object(helpers, "_cache_users", 0), - ): - asyncio.run(scenario()) + stores = SimpleNamespace(responses=SimpleNamespace(read=read)) - assert overlap_closes == [0] - fake_cache.close.assert_awaited_once() + async def ask(key): + return await helpers._store_call("responses", "read", key) - def test_an_expired_response_is_deleted_not_just_ignored(self, tmp_path): - """An expiry alone leaves the row behind and the file grows forever.""" - from unittest.mock import AsyncMock, patch - - backend = SimpleNamespace( - delete_expired_responses=AsyncMock(), - responses=SimpleNamespace( - size=AsyncMock(return_value=0), - keys=_empty, - bulk_delete=AsyncMock(), - get_connection=lambda commit=False: _connection(), - ), + with patch.object(helpers, "_stores", stores): + first = asyncio.run(ask("a")) + thread, outcome = _on_a_thread(lambda: ask("b")) + thread.join(10) + + assert first == "value of a" + assert outcome == {"result": "value of b"} + assert threads == ["cboe-cache", "cboe-cache"] + + def test_what_a_store_yields_comes_back_as_a_list(self): + """Keys and values cannot be iterated from another loop, so are gathered.""" + + async def keys(): + for key in ("a", "b"): + yield key + + stores = SimpleNamespace(redirects=SimpleNamespace(keys=keys)) + + with patch.object(helpers, "_stores", stores): + keys_found = asyncio.run(helpers._store_call("redirects", "keys")) + + assert keys_found == ["a", "b"] + + def test_an_error_in_a_store_reaches_the_caller(self): + """A failure on the cache loop is the failure of the request that asked.""" + stores = SimpleNamespace( + responses=SimpleNamespace(read=AsyncMock(side_effect=OSError("disk"))) ) - with patch.object(helpers, "cache_path", lambda: str(tmp_path / "absent")): - asyncio.run(helpers._sweep(backend)) + with patch.object(helpers, "_stores", stores), pytest.raises(OSError): + asyncio.run(helpers._store_call("responses", "read", "a")) - backend.delete_expired_responses.assert_awaited_once() + def test_the_connection_does_one_thing_at_a_time(self): + """No read may be open while another request's write starts. - def test_a_file_over_the_limit_drops_its_oldest(self, tmp_path): - """What the sweep leaves over the limit is trimmed by the overshoot.""" - from unittest.mock import AsyncMock, patch + The connection is held for reading by a read that is still open, and + asking to write then is refused at once, not waited out, if another + process is writing. + """ + running = 0 + most_at_once = 0 - stored = tmp_path / "probe.sqlite" - stored.write_bytes(b"x" * 1000) - dropped: list = [] + async def read(key): + nonlocal running, most_at_once - async def bulk_delete(keys): - dropped.append(len(keys)) - stored.write_bytes(b"x" * 100) - - backend = SimpleNamespace( - delete_expired_responses=AsyncMock(), - responses=SimpleNamespace( - size=AsyncMock(return_value=10), - keys=lambda: _keys(10), - bulk_delete=bulk_delete, - get_connection=lambda commit=False: _connection(), - ), + running += 1 + most_at_once = max(most_at_once, running) + await asyncio.sleep(0.002) + running -= 1 + + return key + + stores = SimpleNamespace( + responses=SimpleNamespace(read=read, write=read), + redirects=SimpleNamespace(read=read), ) + async def burst(): + return await asyncio.gather( + *( + helpers._store_call(table, method, str(n)) + for n in range(10) + for table, method in ( + ("responses", "read"), + ("responses", "write"), + ("redirects", "read"), + ) + ) + ) + with ( - patch.object(helpers, "cache_path", lambda: str(tmp_path / "probe")), - patch.object(helpers, "CACHE_MAX_BYTES", 500), + patch.object(helpers, "_stores", stores), + patch.object(helpers, "_swept", True), ): - asyncio.run(helpers._sweep(backend)) + workers = [_on_a_thread(burst) for _ in range(4)] - assert dropped == [5] - assert stored.stat().st_size <= 500 + for worker, _ in workers: + worker.join(30) - def test_a_sweep_with_nothing_left_to_drop_stops(self, tmp_path): - """An oversized file whose entries are already gone is not re-trimmed.""" - from unittest.mock import AsyncMock, patch - - stored = tmp_path / "probe.sqlite" - stored.write_bytes(b"x" * 1000) - backend = SimpleNamespace( - delete_expired_responses=AsyncMock(), - responses=SimpleNamespace( - size=AsyncMock(return_value=0), - keys=_empty, - bulk_delete=AsyncMock(), - get_connection=lambda commit=False: _connection(), - ), - ) + assert all("error" not in outcome for _, outcome in workers) + assert most_at_once == 1 + + def test_a_forked_child_forgets_the_parents_loop(self): + """The loop's thread does not survive a fork, so the child must not wait on it.""" + stores = SimpleNamespace(close=AsyncMock()) + inherited: list = [] with ( - patch.object(helpers, "cache_path", lambda: str(tmp_path / "probe")), - patch.object(helpers, "CACHE_MAX_BYTES", 500), + patch.object(helpers, "_cache_lock", threading.Lock()), + patch.object(helpers, "_cache_loop", "parents loop"), + patch.object(helpers, "_cache_guard", "parents guard"), + patch.object(helpers, "_cache_users", 3), + patch.object(helpers, "_stores", stores), + patch.object(helpers, "_inherited", inherited), ): - asyncio.run(helpers._sweep(backend)) + helpers._forget_cache_loop() - backend.responses.bulk_delete.assert_not_awaited() + assert helpers._cache_loop is None + assert helpers._cache_guard is None + assert helpers._cache_users == 0 + assert helpers._stores is None - def test_a_file_under_the_limit_is_left_alone(self, tmp_path): - from unittest.mock import AsyncMock, patch - - stored = tmp_path / "probe.sqlite" - stored.write_bytes(b"x" * 100) - backend = SimpleNamespace( - delete_expired_responses=AsyncMock(), - responses=SimpleNamespace( - size=AsyncMock(return_value=10), - keys=lambda: _keys(10), - bulk_delete=AsyncMock(), - get_connection=lambda commit=False: _connection(), - ), + # Kept, not closed: closing it in the child could undo the parent's work. + assert inherited == [stores] + stores.close.assert_not_called() + + def test_a_forked_child_is_set_to_forget_the_loop(self): + """Where processes fork, a child is told to start over when it is born.""" + register = MagicMock() + + with patch.object(helpers, "os", SimpleNamespace(register_at_fork=register)): + helpers._forget_after_fork() + + register.assert_called_once_with(after_in_child=helpers._forget_cache_loop) + + def test_a_platform_that_cannot_fork_has_nothing_to_set(self): + """Windows has no ``os.register_at_fork``, and has no forked children either.""" + with patch.object(helpers, "os", SimpleNamespace()): + helpers._forget_after_fork() + + @pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork") + def test_a_forked_child_can_make_a_cached_request(self, cache_file, server): + """A forked server worker must not hang waiting on the parent's cache loop.""" + import signal + + url = f"{server.url}/forked.json" + asyncio.run(helpers.get_cboe_data(f"{server.url}/before-the-fork.json")) + pid = os.fork() + + if pid == 0: # pragma: no cover - runs in the child + signal.alarm(20) + code = 1 + + try: + result = asyncio.run(helpers.get_cboe_data(url)) + code = 0 if result == {"path": "/forked.json"} else 2 + finally: + os._exit(code) + + _, status = os.waitpid(pid, 0) + + assert os.waitstatus_to_exitcode(status) == 0 + assert asyncio.run(helpers.get_cboe_data(url)) == {"path": "/forked.json"} + + +class TestSharedAcrossEventLoops: + """Callers on separate threads, each with a loop of its own, share one cache. + + Core's ``run_async`` gives every synchronous call a thread and a loop of its + own. The cache used to hand them all one backend, whose lock and connection + belong to one loop: some callers failed with ``Lock is bound to a different + event loop`` or ``threads can only be started once``, others waited for ever. + """ + + def test_a_repeated_request_is_served_from_the_file(self, cache_file, server): + """The second request never reaches the server.""" + url = f"{server.url}/a.json" + + assert asyncio.run(helpers.get_cboe_data(url)) == {"path": "/a.json"} + assert asyncio.run(helpers.get_cboe_data(url)) == {"path": "/a.json"} + assert server.requests == ["/a.json"] + + def test_threads_with_their_own_loops_share_one_cache(self, cache_file, server): + """Concurrent callers all succeed, and what one fetched the others reuse.""" + threads, rounds, per_loop = 4, 4, 3 + # What is under test is that callers succeed, not how fast, so the limit is + # for a caller that is stuck. A runner with a slow disk is not one. + patience = 60 + + def wave(label): + barrier = threading.Barrier(threads) + + async def burst(): + barrier.wait(patience) + + return await asyncio.wait_for( + asyncio.gather( + *( + helpers.get_cboe_data(f"{server.url}/{label}/{n}.json") + for n in range(per_loop) + ) + ), + timeout=patience, + ) + + workers = [_on_a_thread(burst) for _ in range(threads)] + + for worker, _ in workers: + worker.join(patience * 2) + + return [outcome for _, outcome in workers] + + for round_number in range(rounds): + label = f"round{round_number}" + expected = [{"path": f"/{label}/{n}.json"} for n in range(per_loop)] + + for outcomes in (wave(label), wave(label)): + assert outcomes == [{"result": expected}] * threads, outcomes + + # Four callers asked for each URL at once, so each was fetched at + # least once; the second wave found every one of them on disk. + fetched = [ + path for path in server.requests if path.startswith(f"/{label}/") + ] + assert set(fetched) == {path["path"] for path in expected} + assert len(fetched) <= threads * per_loop + + settled = len(server.requests) + wave("round0") + + assert len(server.requests) == settled + + def test_an_unreadable_row_does_not_stop_requests(self, cache_file, server, caplog): + """A file the sweep cannot get through is logged, and still serves requests.""" + + async def leave_an_unreadable_row(): + backend = helpers._new_backend() + await backend.responses.write("unreadable", {"not": "a response"}) + await backend.close() + + asyncio.run(leave_an_unreadable_row()) + + with patch.object(helpers, "_swept", False): + result = asyncio.run(helpers.get_cboe_data(f"{server.url}/a.json")) + + assert result == {"path": "/a.json"} + assert "Sweeping the Cboe response cache failed" in caplog.text + + +class TestAcrossProcesses: + """Separate processes share the cache file without locking each other out. + + Each process has a connection of its own. One whose read was still open when + another request began to write held the file for reading while it asked to + write, and SQLite refuses that at once when another process is writing: + several percent of the requests failed with ``database is locked``. + """ + + def test_processes_sharing_the_file_all_succeed(self, tmp_path, server): + """Many processes, each with several threads, on one file.""" + processes, threads, rounds, per_round = 12, 4, 12, 8 + worker = Path(__file__).with_name("cache_worker.py") + environment = { + **os.environ, + "PYTHONPATH": os.pathsep.join(filter(None, sys.path)), + } + command = [ + sys.executable, + "-W", + "ignore", + str(worker), + server.url, + str(tmp_path / "shared"), + str(rounds), + str(per_round), + str(threads), + ] + started = [ + subprocess.Popen( # noqa: S603 + command, + env=environment, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + for _ in range(processes) + ] + summaries = [] + + for process in started: + stdout, stderr = process.communicate(timeout=120) + assert process.returncode == 0, stderr + summaries.append(json.loads(stdout.splitlines()[-1])) + + assert [summary["failures"] for summary in summaries] == [{}] * processes + assert {summary["succeeded"] for summary in summaries} == { + threads * rounds * per_round + } + + +class TestWriteAheadLogging: + """Readers and a writer, in one process or in several, do not block each other. + + The cache file is shared by every process that uses Cboe data, and in + SQLite's default mode a process holding it - to write, or to read - stops + the others from doing the opposite for as long as it does. One that holds it + for longer than SQLite will wait fails the request with ``database is + locked``. + """ + + @staticmethod + def _journal_mode(tmp_path): + with closing(sqlite3.connect(tmp_path / "cboe.sqlite")) as database: + return database.execute("PRAGMA journal_mode").fetchone()[0] + + @staticmethod + def _held_by_another_process(tmp_path, statements): + """Hold the cache file the way another process in the middle of work would.""" + database = sqlite3.connect( + tmp_path / "cboe.sqlite", isolation_level=None, check_same_thread=False ) - with ( - patch.object(helpers, "cache_path", lambda: str(tmp_path / "probe")), - patch.object(helpers, "CACHE_MAX_BYTES", 500), - ): - asyncio.run(helpers._sweep(backend)) + for statement in statements: + database.execute(statement).fetchone() - backend.responses.bulk_delete.assert_not_awaited() + return database + + def test_the_cache_file_is_put_in_wal_mode(self, cache_file, server): + """The mode is set on the file, so every process that opens it has it.""" + asyncio.run(helpers.get_cboe_data(f"{server.url}/a.json")) + + assert self._journal_mode(cache_file) == "wal" + + def test_a_hit_is_served_while_another_process_writes(self, cache_file, server): + """A reader does not wait for a writer, however long it takes.""" + url = f"{server.url}/a.json" + asyncio.run(helpers.get_cboe_data(url)) + writer = self._held_by_another_process(cache_file, ["BEGIN EXCLUSIVE"]) + + async def hit(): + return await asyncio.wait_for(helpers.get_cboe_data(url), timeout=3) + + try: + assert asyncio.run(hit()) == {"path": "/a.json"} + finally: + writer.execute("ROLLBACK") + writer.close() + + assert server.requests == ["/a.json"] + def test_a_miss_is_served_while_another_process_reads(self, cache_file, server): + """A writer does not wait for a reader to finish.""" + asyncio.run(helpers.get_cboe_data(f"{server.url}/a.json")) + reader = self._held_by_another_process( + cache_file, ["BEGIN", "SELECT key FROM responses"] + ) + + async def miss(): + return await asyncio.wait_for( + helpers.get_cboe_data(f"{server.url}/b.json"), timeout=3 + ) + + try: + assert asyncio.run(miss()) == {"path": "/b.json"} + finally: + reader.execute("ROLLBACK") + reader.close() + + def test_each_statement_is_committed_as_it_runs(self, cache_file): + """The connection must not hold a transaction open between statements.""" + + async def isolation_level(): + stores = helpers._new_backend() + + try: + async with stores.responses.get_connection() as connection: + return connection.isolation_level + finally: + await stores.close() + + assert asyncio.run(isolation_level()) is None + + def test_a_commit_does_not_wait_for_the_disk(self, cache_file): + """A commit that waits for the disk holds the write lock for as long as it takes.""" + normal = 1 -class TestCacheBackendReuse: - """A closed backend must never be handed back out.""" + async def synchronous(): + stores = helpers._new_backend() - def test_a_second_request_does_not_reuse_a_closed_backend(self): - """Closing on the last request must not poison every request after it. + try: + async with stores.responses.get_connection() as connection: + return await helpers._pragma(connection, "PRAGMA synchronous") + finally: + await stores.close() - ``SQLiteBackend.close()`` clears the connection object it was built with - and never rebuilds it, so holding the reference made the first completed - request break all later ones with ``'NoneType' object has no attribute - '_connection'``. + assert asyncio.run(synchronous()) == (normal,) + + def test_a_write_cancelled_before_its_commit_leaves_no_lock(self, cache_file): + """Another process can write at once, whatever became of the write. + + A write is an ``INSERT`` and then, after the event loop has had a turn, a + ``COMMIT``. Cancelled in between, it used to leave the write lock held. """ - from unittest.mock import patch - with ( - patch.object(helpers, "_cache", None), - patch.object(helpers, "_cache_loop", None), - patch.object(helpers, "_cache_users", 0), - patch.object(helpers, "_swept", True), - ): + async def scenario(): + stores = helpers._new_backend() - async def _two_requests(): - first = await helpers.cache_backend() - await helpers.close_cache_backend(first) - second = await helpers.cache_backend() - return first, second + try: + reached = asyncio.Event() - first, second = asyncio.run(_two_requests()) + async def commit_never_arrives(connection): + reached.set() + await asyncio.sleep(30) - assert second is not first + with patch("aiosqlite.Connection.commit", commit_never_arrives): + write = asyncio.create_task(stores.responses.write("key", "item")) + await asyncio.wait_for(reached.wait(), timeout=10) + write.cancel() - def test_closing_clears_the_shared_reference(self): - """After a close, nothing is left pointing at the dead backend.""" - from unittest.mock import patch + with pytest.raises(asyncio.CancelledError): + await write - with ( - patch.object(helpers, "_cache", None), - patch.object(helpers, "_cache_loop", None), - patch.object(helpers, "_cache_users", 0), - patch.object(helpers, "_swept", True), - ): - backend = asyncio.run(helpers.cache_backend()) - assert helpers._cache is backend + with closing( + sqlite3.connect(cache_file / "cboe.sqlite", timeout=0) + ) as another_process: + another_process.execute("BEGIN IMMEDIATE") + another_process.execute("ROLLBACK") + finally: + await stores.close() - asyncio.run(helpers.close_cache_backend(backend)) + asyncio.run(scenario()) - assert helpers._cache is None - assert helpers._cache_loop is None + def test_a_file_that_cannot_be_switched_still_works(self, caplog): + """A file that stays in its default mode is a slower cache, not a failure.""" - def test_closing_a_stale_backend_leaves_the_current_one(self): - """A late close from an old request must not drop a newer shared backend.""" - from unittest.mock import AsyncMock, patch + @asynccontextmanager + async def get_connection(commit=False): + raise sqlite3.OperationalError("database is locked") + yield - stale = SimpleNamespace(close=AsyncMock()) - current = SimpleNamespace(close=AsyncMock()) + stores = SimpleNamespace( + responses=SimpleNamespace(get_connection=get_connection) + ) - with ( - patch.object(helpers, "_cache", current), - patch.object(helpers, "_cache_loop", "loop"), - ): - asyncio.run(helpers.close_cache_backend(stale)) + asyncio.run(helpers._use_wal(stores)) - assert helpers._cache is current - assert helpers._cache_loop == "loop" + assert "Could not put the Cboe response cache in WAL mode" in caplog.text - stale.close.assert_awaited_once() - current.close.assert_not_called() + def test_a_file_that_stays_in_another_mode_is_said_so(self, caplog): + """A file system that cannot keep the shared memory answers with its old mode.""" + database = _Database() - def test_a_failing_close_is_swallowed(self): - """A close error must not surface as the request's failure.""" - from unittest.mock import AsyncMock, patch + async def execute(statement): + return _Cursor(("delete",)) - backend = SimpleNamespace(close=AsyncMock(side_effect=Exception("locked"))) + database.execute = execute - with ( - patch.object(helpers, "_cache", backend), - patch.object(helpers, "_cache_loop", "loop"), - ): - asyncio.run(helpers.close_cache_backend(backend)) + asyncio.run(helpers._use_wal(_stores(database))) + + assert "staying in" in caplog.text + + def test_a_sweep_of_a_wal_file_trims_to_the_limit_and_no_further( + self, cache_file, server + ): + """The file on disk lags a vacuum, so the sweep must not size it by that. + + After the first ``VACUUM`` the new, smaller file waits in the log, and the + file on disk is still its old size. A sweep that believed it went on + dropping responses that were never in the way. + """ + asyncio.run(helpers.get_cboe_data(f"{server.url}/seed.json")) + stored = cache_file / "cboe.sqlite" + + with closing(sqlite3.connect(stored)) as database: + row = database.execute("SELECT value FROM responses").fetchone()[0] + database.executemany( + "INSERT OR REPLACE INTO responses VALUES (?, ?)", + ((f"filler{number}", row) for number in range(400)), + ) + database.commit() + + assert self._journal_mode(cache_file) == "wal" + limit = stored.stat().st_size // 2 + + with patch.object(helpers, "CACHE_MAX_BYTES", limit): + asyncio.run(helpers._sweep_file()) - assert helpers._cache is None + assert limit // 2 < stored.stat().st_size <= limit diff --git a/openbb_platform/providers/cboe/tests/test_options_analysis.py b/openbb_platform/providers/cboe/tests/test_options_analysis.py index 5e3b233e5d73..e75b3dfaf60b 100644 --- a/openbb_platform/providers/cboe/tests/test_options_analysis.py +++ b/openbb_platform/providers/cboe/tests/test_options_analysis.py @@ -189,6 +189,10 @@ def test_by_strike_for_one_expiration(self, options_chain): assert output.chart.content["data"] +# pandas warns that way when a frame that is a view of another is written to. The +# class it uses is not in every version, so it is matched by what it says, and the +# message starts on a new line, hence the flag that lets ``.`` match one. +@pytest.mark.filterwarnings("error:(?s).*trying to be set on a copy of a slice") class TestSurface: """3-D surface builder.""" diff --git a/openbb_platform/providers/cboe/tests/test_shared_cache.py b/openbb_platform/providers/cboe/tests/test_shared_cache.py new file mode 100644 index 000000000000..e74dc09d5bd0 --- /dev/null +++ b/openbb_platform/providers/cboe/tests/test_shared_cache.py @@ -0,0 +1,108 @@ +"""Tests for openbb_cboe.utils.shared_cache.""" + +import asyncio +from unittest.mock import AsyncMock + +import pytest + +from openbb_cboe.utils.shared_cache import SharedCache, SharedStore + + +async def _gather(iterable): + """Collect everything an async iterable yields.""" + return [item async for item in iterable] + + +class TestSharedStore: + """A store hands every operation to ``call``, whichever loop it is used on.""" + + @pytest.mark.parametrize( + ("method", "args", "returned"), + [ + ("contains", ("key",), True), + ("read", ("key",), "item"), + ("size", (), 7), + ("write", ("key", "item"), None), + ("delete", ("key",), None), + ("bulk_delete", ({"key"},), None), + ("clear", (), None), + ], + ) + def test_an_operation_is_run_where_the_connection_is(self, method, args, returned): + """The table, the method and its arguments go over, and the result comes back.""" + call = AsyncMock(return_value=returned) + store = SharedStore("responses", call) + + result = asyncio.run(getattr(store, method)(*args)) + + assert result == returned + call.assert_awaited_once_with("responses", method, *args) + + @pytest.mark.parametrize("method", ["keys", "values"]) + def test_what_is_yielded_is_yielded_here(self, method): + """Keys and values arrive gathered, and are yielded on the caller's loop.""" + call = AsyncMock(return_value=["a", "b"]) + store = SharedStore("redirects", call) + + assert asyncio.run(_gather(getattr(store, method)())) == ["a", "b"] + call.assert_awaited_once_with("redirects", method) + + def test_closing_leaves_the_connection_to_the_cache(self): + """Other requests are still using the connection, so a session may not close it.""" + call = AsyncMock() + + asyncio.run(SharedStore("responses", call).close()) + + call.assert_not_awaited() + + def test_pop_reads_then_deletes(self): + """The base class's ``pop`` works through the same two operations.""" + call = AsyncMock(side_effect=lambda table, method, *args: "item") + + assert asyncio.run(SharedStore("responses", call).pop("key")) == "item" + assert [step.args[1] for step in call.await_args_list] == ["read", "delete"] + + def test_an_error_where_the_connection_is_reaches_the_caller(self): + """A failed read is the failure of the request that made it.""" + call = AsyncMock(side_effect=OSError("disk")) + + with pytest.raises(OSError): + asyncio.run(SharedStore("responses", call).read("key")) + + +class TestSharedCache: + """The backend decides expiry and keys itself, and stores through ``call``.""" + + def test_its_tables_are_the_responses_and_the_redirects(self): + """Each table of the backend is a store of the same name.""" + call = AsyncMock() + cache = SharedCache(call) + + asyncio.run(cache.responses.write("key", "item")) + asyncio.run(cache.redirects.write("alias", "key")) + + assert [step.args[:2] for step in call.await_args_list] == [ + ("responses", "write"), + ("redirects", "write"), + ] + + def test_it_keeps_the_settings_it_is_given(self): + """Expiry is applied on the caller's loop, from the backend's own settings.""" + cache = SharedCache( + AsyncMock(), + expire_after=60, + urls_expire_after={"*/book/*": 5}, + autoclose=False, + ) + + assert cache.expire_after == 60 + assert cache.urls_expire_after == {"*/book/*": 5} + assert cache.autoclose is False + + def test_closing_it_does_not_close_the_shared_connection(self): + """Closing the backend is not closing the connection other requests use.""" + call = AsyncMock() + + asyncio.run(SharedCache(call).close()) + + call.assert_not_awaited()