Merge pull request 'fix(cache): свипер медиа-кеша — безусловный DB-diff + интервал в конфиг + дедуп очереди' (#45) from fix/25-media-sweeper into main

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)
 

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.

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,
     }

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()