fix(state): cap retry attempts for pending downloads

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

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

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:

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."""