From 356be2590c2a17299b17bdde4fa1087076ff9bb3 Mon Sep 17 00:00:00 2001 From: g0ldyy <153996346+g0ldyy@users.noreply.github.com> Date: Wed, 31 Dec 2025 12:28:37 +0100 Subject: [PATCH 1/4] refactor: optimize database indexes and introduce an index migration step --- comet/core/database.py | 177 +++++++++++++++------------------- deployment/docker-compose.yml | 2 +- 2 files changed, 78 insertions(+), 101 deletions(-) diff --git a/comet/core/database.py b/comet/core/database.py index c31afa7..51811ad 100644 --- a/comet/core/database.py +++ b/comet/core/database.py @@ -21,6 +21,8 @@ async def setup_database(): await database.connect() + await _migrate_indexes() + await database.execute( """ CREATE TABLE IF NOT EXISTS db_version ( @@ -159,7 +161,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS torrents_series_both_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_torrents_series ON torrents (media_id, info_hash, season, episode) WHERE season IS NOT NULL AND episode IS NOT NULL """ @@ -167,7 +169,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS torrents_season_only_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_torrents_season ON torrents (media_id, info_hash, season) WHERE season IS NOT NULL AND episode IS NULL """ @@ -175,7 +177,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS torrents_episode_only_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_torrents_episode ON torrents (media_id, info_hash, episode) WHERE season IS NULL AND episode IS NOT NULL """ @@ -183,7 +185,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS torrents_no_season_episode_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_torrents_movie ON torrents (media_id, info_hash) WHERE season IS NULL AND episode IS NULL """ @@ -207,7 +209,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS debrid_series_both_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_debrid_series ON debrid_availability (debrid_service, info_hash, season, episode) WHERE season IS NOT NULL AND episode IS NOT NULL """ @@ -215,7 +217,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS debrid_season_only_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_debrid_season ON debrid_availability (debrid_service, info_hash, season) WHERE season IS NOT NULL AND episode IS NULL """ @@ -223,7 +225,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS debrid_episode_only_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_debrid_episode ON debrid_availability (debrid_service, info_hash, episode) WHERE season IS NULL AND episode IS NOT NULL """ @@ -231,7 +233,7 @@ async def setup_database(): await database.execute( """ - CREATE UNIQUE INDEX IF NOT EXISTS debrid_no_season_episode_idx + CREATE UNIQUE INDEX IF NOT EXISTS unq_debrid_movie ON debrid_availability (debrid_service, info_hash) WHERE season IS NULL AND episode IS NULL """ @@ -372,87 +374,42 @@ async def setup_database(): ) # ============================================================================= - # TORRENTS TABLE INDEXES - Most critical for performance + # TORRENTS TABLE INDEXES # ============================================================================= - # Primary lookup index: media_id + season + episode + timestamp (cache TTL filter) + # Primary lookup index: media_id + season + episode (nullable) + timestamp + # Covers: get_cached_torrents, check_torrents_cache await database.execute( """ - CREATE INDEX IF NOT EXISTS idx_torrents_media_cache_lookup + CREATE INDEX IF NOT EXISTS idx_torrents_lookup ON torrents (media_id, season, episode, timestamp) """ ) - # Info hash lookup for playback (very frequent) + # Cleanup index: timestamp + # Covers: DELETE WHERE timestamp + ttl < current await database.execute( """ - CREATE INDEX IF NOT EXISTS idx_torrents_info_hash - ON torrents (info_hash) - """ - ) - - # Analytics queries: tracker-based aggregation - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_torrents_tracker_analytics - ON torrents (tracker, seeders, size) - """ - ) - - # Size filtering for user preferences - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_torrents_size_filter - ON torrents (size, timestamp) - """ - ) - - # Seeders ordering for quality ranking - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_torrents_seeders_desc - ON torrents (seeders DESC, timestamp) + CREATE INDEX IF NOT EXISTS idx_torrents_timestamp + ON torrents (timestamp) """ ) # ============================================================================= - # DEBRID_AVAILABILITY TABLE INDEXES - Critical for cache performance + # DEBRID_AVAILABILITY TABLE INDEXES # ============================================================================= - # Primary cache lookup: service + info_hash list + timestamp + # Primary lookup index: service + info_hash + timestamp + # Covers: get_cached_availability (info_hash IN ...) await database.execute( """ - CREATE INDEX IF NOT EXISTS idx_debrid_service_hash_cache + CREATE INDEX IF NOT EXISTS idx_debrid_lookup ON debrid_availability (debrid_service, info_hash, timestamp) """ ) - # Season/episode filtering for series content - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_debrid_season_episode_filter - ON debrid_availability (debrid_service, season, episode, timestamp) - """ - ) - - # Service-based analytics and cleanup - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_debrid_service_timestamp - ON debrid_availability (debrid_service, timestamp) - """ - ) - - # Title filtering for OffCloud special case - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_debrid_title_filter - ON debrid_availability (debrid_service, info_hash, title, timestamp) - """ - ) - # ============================================================================= - # DOWNLOAD_LINKS_CACHE TABLE INDEXES - Playback performance + # DOWNLOAD_LINKS_CACHE TABLE INDEXES # ============================================================================= # Primary playback lookup: debrid_key + info_hash + season + episode @@ -472,7 +429,7 @@ async def setup_database(): ) # ============================================================================= - # METADATA_CACHE TABLE INDEXES - Metadata performance + # METADATA_CACHE TABLE INDEXES # ============================================================================= # Primary cache lookup: media_id + timestamp @@ -492,7 +449,7 @@ async def setup_database(): ) # ============================================================================= - # FIRST_SEARCHES TABLE INDEXES - Search optimization + # FIRST_SEARCHES TABLE INDEXES # ============================================================================= # Primary search check: media_id (already PRIMARY KEY, but explicit for clarity) @@ -505,7 +462,7 @@ async def setup_database(): ) # ============================================================================= - # ACTIVE_CONNECTIONS TABLE INDEXES - Admin dashboard performance + # ACTIVE_CONNECTIONS TABLE INDEXES # ============================================================================= # Admin dashboard ordering: timestamp DESC (most recent first) @@ -533,7 +490,7 @@ async def setup_database(): ) # ============================================================================= - # SCRAPE_LOCKS TABLE INDEXES - Lock management + # SCRAPE_LOCKS TABLE INDEXES # ============================================================================= # Expired locks cleanup: expires_at < current_time @@ -553,7 +510,7 @@ async def setup_database(): ) # ============================================================================= - # ADMIN_SESSIONS TABLE INDEXES - Authentication performance + # ADMIN_SESSIONS TABLE INDEXES # ============================================================================= # Session cleanup: expires_at < current_time @@ -565,7 +522,7 @@ async def setup_database(): ) # ============================================================================= - # BACKGROUND_SCRAPER_STATE TABLE INDEXES - Scraper performance + # BACKGROUND_SCRAPER_STATE TABLE INDEXES # ============================================================================= # Media type filtering for scraper analytics @@ -592,34 +549,6 @@ async def setup_database(): """ ) - # ============================================================================= - # COMPOSITE INDEXES FOR COMPLEX QUERIES - # ============================================================================= - - # Torrents: media + quality filtering + cache validity - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_torrents_quality_cache - ON torrents (media_id, seeders DESC, size DESC, timestamp) - """ - ) - - # Debrid: comprehensive availability lookup - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_debrid_comprehensive - ON debrid_availability (debrid_service, info_hash, season, episode, size, timestamp) - """ - ) - - # Background scraper: comprehensive state tracking - await database.execute( - """ - CREATE INDEX IF NOT EXISTS idx_scraper_comprehensive - ON background_scraper_state (media_type, total_torrents_found, scraped_at) - """ - ) - await database.execute( """ CREATE INDEX IF NOT EXISTS idx_anime_mapping_imdb @@ -801,6 +730,54 @@ async def cleanup_expired_sessions(): await asyncio.sleep(5) # Clean up every 5 seconds +async def _migrate_indexes(): + try: + if settings.DATABASE_TYPE == "sqlite": + check_query = "SELECT name FROM sqlite_master WHERE type='index' AND name='torrents_series_both_idx'" + else: + check_query = "SELECT indexname FROM pg_indexes WHERE indexname='torrents_series_both_idx'" + + exists = await database.fetch_val(check_query) + + if not exists: + return + + logger.log("COMET", "Database: Migrating indexes (dropping legacy indexes)...") + + old_indexes = [ + "torrents_series_both_idx", + "torrents_season_only_idx", + "torrents_episode_only_idx", + "torrents_no_season_episode_idx", + "idx_torrents_media_cache_lookup", + "idx_torrents_info_hash", + "idx_torrents_tracker_analytics", + "idx_torrents_size_filter", + "idx_torrents_seeders_desc", + "idx_torrents_quality_cache", + "idx_torrents_media_season_episode", + "debrid_series_both_idx", + "debrid_season_only_idx", + "debrid_episode_only_idx", + "debrid_no_season_episode_idx", + "idx_debrid_service_hash_cache", + "idx_debrid_season_episode_filter", + "idx_debrid_service_timestamp", + "idx_debrid_title_filter", + "idx_debrid_comprehensive", + "idx_debrid_info_hash_season_episode", + "idx_debrid_timestamp", + ] + + for index_name in old_indexes: + await database.execute(f"DROP INDEX IF EXISTS {index_name}") + + logger.log("COMET", "Database: Legacy indexes dropped. New indexes will be created.") + + except Exception as e: + logger.warning(f"Error during index migration: {e}") + + async def teardown_database(): try: await database.disconnect() diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index 5eef29c..ee36e56 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -51,7 +51,7 @@ services: - "-c" - "max_connections=100" volumes: - - postgres_data:/var/lib/postgresql/18/docker + - postgres_data:/var/lib/postgresql/ healthcheck: test: ["CMD-SHELL", "pg_isready -U comet -d comet"] interval: 5s From 45ff7c173abb49a050c4ba14103fe6beae845d32 Mon Sep 17 00:00:00 2001 From: g0ldyy <153996346+g0ldyy@users.noreply.github.com> Date: Wed, 31 Dec 2025 13:27:48 +0100 Subject: [PATCH 2/4] feat: improve torrent batch processing with in-memory deduplication, PostgreSQL advisory locks, and enhance metadata handling --- comet/api/endpoints/config.py | 1 + comet/api/endpoints/playback.py | 5 +- comet/core/database.py | 4 +- comet/metadata/manager.py | 3 + comet/services/torrent_manager.py | 146 ++++++++++++++++++++++++++---- comet/templates/index.html | 54 ++++++++--- comet/utils/parsing.py | 16 +++- 7 files changed, 189 insertions(+), 40 deletions(-) diff --git a/comet/api/endpoints/config.py b/comet/api/endpoints/config.py index 5948b0c..97525de 100644 --- a/comet/api/endpoints/config.py +++ b/comet/api/endpoints/config.py @@ -29,5 +29,6 @@ async def configure(request: Request): else "", "webConfig": web_config, "proxyDebridStream": settings.PROXY_DEBRID_STREAM, + "disableTorrentStreams": settings.DISABLE_TORRENT_STREAMS, }, ) diff --git a/comet/api/endpoints/playback.py b/comet/api/endpoints/playback.py index eef73b8..020b031 100644 --- a/comet/api/endpoints/playback.py +++ b/comet/api/endpoints/playback.py @@ -12,6 +12,7 @@ from comet.debrid.manager import get_debrid from comet.metadata.manager import MetadataScraper from comet.services.streaming.manager import custom_handle_stream_request from comet.utils.network import NO_CACHE_HEADERS, get_client_ip +from comet.utils.parsing import parse_optional_int router = APIRouter() @@ -34,8 +35,8 @@ async def playback( ): config = config_check(b64config) - season = int(season) if season != "n" else None - episode = int(episode) if episode != "n" else None + season = parse_optional_int(season) + episode = parse_optional_int(episode) async with aiohttp.ClientSession() as session: cached_link = await database.fetch_one( diff --git a/comet/core/database.py b/comet/core/database.py index 51811ad..a9eb5d1 100644 --- a/comet/core/database.py +++ b/comet/core/database.py @@ -772,7 +772,9 @@ async def _migrate_indexes(): for index_name in old_indexes: await database.execute(f"DROP INDEX IF EXISTS {index_name}") - logger.log("COMET", "Database: Legacy indexes dropped. New indexes will be created.") + logger.log( + "COMET", "Database: Legacy indexes dropped. New indexes will be created." + ) except Exception as e: logger.warning(f"Error during index migration: {e}") diff --git a/comet/metadata/manager.py b/comet/metadata/manager.py index 46c15a7..c9ba36d 100644 --- a/comet/metadata/manager.py +++ b/comet/metadata/manager.py @@ -83,6 +83,9 @@ class MetadataScraper: ) def normalize_metadata(self, metadata: dict, season: int, episode: int): + if not metadata: + return None + title, year, year_end = metadata if title is None: # metadata retrieving failed diff --git a/comet/services/torrent_manager.py b/comet/services/torrent_manager.py index 25f4e3f..441d95d 100644 --- a/comet/services/torrent_manager.py +++ b/comet/services/torrent_manager.py @@ -250,7 +250,7 @@ class TorrentUpdateQueue: self.batch_size = batch_size self.flush_interval = flush_interval self.is_running = False - self.batches = {"to_delete": [], "upserts": []} + self.batches = {"to_delete": set(), "upserts": {}} async def add_torrent_info(self, file_info: dict, media_id: str = None): await self.queue.put((file_info, media_id)) @@ -317,30 +317,32 @@ class TorrentUpdateQueue: await self._flush_batch() def _reset_batches(self): - for key in self.batches: - if len(self.batches[key]) > 0: + for key, batch in self.batches.items(): + if len(batch) > 0: logger.warning( - f"Ignoring {len(self.batches[key])} items in problematic '{key}' batch" + f"Ignoring {len(batch)} items in problematic '{key}' batch" ) - self.batches[key] = [] + batch.clear() async def _flush_batch(self): try: if self.batches["to_delete"]: + delete_items = list(self.batches["to_delete"]) sub_batch_size = 100 - for i in range(0, len(self.batches["to_delete"]), sub_batch_size): + for i in range(0, len(delete_items), sub_batch_size): try: - sub_batch = self.batches["to_delete"][i : i + sub_batch_size] + sub_batch = delete_items[i : i + sub_batch_size] placeholders = [] params = {} for idx, item in enumerate(sub_batch): + info_hash, season = item key_suffix = f"_{idx}" placeholders.append( f"(CAST(:info_hash{key_suffix} AS TEXT), CAST(:season{key_suffix} AS INTEGER))" ) - params[f"info_hash{key_suffix}"] = item["info_hash"] - params[f"season{key_suffix}"] = item["season"] + params[f"info_hash{key_suffix}"] = info_hash + params[f"season{key_suffix}"] = season async with database.transaction(): delete_query = f""" @@ -354,28 +356,28 @@ class TorrentUpdateQueue: except Exception as e: logger.warning(f"Error processing delete batch: {e}") - self.batches["to_delete"] = [] + self.batches["to_delete"].clear() if self.batches["upserts"]: grouped: dict[str, list[dict]] = defaultdict(list) - for params in self.batches["upserts"]: + for params in self.batches["upserts"].values(): key = _determine_conflict_key(params["season"], params["episode"]) grouped[key].append(params) for key, rows in grouped.items(): query = _get_torrent_upsert_query(key) try: - async with database.transaction(): - await database.execute_many(query, rows) + await _execute_batched_upsert(query, rows) except Exception as e: logger.warning(f"Error processing upsert batch: {e}") - if len(self.batches["upserts"]) > 0: + total_upserts = len(self.batches["upserts"]) + if total_upserts > 0: logger.log( "SCRAPER", - f"Upserted {len(self.batches['upserts'])} torrents in batch", + f"Upserted {total_upserts} torrents in batch", ) - self.batches["upserts"] = [] + self.batches["upserts"].clear() except Exception as e: logger.warning(f"Error in flush_batch: {e}") @@ -400,11 +402,28 @@ class TorrentUpdateQueue: "media_id": media_id, } - self.batches["upserts"].append(params) + params["lock_key"] = _compute_advisory_lock_key( + media_id, + file_info["info_hash"], + file_info["season"], + file_info["episode"], + ) + + upsert_key = _build_upsert_key( + file_info["info_hash"], + file_info["season"], + file_info["episode"], + media_id, + ) + + # In-memory deduplication: keep the freshest timestamp + existing = self.batches["upserts"].get(upsert_key) + if not existing or params["timestamp"] > existing["timestamp"]: + self.batches["upserts"][upsert_key] = params if file_info["episode"] is not None: - self.batches["to_delete"].append( - {"info_hash": file_info["info_hash"], "season": file_info["season"]} + self.batches["to_delete"].add( + (file_info["info_hash"], file_info["season"]) ) await self._check_batch_size() @@ -461,8 +480,13 @@ POSTGRES_UPDATE_SET = """ sources = EXCLUDED.sources, parsed = EXCLUDED.parsed, timestamp = EXCLUDED.timestamp + WHERE COALESCE(torrents.timestamp, 0) < EXCLUDED.timestamp """ +POSTGRES_LOCK_TIMEOUT = "750ms" +POSTGRES_LOCK_RETRY_ATTEMPTS = 1 +POSTGRES_RETRYABLE_SQLSTATES = {"55P03", "40P01"} + POSTGRES_CONFLICT_TARGETS = { "series": "(media_id, info_hash, season, episode) WHERE season IS NOT NULL AND episode IS NOT NULL", "season_only": "(media_id, info_hash, season) WHERE season IS NOT NULL AND episode IS NULL", @@ -483,6 +507,62 @@ def _determine_conflict_key(season, episode) -> str: return "none" +def _build_upsert_key(info_hash, season, episode, media_id): + return (media_id, info_hash, season, episode) + + +def _compute_advisory_lock_key(media_id, info_hash, season, episode) -> int: + payload = f"{media_id}|{info_hash}|{season}|{episode}".encode("utf-8") + digest = hashlib.sha1(payload).digest() + # Use signed=True to get a signed 64-bit int directly, which Postgres expects + return int.from_bytes(digest[:8], byteorder="big", signed=True) + + +async def _execute_batched_upsert(query: str, rows): + if not rows: + return + + ordered_rows = sorted(rows, key=lambda row: row.get("lock_key")) + + sanitized_rows = [ + {key: value for key, value in row.items() if key != "lock_key"} + for row in ordered_rows + ] + + attempts = ( + POSTGRES_LOCK_RETRY_ATTEMPTS + 1 + if settings.DATABASE_TYPE == "postgresql" + else 1 + ) + + for attempt in range(attempts): + try: + async with database.transaction(): + if settings.DATABASE_TYPE == "postgresql": + await database.execute( + f"SET LOCAL lock_timeout = '{POSTGRES_LOCK_TIMEOUT}'" + ) + for row in ordered_rows: + lock_key = row.get("lock_key") + if lock_key is None: + continue + await database.execute( + "SELECT pg_advisory_xact_lock(CAST(:lock_key AS BIGINT))", + {"lock_key": lock_key}, + ) + + await database.execute_many(query, sanitized_rows) + return + except Exception as exc: + if ( + settings.DATABASE_TYPE != "postgresql" + or not _is_retryable_lock_error(exc) + or attempt == attempts - 1 + ): + raise + await asyncio.sleep(0.2 * (attempt + 1)) + + def _get_torrent_upsert_query(conflict_key: str) -> str: if settings.DATABASE_TYPE == "sqlite": return SQLITE_UPSERT_QUERY @@ -504,7 +584,33 @@ async def _upsert_torrent_record(params: dict): query = _get_torrent_upsert_query( _determine_conflict_key(params.get("season"), params.get("episode")) ) - await database.execute(query, params) + if settings.DATABASE_TYPE != "postgresql": + await database.execute(query, params) + return + + for attempt in range(POSTGRES_LOCK_RETRY_ATTEMPTS + 1): + try: + async with database.transaction(): + await database.execute( + f"SET LOCAL lock_timeout = '{POSTGRES_LOCK_TIMEOUT}'" + ) + await database.execute(query, params) + return + except Exception as exc: # pragma: no cover - driver specific + if ( + not _is_retryable_lock_error(exc) + or attempt == POSTGRES_LOCK_RETRY_ATTEMPTS + ): + raise + await asyncio.sleep(0.2 * (attempt + 1)) + + +def _is_retryable_lock_error(exc: Exception) -> bool: + sqlstate = getattr(exc, "sqlstate", None) + if sqlstate in POSTGRES_RETRYABLE_SQLSTATES: + return True + message = str(exc).lower() + return "lock timeout" in message or "deadlock detected" in message torrent_update_queue = TorrentUpdateQueue() diff --git a/comet/templates/index.html b/comet/templates/index.html index cc7dd6c..c1475b8 100644 --- a/comet/templates/index.html +++ b/comet/templates/index.html @@ -409,16 +409,26 @@ help-text="Debrid Stream Proxying allows you to use your Debrid Service from multiple IPs at same time!"> + {% set default_debrid_service = 'realdebrid' if disableTorrentStreams else 'torrent' %}
- + + {% if not disableTorrentStreams %} Torrent + {% endif %} + Real-Debrid TorBox + All-Debrid + Debrid-Link + Premiumize Debrider EasyDebrid - Real-Debrid - Debrid-Link - All-Debrid - Premiumize Offcloud PikPak @@ -455,9 +465,13 @@