diff --git a/src/config/schema.py b/src/config/schema.py index c6e3f66..414af34 100644 --- a/src/config/schema.py +++ b/src/config/schema.py @@ -113,6 +113,10 @@ class Config(BaseModel): # Download tracking track_downloads: bool = Field(default=True) + # Give up on a pending message after this many failed download attempts + # (each attempt already includes the per-download retry budget) + max_download_attempts: int = Field(default=5, ge=1) + # Logging verbosity: Literal["quiet", "normal", "verbose"] = Field(default="normal") diff --git a/src/main.py b/src/main.py index d2e4c74..a7ab511 100644 --- a/src/main.py +++ b/src/main.py @@ -328,7 +328,11 @@ async def run_check(cfg, log, state_store, client, sources_with_filters, history total_downloaded = 0 while True: - batch_ids = pending.get_oldest(cursor_key, batch_size) + # Exclude messages that exhausted their attempt budget so one + # permanently failing message can't loop this check forever + batch_ids = pending.get_oldest( + cursor_key, batch_size, cfg.max_download_attempts + ) if not batch_ids: break @@ -367,12 +371,28 @@ async def run_check(cfg, log, state_store, client, sources_with_filters, history pending.remove_batch(cursor_key, list(remove_ids)) total_downloaded += downloaded + # Track failures so repeatedly failing messages are eventually + # dead-lettered instead of retried indefinitely + if failed_ids: + pending.increment_attempts(cursor_key, list(failed_ids)) + log.warning( + f"{len(failed_ids)} download(s) failed, will retry " + f"(up to {cfg.max_download_attempts} attempts per message)" + ) + if max_downloads: break # Capped mode: one batch only - remaining = pending.count(cursor_key) + remaining = pending.count(cursor_key, cfg.max_download_attempts) if remaining: log.info(f"{remaining} message(s) still pending for next run") + exhausted = pending.count_exhausted(cursor_key, cfg.max_download_attempts) + if exhausted: + log.error( + f"{exhausted} message(s) gave up after {cfg.max_download_attempts} " + f"failed attempts and will not be retried (kept in pending table " + f"for inspection)" + ) log.info(f"Downloaded {total_downloaded} files") diff --git a/src/state/pending.py b/src/state/pending.py index 3c63e9b..22b603b 100644 --- a/src/state/pending.py +++ b/src/state/pending.py @@ -56,9 +56,17 @@ def __init__(self, db_path: str): cursor_key TEXT NOT NULL, message_id INTEGER NOT NULL, added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + attempts INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (cursor_key, message_id) ) """) + # Migrate databases created before the attempts column existed + try: + self._connection.execute( + "ALTER TABLE pending_downloads ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0" + ) + except sqlite3.OperationalError: + pass # Column already exists self._connection.commit() def _check_connection(self) -> None: @@ -106,12 +114,17 @@ def add_batch(self, cursor_key: str, message_ids: list[int]) -> None: raise StateError(f"Failed to add batch after {max_retries} retries (database locked)") - def get_oldest(self, cursor_key: str, limit: int) -> list[int]: + def get_oldest( + self, cursor_key: str, limit: int, max_attempts: Optional[int] = None + ) -> list[int]: """Get the oldest pending message IDs for a source. Args: cursor_key: Source cursor key limit: Maximum number of IDs to return + max_attempts: If set, exclude entries with attempts >= max_attempts + (exhausted entries stay in the table for inspection but are + no longer scheduled for download) Returns: List of message IDs ordered ascending (oldest first) @@ -120,12 +133,78 @@ def get_oldest(self, cursor_key: str, limit: int) -> list[int]: StateError: If database operation fails """ self._check_connection() + if max_attempts is not None: + cursor = self._connection.execute( + "SELECT message_id FROM pending_downloads " + "WHERE cursor_key = ? AND attempts < ? " + "ORDER BY message_id ASC LIMIT ?", + (cursor_key, max_attempts, limit), + ) + else: + cursor = self._connection.execute( + "SELECT message_id FROM pending_downloads " + "WHERE cursor_key = ? ORDER BY message_id ASC LIMIT ?", + (cursor_key, limit), + ) + return [row[0] for row in cursor.fetchall()] + + def increment_attempts(self, cursor_key: str, message_ids: list[int]) -> None: + """Increment the attempt counter for failed downloads. + + Args: + cursor_key: Source cursor key + message_ids: List of message IDs that failed to download + + Raises: + StateError: If database operation fails + """ + self._check_connection() + if not message_ids: + return + + max_retries = 3 + delays = [0.1, 0.2, 0.4] + + for attempt in range(max_retries): + try: + self._connection.execute("BEGIN") + placeholders = ",".join("?" * len(message_ids)) + self._connection.execute( + f"UPDATE pending_downloads SET attempts = attempts + 1 " + f"WHERE cursor_key = ? AND message_id IN ({placeholders})", + [cursor_key] + list(message_ids), + ) + self._connection.commit() + return + except sqlite3.OperationalError as e: + self._connection.rollback() + if "database is locked" in str(e).lower() and attempt < max_retries - 1: + time.sleep(delays[attempt]) + continue + raise StateError(f"Database operation failed: {e}") + except Exception as e: + self._connection.rollback() + raise StateError(f"Unexpected error during increment_attempts: {e}") + + raise StateError(f"Failed to increment attempts after {max_retries} retries (database locked)") + + def count_exhausted(self, cursor_key: str, max_attempts: int) -> int: + """Count entries that exceeded the attempt limit (dead-lettered). + + Args: + cursor_key: Source cursor key + max_attempts: Attempt limit + + Returns: + Number of entries with attempts >= max_attempts + """ + self._check_connection() cursor = self._connection.execute( - "SELECT message_id FROM pending_downloads " - "WHERE cursor_key = ? ORDER BY message_id ASC LIMIT ?", - (cursor_key, limit), + "SELECT COUNT(*) FROM pending_downloads " + "WHERE cursor_key = ? AND attempts >= ?", + (cursor_key, max_attempts), ) - return [row[0] for row in cursor.fetchall()] + return cursor.fetchone()[0] def remove_batch(self, cursor_key: str, message_ids: list[int]) -> None: """Remove processed message IDs from the pending queue. @@ -166,20 +245,29 @@ def remove_batch(self, cursor_key: str, message_ids: list[int]) -> None: raise StateError(f"Failed to remove batch after {max_retries} retries (database locked)") - def count(self, cursor_key: str) -> int: + def count(self, cursor_key: str, max_attempts: Optional[int] = None) -> int: """Count pending message IDs for a source. Args: cursor_key: Source cursor key + max_attempts: If set, count only entries still eligible for + download (attempts < max_attempts) Returns: Number of pending messages """ self._check_connection() - cursor = self._connection.execute( - "SELECT COUNT(*) FROM pending_downloads WHERE cursor_key = ?", - (cursor_key,), - ) + if max_attempts is not None: + cursor = self._connection.execute( + "SELECT COUNT(*) FROM pending_downloads " + "WHERE cursor_key = ? AND attempts < ?", + (cursor_key, max_attempts), + ) + else: + cursor = self._connection.execute( + "SELECT COUNT(*) FROM pending_downloads WHERE cursor_key = ?", + (cursor_key,), + ) return cursor.fetchone()[0] def has_pending(self, cursor_key: str) -> bool: diff --git a/tests/unit/state/test_pending.py b/tests/unit/state/test_pending.py index fdcc780..653ae6c 100644 --- a/tests/unit/state/test_pending.py +++ b/tests/unit/state/test_pending.py @@ -92,6 +92,84 @@ def test_empty_batch_operations(self, history_db): assert pending.count("src_1") == 0 +class TestPendingDownloadsAttempts: + """Attempt accounting and dead-lettering of repeatedly failing entries.""" + + def test_new_entries_have_zero_attempts(self, history_db): + """Fresh entries should be eligible regardless of max_attempts.""" + with PendingDownloads(history_db) as pending: + pending.add_batch("src_1", [1, 2, 3]) + assert pending.get_oldest("src_1", 10, max_attempts=1) == [1, 2, 3] + + def test_increment_attempts(self, history_db): + """increment_attempts should only affect the given IDs.""" + with PendingDownloads(history_db) as pending: + pending.add_batch("src_1", [1, 2, 3]) + pending.increment_attempts("src_1", [2]) + + # ID 2 now has 1 attempt, excluded when max_attempts=1 + assert pending.get_oldest("src_1", 10, max_attempts=1) == [1, 3] + # Still included with a higher limit + assert pending.get_oldest("src_1", 10, max_attempts=2) == [1, 2, 3] + + def test_exhausted_entries_excluded_from_get_oldest(self, history_db): + """Entries at the attempt limit must not be scheduled again.""" + with PendingDownloads(history_db) as pending: + pending.add_batch("src_1", [10, 20]) + for _ in range(3): + pending.increment_attempts("src_1", [10]) + + assert pending.get_oldest("src_1", 10, max_attempts=3) == [20] + # Exhausted entry remains in the table + assert pending.count("src_1") == 2 + assert pending.count("src_1", max_attempts=3) == 1 + assert pending.count_exhausted("src_1", 3) == 1 + + def test_increment_attempts_empty_list_is_noop(self, history_db): + """Empty list should be a no-op.""" + with PendingDownloads(history_db) as pending: + pending.add_batch("src_1", [1]) + pending.increment_attempts("src_1", []) + assert pending.get_oldest("src_1", 10, max_attempts=1) == [1] + + def test_increment_attempts_isolated_per_source(self, history_db): + """Attempts on one source must not affect another.""" + with PendingDownloads(history_db) as pending: + pending.add_batch("src_a", [1]) + pending.add_batch("src_b", [1]) + pending.increment_attempts("src_a", [1]) + + assert pending.get_oldest("src_a", 10, max_attempts=1) == [] + assert pending.get_oldest("src_b", 10, max_attempts=1) == [1] + + def test_migrates_legacy_table_without_attempts_column(self, history_db): + """Databases created before the attempts column should be upgraded.""" + import sqlite3 + + # Simulate a legacy database (no attempts column) + conn = sqlite3.connect(history_db) + conn.execute(""" + CREATE TABLE pending_downloads ( + cursor_key TEXT NOT NULL, + message_id INTEGER NOT NULL, + added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (cursor_key, message_id) + ) + """) + conn.execute( + "INSERT INTO pending_downloads (cursor_key, message_id) VALUES (?, ?)", + ("src_1", 42), + ) + conn.commit() + conn.close() + + with PendingDownloads(history_db) as pending: + # Legacy rows get attempts=0 and stay eligible + assert pending.get_oldest("src_1", 10, max_attempts=5) == [42] + pending.increment_attempts("src_1", [42]) + assert pending.count_exhausted("src_1", 1) == 1 + + class TestPendingDownloadsSourceIsolation: """Different cursor_keys should be independent."""