diff --git a/api_server.py b/api_server.py index 9ace269..94ecca6 100644 --- a/api_server.py +++ b/api_server.py @@ -93,6 +93,14 @@ Config = get_settings() HTTP_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(3) # semaphore for live HTTP media requests BACKGROUND_DOWNLOAD_SEMAPHORE = asyncio.Semaphore(2) # semaphore for background cache worker download_queue = asyncio.Queue(maxsize=100) +# Keys currently enqueued for (or being processed by) the background download worker. +# download_new_files re-scans the whole media table every sweep and, under a FloodWait, +# the worker drains slowly — without this guard the SAME not-yet-downloaded file gets +# re-enqueued on every pass, flooding the queue with duplicates. The event loop is +# single-threaded and every check/mutation below runs with no await in between, so a plain +# set is race-free: download_new_files adds the key before put_nowait; the worker discards +# it in its finally alongside task_done(). +_queued_media: set[tuple[str, int, str]] = set() # --- Failing-download backoff (negative cache) -------------------------------- # A media file whose download keeps failing (hang/timeout/not-found) must not be # retried on every 60s cache sweep, nor keep occupying a scarce Pyrogram @@ -850,14 +858,22 @@ async def download_new_files(media_files: list, cache_dir: str) -> None: cache_path = os.path.join(post_dir, file_unique_id) if not os.path.exists(cache_path): + # Skip files already queued/in-flight so a slow-draining queue (FloodWait) + # isn't refilled with duplicates on every sweep. Same key shape as the + # worker's bg_key. No await between this check and the add -> race-free. + key = (str(channel), int(post_id), file_unique_id) + if key in _queued_media: + continue try: # put_nowait so a full queue raises QueueFull instead of blocking # cache_media_files (and thus the sweeper) forever. `await put()` # never raises QueueFull, which made the except below dead code. + _queued_media.add(key) download_queue.put_nowait((channel, post_id, file_unique_id)) files_queued += 1 logger.debug(f"Queued for background download: {channel}/{post_id}/{file_unique_id}") except asyncio.QueueFull: + _queued_media.discard(key) logger.warning(f"Download queue is full, skipping {channel}/{post_id}/{file_unique_id}") break @@ -893,11 +909,15 @@ async def background_download_worker(): _record_download_failure(bg_key) logger.error(f"Background download error for {channel}/{post_id}/{file_unique_id}: {e}") finally: + # Free the dedup slot so this file can be re-enqueued by a later sweep if it is + # still missing (e.g. the download failed or was throttled). No await here, so the + # discard and task_done() happen atomically w.r.t. download_new_files. + _queued_media.discard(bg_key) download_queue.task_done() async def cache_media_files() -> None: """Background task for cache management: removes old files and downloads new ones""" - delay = 60 + delay = Config["cache_sweep_interval"] while True: try: # Load all media file ID records from SQLite @@ -906,23 +926,29 @@ async def cache_media_files() -> None: cache_dir = os.path.abspath("./data/cache") updated_media_files, files_removed = await asyncio.to_thread(remove_old_cached_files_sync, media_files, cache_dir) - if files_removed > 0: - try: - # Determine which entries were removed and delete them from SQLite - updated_set = { - (r['channel'], r['post_id'], r['file_unique_id']) - for r in updated_media_files - } - removed_entries = [ - (r['channel'], r['post_id'], r['file_unique_id']) - for r in media_files - if (r['channel'], r['post_id'], r['file_unique_id']) not in updated_set - ] - if removed_entries: - await asyncio.to_thread(remove_media_file_ids_sync, DB_PATH, removed_entries) - logger.info(f"Removed {files_removed} old files from cache") - except Exception as e: - logger.error(f"Failed to remove old entries from SQLite: {str(e)}") + try: + # Compute the DB diff UNCONDITIONALLY (not only when files_removed > 0): + # remove_old_cached_files_sync drops an entry older than 20 days from the + # surviving list even when its file was already gone from disk (that branch + # does NOT bump files_removed). If we only purged when a real file was removed, + # such a fileless-but-expired row would stay in SQLite forever. The surviving + # set reflects what is actually left after the sweep; removed_entries are the + # rows present in the OLD set but not in the surviving set. + updated_set = { + (r['channel'], r['post_id'], r['file_unique_id']) + for r in updated_media_files + } + removed_entries = [ + (r['channel'], r['post_id'], r['file_unique_id']) + for r in media_files + if (r['channel'], r['post_id'], r['file_unique_id']) not in updated_set + ] + if removed_entries: + await asyncio.to_thread(remove_media_file_ids_sync, DB_PATH, removed_entries) + logger.info(f"cache_sweep: purged {len(removed_entries)} entries " + f"({files_removed} files removed from disk)") + except Exception as e: + logger.error(f"Failed to remove old entries from SQLite: {str(e)}") await download_new_files(updated_media_files, cache_dir) diff --git a/config.py b/config.py index 8e73643..a6d8411 100644 --- a/config.py +++ b/config.py @@ -140,6 +140,12 @@ def get_settings() -> dict[str, Any]: "media_download_timeout_min": _parse_int_env("MEDIA_DOWNLOAD_TIMEOUT_MIN", 120), "media_download_timeout_max": _parse_int_env("MEDIA_DOWNLOAD_TIMEOUT_MAX", 1800), "media_download_min_speed": _parse_int_env("MEDIA_DOWNLOAD_MIN_SPEED", 256 * 1024), + # Interval (seconds) between cache-sweep passes in cache_media_files. Each pass is a + # full media_file_ids table scan plus an os.walk of the whole cache tree, so running + # it every 60s under a "20 days"/"1 hour" retention policy is wasteful. A value below + # the 60s floor is rejected at startup (_parse_int_env exits) so a misconfiguration + # can never turn the sweeper into a hot loop. + "cache_sweep_interval": _parse_int_env("CACHE_SWEEP_INTERVAL", 900, minimum=60), # Size of the asyncio default ThreadPoolExecutor. SQLite/python-magic/pickle/os.walk # all run via asyncio.to_thread; the interpreter default (min(32, cpu+4)) is only 5-6 # on a 1-2 CPU container, which starves those under load. 32 gives ample headroom. diff --git a/tests/mock_config.py b/tests/mock_config.py index beeca8a..ed5ce2a 100644 --- a/tests/mock_config.py +++ b/tests/mock_config.py @@ -40,4 +40,5 @@ def get_settings(): "io_thread_pool_size": 32, "tg_max_concurrent_transmissions": 3, "media_timeout_restart_threshold": 5, + "cache_sweep_interval": 900, } diff --git a/tests/test_media_sweeper.py b/tests/test_media_sweeper.py new file mode 100644 index 0000000..02affbd --- /dev/null +++ b/tests/test_media_sweeper.py @@ -0,0 +1,120 @@ +# flake8: noqa +# pylint: disable=protected-access, missing-function-docstring, redefined-outer-name, line-too-long +# pylance: disable=reportMissingImports, reportMissingModuleSource +""" +Issue #25 (package D) regressions: + + * Задание 12 — the cache sweep must purge a >20-day-old SQLite row whose file is already + gone from disk EVEN when nothing was removed from disk in that pass (files_removed == 0). + Previously the DB diff ran only `if files_removed > 0`, so such a fileless-but-expired + row stayed in the table forever. + * Задание 14 — download_new_files must not re-enqueue an already-queued/in-flight file on + every sweep pass (dedup via the module-level _queued_media set), and the worker must free + the dedup slot in its finally so the file can be re-enqueued later. +""" +import sqlite3 +from datetime import datetime + +import pytest + +import api_server +from file_io import init_db_sync, get_all_media_file_ids_sync + + +class _StopLoop(Exception): + """Sentinel used to break the otherwise-infinite background loops after one iteration.""" + + +# --------------------------------------------------------------------------- # +# Задание 12 — expired row with a missing file is purged when files_removed == 0 +# --------------------------------------------------------------------------- # +@pytest.mark.asyncio +async def test_expired_row_missing_file_purged_when_no_disk_removal(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + db = str(tmp_path / "sweep.db") + init_db_sync(db) + monkeypatch.setattr(api_server, "DB_PATH", db) + + # A row 21 days old whose cache file does NOT exist on disk. remove_old_cached_files_sync + # drops it from the surviving list without incrementing files_removed -> files_removed == 0. + old_ts = datetime.now().timestamp() - 21 * 24 * 3600 + conn = sqlite3.connect(db) + conn.execute( + "INSERT INTO media_file_ids (channel, post_id, file_unique_id, added) VALUES (?,?,?,?)", + ("ghostchan", 42, "ghostfid", old_ts), + ) + conn.commit() + conn.close() + + # Run exactly one sweep iteration, then break out of the while-True loop. + async def fake_sleep(_delay): + raise _StopLoop + monkeypatch.setattr(api_server.asyncio, "sleep", fake_sleep) + + with pytest.raises(_StopLoop): + await api_server.cache_media_files() + + # The expired, fileless row is gone despite no file having been removed from disk. + assert get_all_media_file_ids_sync(db) == [] + + +# --------------------------------------------------------------------------- # +# Задание 14 — dedup: two identical passes enqueue exactly one item +# --------------------------------------------------------------------------- # +@pytest.mark.asyncio +async def test_download_new_files_dedups_across_passes(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + cache_dir = str(tmp_path / "cache") + + api_server._queued_media.clear() + while not api_server.download_queue.empty(): + api_server.download_queue.get_nowait() + + media_files = [{"channel": "dupchan", "post_id": 7, "file_unique_id": "dupfid"}] + + await api_server.download_new_files(media_files, cache_dir) + await api_server.download_new_files(media_files, cache_dir) + + assert api_server.download_queue.qsize() == 1 + assert ("dupchan", 7, "dupfid") in api_server._queued_media + + # Cleanup shared module state. + api_server._queued_media.clear() + while not api_server.download_queue.empty(): + api_server.download_queue.get_nowait() + + +# --------------------------------------------------------------------------- # +# Задание 14 — the worker frees the dedup slot so the file can be re-enqueued later +# --------------------------------------------------------------------------- # +@pytest.mark.asyncio +async def test_worker_clears_queued_key_after_processing(monkeypatch): + key = ("wchan", 3, "wfid") + api_server._queued_media.clear() + api_server._queued_media.add(key) + while not api_server.download_queue.empty(): + api_server.download_queue.get_nowait() + api_server.download_queue.put_nowait(("wchan", 3, "wfid")) + + async def fake_download(channel, post_id, file_unique_id): + return ("/x", False) + monkeypatch.setattr(api_server, "download_media_file", fake_download) + + async def fake_sleep(_delay): + return None + monkeypatch.setattr(api_server.asyncio, "sleep", fake_sleep) + + # Break the while-True worker loop after it has run its finally exactly once. + real_task_done = api_server.download_queue.task_done + def stop_task_done(): + real_task_done() + raise _StopLoop + monkeypatch.setattr(api_server.download_queue, "task_done", stop_task_done) + + with pytest.raises(_StopLoop): + await api_server.background_download_worker() + + # Slot freed -> a later sweep may re-enqueue the (still missing) file. + assert key not in api_server._queued_media + + api_server._queued_media.clear()