From 6b831243ca289b7ae9323018161e4fb9a092396c Mon Sep 17 00:00:00 2001 From: g0ldyy <153996346+g0ldyy@users.noreply.github.com> Date: Sun, 4 Jan 2026 21:47:51 +0100 Subject: [PATCH 1/3] feat: introduce `update_interval` for torrent upsert logic and implement batched upsert for SQLite --- comet/services/torrent_manager.py | 86 +++++++++++++++++++++++++++++-- 1 file changed, 81 insertions(+), 5 deletions(-) diff --git a/comet/services/torrent_manager.py b/comet/services/torrent_manager.py index cecbe53..67ead77 100644 --- a/comet/services/torrent_manager.py +++ b/comet/services/torrent_manager.py @@ -401,6 +401,9 @@ class TorrentUpdateQueue: "media_id": media_id, } + if settings.DATABASE_TYPE == "postgresql": + params["update_interval"] = UPDATE_INTERVAL + params["lock_key"] = _compute_advisory_lock_key( media_id, file_info["info_hash"], @@ -426,7 +429,6 @@ class TorrentUpdateQueue: ) await self._check_batch_size() - except Exception as e: logger.warning(f"Error processing file info: {e}") finally: @@ -479,7 +481,21 @@ POSTGRES_UPDATE_SET = """ sources = EXCLUDED.sources, parsed = EXCLUDED.parsed, timestamp = EXCLUDED.timestamp - WHERE COALESCE(torrents.timestamp, 0) < EXCLUDED.timestamp + WHERE + ( + torrents.media_id IS DISTINCT FROM EXCLUDED.media_id OR + torrents.file_index IS DISTINCT FROM EXCLUDED.file_index OR + torrents.title IS DISTINCT FROM EXCLUDED.title OR + torrents.seeders IS DISTINCT FROM EXCLUDED.seeders OR + torrents.size IS DISTINCT FROM EXCLUDED.size OR + torrents.tracker IS DISTINCT FROM EXCLUDED.tracker OR + torrents.sources IS DISTINCT FROM EXCLUDED.sources OR + torrents.parsed IS DISTINCT FROM EXCLUDED.parsed + ) + OR + ( + COALESCE(torrents.timestamp, 0) < (EXCLUDED.timestamp - :update_interval) + ) """ POSTGRES_LOCK_TIMEOUT = "750ms" @@ -517,10 +533,62 @@ def _compute_advisory_lock_key(media_id, info_hash, season, episode) -> int: return int.from_bytes(digest[:8], byteorder="big", signed=True) +async def _execute_sqlite_batched_upsert(rows: list[dict]): + if not rows: + return + + keys_to_ignore = {"lock_key", "update_interval"} + columns = [k for k in rows[0].keys() if k not in keys_to_ignore] + + sanitized_rows = [{k: row[k] for k in columns} for row in rows] + + placeholders = ", ".join(f":{col}" for col in columns) + col_names = ", ".join(columns) + + async with database.transaction(): + await database.execute("DROP TABLE IF EXISTS temp_batch_torrents") + await database.execute( + "CREATE TEMP TABLE temp_batch_torrents AS SELECT * FROM torrents WHERE 0" + ) + + await database.execute_many( + f"INSERT INTO temp_batch_torrents ({col_names}) VALUES ({placeholders})", + sanitized_rows, + ) + + check_cols = [c for c in columns if c != "timestamp"] + conditions = " AND ".join( + f"t.{col} IS temp_batch_torrents.{col}" for col in check_cols + ) + + # Remove records from the batch that: + # 1. Already exist in the DB (based on all non-timestamp fields) + # 2. AND are recent enough (timestampDiff < UPDATE_INTERVAL) + prune_query = f""" + DELETE FROM temp_batch_torrents + WHERE EXISTS ( + SELECT 1 FROM torrents t + WHERE + {conditions} AND + t.timestamp >= (temp_batch_torrents.timestamp - :update_interval) + ) + """ + + await database.execute(prune_query, {"update_interval": UPDATE_INTERVAL}) + await database.execute( + "INSERT OR REPLACE INTO torrents SELECT * FROM temp_batch_torrents" + ) + await database.execute("DROP TABLE temp_batch_torrents") + + async def _execute_batched_upsert(query: str, rows): if not rows: return + if settings.DATABASE_TYPE == "sqlite": + await _execute_sqlite_batched_upsert(rows) + return + ordered_rows = sorted(rows, key=lambda row: row.get("lock_key")) sanitized_rows = [ @@ -579,13 +647,21 @@ def _get_torrent_upsert_query(conflict_key: str) -> str: return TORRENT_INSERT_TEMPLATE +UPDATE_INTERVAL = 31536000 # Default 1 year +if settings.LIVE_TORRENT_CACHE_TTL >= 0: + UPDATE_INTERVAL = max(60, settings.LIVE_TORRENT_CACHE_TTL // 2) + + async def _upsert_torrent_record(params: dict): + if settings.DATABASE_TYPE == "sqlite": + await _execute_sqlite_batched_upsert([params]) + return + query = _get_torrent_upsert_query( _determine_conflict_key(params.get("season"), params.get("episode")) ) - if settings.DATABASE_TYPE != "postgresql": - await database.execute(query, params) - return + + params["update_interval"] = UPDATE_INTERVAL for attempt in range(POSTGRES_LOCK_RETRY_ATTEMPTS + 1): try: From 5ff5da612965e535acabf05d325833d483b3c47b Mon Sep 17 00:00:00 2001 From: g0ldyy <153996346+g0ldyy@users.noreply.github.com> Date: Sun, 4 Jan 2026 22:02:41 +0100 Subject: [PATCH 2/3] Refactor: optimize torrent sqlite upsert by pre-fetching existing records and performing in-memory change detection before `INSERT OR REPLACE` --- comet/services/torrent_manager.py | 126 +++++++++++++++++++++--------- 1 file changed, 89 insertions(+), 37 deletions(-) diff --git a/comet/services/torrent_manager.py b/comet/services/torrent_manager.py index 67ead77..c13d025 100644 --- a/comet/services/torrent_manager.py +++ b/comet/services/torrent_manager.py @@ -537,48 +537,100 @@ async def _execute_sqlite_batched_upsert(rows: list[dict]): if not rows: return + # Fetch relevant existing records to compare against + media_ids = list({row["media_id"] for row in rows if row.get("media_id")}) + if not media_ids: + async with database.transaction(): + await _execute_standard_sqlite_insert(rows) + return + + placeholders = ",".join(f":mid{i}" for i in range(len(media_ids))) + params = {f"mid{i}": mid for i, mid in enumerate(media_ids)} + + existing_rows = await database.fetch_all( + f"SELECT * FROM torrents WHERE media_id IN ({placeholders})", params + ) + + # Build lookup map: (media_id, info_hash, season, episode) -> row + existing_map = {} + for row in existing_rows: + key = ( + row["media_id"], + row["info_hash"], + row["season"], + row["episode"], + ) + existing_map[key] = row + + # Filter in Python + to_insert = [] + # Columns to check for changes (everything except timestamp and lock_key/update_interval) + check_cols = [ + "file_index", + "title", + "seeders", + "size", + "tracker", + "sources", + "parsed", + ] + + for row in rows: + key = ( + row["media_id"], + row["info_hash"], + row["season"], + row["episode"], + ) + existing = existing_map.get(key) + + if not existing: + # New record + to_insert.append(row) + continue + + # Check for data changes + changed = False + for col in check_cols: + # Simple equality check. + if row.get(col) != existing[col]: + changed = True + break + + if changed: + to_insert.append(row) + continue + + # Check TTL + existing_ts = existing["timestamp"] + if existing_ts < (row["timestamp"] - UPDATE_INTERVAL): + to_insert.append(row) + + if to_insert: + await _execute_standard_sqlite_insert(to_insert) + + +async def _execute_standard_sqlite_insert(rows: list[dict]): keys_to_ignore = {"lock_key", "update_interval"} columns = [k for k in rows[0].keys() if k not in keys_to_ignore] - sanitized_rows = [{k: row[k] for k in columns} for row in rows] - placeholders = ", ".join(f":{col}" for col in columns) - col_names = ", ".join(columns) + query = f""" + INSERT OR REPLACE INTO torrents ({", ".join(columns)}) + VALUES ({", ".join(f":{col}" for col in columns)}) + """ - async with database.transaction(): - await database.execute("DROP TABLE IF EXISTS temp_batch_torrents") - await database.execute( - "CREATE TEMP TABLE temp_batch_torrents AS SELECT * FROM torrents WHERE 0" - ) - - await database.execute_many( - f"INSERT INTO temp_batch_torrents ({col_names}) VALUES ({placeholders})", - sanitized_rows, - ) - - check_cols = [c for c in columns if c != "timestamp"] - conditions = " AND ".join( - f"t.{col} IS temp_batch_torrents.{col}" for col in check_cols - ) - - # Remove records from the batch that: - # 1. Already exist in the DB (based on all non-timestamp fields) - # 2. AND are recent enough (timestampDiff < UPDATE_INTERVAL) - prune_query = f""" - DELETE FROM temp_batch_torrents - WHERE EXISTS ( - SELECT 1 FROM torrents t - WHERE - {conditions} AND - t.timestamp >= (temp_batch_torrents.timestamp - :update_interval) - ) - """ - - await database.execute(prune_query, {"update_interval": UPDATE_INTERVAL}) - await database.execute( - "INSERT OR REPLACE INTO torrents SELECT * FROM temp_batch_torrents" - ) - await database.execute("DROP TABLE temp_batch_torrents") + # Retry logic for busy database + for attempt in range(5): + try: + async with database.transaction(): + await database.execute_many(query, sanitized_rows) + return + except Exception as e: + if attempt < 4: + await asyncio.sleep(0.2 * (attempt + 1)) + continue + raise e async def _execute_batched_upsert(query: str, rows): From b1cf30b4e82f51ec6695e10efde4589979220f09 Mon Sep 17 00:00:00 2001 From: g0ldyy <153996346+g0ldyy@users.noreply.github.com> Date: Sun, 4 Jan 2026 22:08:56 +0100 Subject: [PATCH 3/3] fix: preserve original traceback when re-raising exceptions --- comet/services/torrent_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/comet/services/torrent_manager.py b/comet/services/torrent_manager.py index c13d025..81353f1 100644 --- a/comet/services/torrent_manager.py +++ b/comet/services/torrent_manager.py @@ -630,7 +630,7 @@ async def _execute_standard_sqlite_insert(rows: list[dict]): if attempt < 4: await asyncio.sleep(0.2 * (attempt + 1)) continue - raise e + raise async def _execute_batched_upsert(query: str, rows):