mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-11-15 09:48:34 +00:00
perf: cut repeated work on the status and activity paths (#1545)
Two small cuts to work that repeats on every poll or request. The delivery-state sync behind /api/status and the activity snapshot listed every fulfilled request (SELECT * plus a JSON parse of book_data and release_data per row) before it looked at the queue. That history is never purged, so an idle instance paid for it on every poll. It now returns straight away when no queue bucket holds a task, since nothing could map to a request then. Results and emitted updates are unchanged. The activity snapshot gains less from this, because its status merges download history rows, so it is usually not empty. In proxy auth mode the before-request hook looked the user up by name and by session id on every API call, cover images included, each on a new SQLite connection. UserDB.get_user_cached answers repeat lookups from memory for up to five seconds, bounded at 256 entries. create_user, update_user and delete_user drop the cache as soon as they commit, and a lookup that overlaps a write is not stored, so a role change or deletion made in the app applies on the next request. Changes made outside that UserDB instance (another process, or the separate instances some helpers open) show within the five seconds. Only the middleware uses the cached path; every other get_user caller is unchanged. Each change has a test that failed before it: one counts list_requests calls for an empty queue, the others count get_user calls across repeated lookups and middleware runs, check expiry with a patched clock, and check that writes and caller mutations do not leave a stale row behind. The full suite passes (4432 passed, 48 skipped, e2e excluded), as do ruff and basedpyright on the changed files. I also looked at throttling the full status_update broadcast when only the progress message changes during torrent and usenet polling, and left it out. Those handlers poll every 2 seconds and DOWNLOAD_PROGRESS_UPDATE_INTERVAL defaults to 1 second, so a throttle on that interval would skip nothing at the default setting. A longer window would slow the speed and ETA text in the UI, and without a trailing flush it would hide the last message of a quick burst, such as "Waiting for bot response" arriving just after "Requesting file from bot", for the whole wait that follows. --------- Co-authored-by: CaliBrain <calibrain@l4n.xyz>
This commit is contained in:
1 parent
237ddd53a2
commit
72b5d04939
6 files changed
+180
-2
No files matched your search
@@ -204,6 +204,11 @@ def sync_delivery_states_from_queue_status(
|
||||
user_id: int | None = None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Persist delivery-state transitions for fulfilled requests based on queue status."""
|
||||
# Runs on every status poll. With no task in the queue nothing can change, so skip
|
||||
# reading (and parsing) the whole fulfilled-request history.
|
||||
if not any(queue_status.get(status_key) for status_key in QueueStatus):
|
||||
return []
|
||||
|
||||
fulfilled_rows = user_db.list_requests(user_id=user_id, status=RequestStatus.FULFILLED)
|
||||
if not fulfilled_rows:
|
||||
return []
|
||||
|
||||
@@ -4,6 +4,7 @@ import json
|
||||
import os
|
||||
import sqlite3
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, ClassVar
|
||||
|
||||
@@ -25,6 +26,11 @@ from shelfmark.core.utils import DB_BUSY_TIMEOUT_SECONDS
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
# How long get_user_cached may answer from memory. Writes through the same UserDB drop
|
||||
# the cache at once, so this only bounds changes made by another process or instance.
|
||||
USER_CACHE_TTL_SECONDS = 5.0
|
||||
_USER_CACHE_MAX_ENTRIES = 256
|
||||
|
||||
_CREATE_TABLES_SQL = """
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
@@ -188,6 +194,9 @@ class UserDB:
|
||||
"""Initialize the user database wrapper for the given SQLite path."""
|
||||
self._db_path = db_path
|
||||
self._lock = threading.Lock()
|
||||
self._user_cache: dict[tuple[str, object], tuple[float, dict[str, Any] | None]] = {}
|
||||
self._user_cache_lock = threading.Lock()
|
||||
self._user_cache_generation = 0
|
||||
|
||||
@property
|
||||
def db_path(self) -> str:
|
||||
@@ -318,6 +327,7 @@ class UserDB:
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
self._drop_user_cache()
|
||||
user_id = cursor.lastrowid
|
||||
if not isinstance(user_id, int):
|
||||
msg = "Failed to create user"
|
||||
@@ -353,6 +363,36 @@ class UserDB:
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
def get_user_cached(
|
||||
self, *, user_id: int | None = None, username: str | None = None
|
||||
) -> dict[str, Any] | None:
|
||||
"""Get a user like get_user, answering repeats from memory for a few seconds.
|
||||
|
||||
For hot paths such as per-request proxy auth, where every API call (cover
|
||||
images included) would otherwise open a connection for the same row.
|
||||
"""
|
||||
key = ("id", user_id) if user_id is not None else ("username", username)
|
||||
now = time.monotonic()
|
||||
with self._user_cache_lock:
|
||||
cached = self._user_cache.get(key)
|
||||
if cached is not None and cached[0] > now:
|
||||
return dict(cached[1]) if cached[1] else None
|
||||
generation = self._user_cache_generation
|
||||
|
||||
user = self.get_user(user_id=user_id, username=username)
|
||||
with self._user_cache_lock:
|
||||
# A write that landed during the read may have made this row stale.
|
||||
if generation == self._user_cache_generation:
|
||||
if len(self._user_cache) >= _USER_CACHE_MAX_ENTRIES:
|
||||
self._user_cache.clear()
|
||||
self._user_cache[key] = (now + USER_CACHE_TTL_SECONDS, user)
|
||||
return dict(user) if user else None
|
||||
|
||||
def _drop_user_cache(self) -> None:
|
||||
with self._user_cache_lock:
|
||||
self._user_cache.clear()
|
||||
self._user_cache_generation += 1
|
||||
|
||||
def _get_user_by_id(self, conn: sqlite3.Connection, user_id: int) -> dict[str, Any] | None:
|
||||
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
||||
return dict(row) if row else None
|
||||
@@ -399,6 +439,7 @@ class UserDB:
|
||||
for column, value in kwargs.items():
|
||||
conn.execute(self._USER_UPDATE_STATEMENTS[column], (value, user_id))
|
||||
conn.commit()
|
||||
self._drop_user_cache()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
@@ -427,6 +468,7 @@ class UserDB:
|
||||
)
|
||||
conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
||||
conn.commit()
|
||||
self._drop_user_cache()
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
+2
-2
@@ -969,7 +969,7 @@ def proxy_auth_middleware() -> Response | tuple[Response, int] | None:
|
||||
]
|
||||
is_admin = admin_group_name in user_groups
|
||||
elif user_db is not None:
|
||||
existing_db_user = user_db.get_user(username=username)
|
||||
existing_db_user = user_db.get_user_cached(username=username)
|
||||
if existing_db_user:
|
||||
is_admin = existing_db_user.get("role") == "admin"
|
||||
else:
|
||||
@@ -993,7 +993,7 @@ def proxy_auth_middleware() -> Response | tuple[Response, int] | None:
|
||||
|
||||
if raw_db_user_id is not None:
|
||||
try:
|
||||
session_db_user = user_db.get_user(user_id=int(raw_db_user_id))
|
||||
session_db_user = user_db.get_user_cached(user_id=int(raw_db_user_id))
|
||||
except TypeError, ValueError:
|
||||
session_db_user = None
|
||||
|
||||
|
||||
@@ -285,6 +285,43 @@ class TestProxyProvisioningRole:
|
||||
assert bob["is_admin"] is False
|
||||
assert temp_user_db.get_user(username="bob")["role"] == "user"
|
||||
|
||||
def test_repeat_requests_reuse_the_user_lookups_until_the_user_changes(
|
||||
self, main_module, temp_user_db, monkeypatch
|
||||
):
|
||||
monkeypatch.setattr(main_module, "user_db", temp_user_db)
|
||||
temp_user_db.create_user(username="ops", role="admin", auth_source="proxy")
|
||||
bob = temp_user_db.create_user(username="bob", role="user", auth_source="proxy")
|
||||
calls = []
|
||||
real_get_user = temp_user_db.get_user
|
||||
|
||||
def counting_get_user(**kwargs):
|
||||
calls.append(kwargs)
|
||||
return real_get_user(**kwargs)
|
||||
|
||||
monkeypatch.setattr(temp_user_db, "get_user", counting_get_user)
|
||||
|
||||
def run_middleware():
|
||||
with (
|
||||
patch.object(main_module, "get_auth_mode", return_value="proxy"),
|
||||
main_module.app.test_request_context(
|
||||
"/api/covers/abc", headers={"X-Auth-User": "bob"}
|
||||
),
|
||||
):
|
||||
main_module.session["user_id"] = "bob"
|
||||
main_module.session["db_user_id"] = bob["id"]
|
||||
assert main_module.proxy_auth_middleware() is None
|
||||
return main_module.session["is_admin"]
|
||||
|
||||
assert run_middleware() is False
|
||||
lookups_per_request = len(calls)
|
||||
assert lookups_per_request == 2
|
||||
|
||||
assert run_middleware() is False
|
||||
assert len(calls) == lookups_per_request
|
||||
|
||||
temp_user_db.update_user(bob["id"], role="admin")
|
||||
assert run_middleware() is True
|
||||
|
||||
|
||||
class TestLoginLockoutRepair:
|
||||
def test_is_account_locked_repairs_missing_timestamp(self, main_module):
|
||||
|
||||
@@ -894,6 +894,35 @@ def test_sync_delivery_states_a_failure_alone_still_reopens(user_db):
|
||||
assert user_db.get_request(request["id"])["status"] == "pending"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"queue_status",
|
||||
[{}, {"queued": {}, "downloading": {}, "complete": {}, "error": {}}],
|
||||
)
|
||||
def test_sync_delivery_states_an_empty_queue_skips_the_request_history(
|
||||
user_db, monkeypatch, queue_status
|
||||
):
|
||||
user = user_db.create_user(username="alice")
|
||||
_fulfilled_with_release(user_db, user["id"], "idle")
|
||||
calls = []
|
||||
real_list_requests = user_db.list_requests
|
||||
|
||||
def counting_list_requests(**kwargs):
|
||||
calls.append(kwargs)
|
||||
return real_list_requests(**kwargs)
|
||||
|
||||
monkeypatch.setattr(user_db, "list_requests", counting_list_requests)
|
||||
|
||||
assert sync_delivery_states_from_queue_status(user_db, queue_status=queue_status) == []
|
||||
assert calls == []
|
||||
|
||||
# A task in the queue still reads the history and syncs as before.
|
||||
updated = sync_delivery_states_from_queue_status(
|
||||
user_db, queue_status={"downloading": {"idle": {"id": "idle"}}}
|
||||
)
|
||||
assert len(calls) == 1
|
||||
assert [row["delivery_state"] for row in updated] == ["downloading"]
|
||||
|
||||
|
||||
def test_create_request_rejects_non_dict_book_data(user_db):
|
||||
user = user_db.create_user(username="alice")
|
||||
with pytest.raises(RequestServiceError, match="book_data must be an object"):
|
||||
|
||||
@@ -611,6 +611,71 @@ class TestUserCRUD:
|
||||
assert "charlie" in usernames
|
||||
|
||||
|
||||
class TestCachedUserLookup:
|
||||
"""Tests for the short-lived lookup cache used by per-request auth."""
|
||||
|
||||
@pytest.fixture
|
||||
def counted(self, user_db, monkeypatch):
|
||||
calls = []
|
||||
real_get_user = user_db.get_user
|
||||
|
||||
def counting_get_user(**kwargs):
|
||||
calls.append(kwargs)
|
||||
return real_get_user(**kwargs)
|
||||
|
||||
monkeypatch.setattr(user_db, "get_user", counting_get_user)
|
||||
return calls
|
||||
|
||||
def test_repeat_lookups_are_answered_from_memory(self, user_db, counted):
|
||||
user = user_db.create_user(username="john", role="admin")
|
||||
|
||||
for _ in range(3):
|
||||
assert user_db.get_user_cached(username="john")["id"] == user["id"]
|
||||
assert user_db.get_user_cached(user_id=user["id"])["username"] == "john"
|
||||
assert user_db.get_user_cached(username="nobody") is None
|
||||
assert user_db.get_user_cached(username="nobody") is None
|
||||
|
||||
assert len(counted) == 3
|
||||
|
||||
def test_entries_expire_after_the_ttl(self, user_db, counted, monkeypatch):
|
||||
import shelfmark.core.user_db as user_db_module
|
||||
|
||||
user_db.create_user(username="john")
|
||||
now = [1000.0]
|
||||
monkeypatch.setattr(user_db_module.time, "monotonic", lambda: now[0])
|
||||
|
||||
user_db.get_user_cached(username="john")
|
||||
now[0] += user_db_module.USER_CACHE_TTL_SECONDS - 0.1
|
||||
user_db.get_user_cached(username="john")
|
||||
assert len(counted) == 1
|
||||
|
||||
now[0] += 0.2
|
||||
user_db.get_user_cached(username="john")
|
||||
assert len(counted) == 2
|
||||
|
||||
def test_writes_through_the_db_drop_cached_entries(self, user_db):
|
||||
user = user_db.create_user(username="john", role="admin")
|
||||
assert user_db.get_user_cached(username="john")["role"] == "admin"
|
||||
assert user_db.get_user_cached(username="jane") is None
|
||||
|
||||
user_db.update_user(user["id"], role="user")
|
||||
assert user_db.get_user_cached(username="john")["role"] == "user"
|
||||
|
||||
user_db.create_user(username="jane")
|
||||
assert user_db.get_user_cached(username="jane") is not None
|
||||
|
||||
user_db.delete_user(user["id"])
|
||||
assert user_db.get_user_cached(username="john") is None
|
||||
assert user_db.get_user_cached(user_id=user["id"]) is None
|
||||
|
||||
def test_callers_cannot_change_the_cached_row(self, user_db):
|
||||
user_db.create_user(username="john", role="user")
|
||||
|
||||
user_db.get_user_cached(username="john")["role"] = "admin"
|
||||
|
||||
assert user_db.get_user_cached(username="john")["role"] == "user"
|
||||
|
||||
|
||||
class TestUserSettings:
|
||||
"""Tests for per-user settings."""
|
||||
|
||||
|
||||
Reference in new issue
Block a user