Finish external HTTP client resilience: add shared aiohttp config, retry support, and update task status

backend/app/config.py

@@ -60,6 +60,26 @@ class Settings(BaseSettings):
             "Ignored when session_cache_enabled is false."
         ),
     )
+    http_request_timeout_seconds: float = Field(
+        default=20.0,
+        ge=0.0,
+        description="Maximum total time in seconds for outbound external HTTP requests.",
+    )
+    http_connect_timeout_seconds: float = Field(
+        default=5.0,
+        ge=0.0,
+        description="Maximum time in seconds to establish outbound external HTTP connections.",
+    )
+    http_max_connections: int = Field(
+        default=10,
+        ge=1,
+        description="Maximum number of concurrent outbound HTTP connections.",
+    )
+    http_keepalive_timeout_seconds: float = Field(
+        default=15.0,
+        ge=0.0,
+        description="How long idle keepalive connections are retained by the HTTP connector.",
+    )
     timezone: str = Field(
         default="UTC",
         description="IANA timezone name used when displaying timestamps in the UI.",

backend/app/services/blocklist_service.py

@@ -14,11 +14,13 @@ under the key ``"blocklist_schedule"``.
 
 from __future__ import annotations
 
+import asyncio
 import importlib
 import json
 from collections.abc import Awaitable
 from typing import TYPE_CHECKING
 
+import aiohttp
 import structlog
 
 from app.models.blocklist import (
@@ -57,6 +59,55 @@ _PREVIEW_LINES: int = 20
 #: Maximum bytes to download for a preview (first 64 KB).
 _PREVIEW_MAX_BYTES: int = 65536
 
+#: HTTP status codes that should be retried for blocklist downloads.
+_BLOCKLIST_HTTP_RETRY_STATUSES: frozenset[int] = frozenset({429, 500, 502, 503, 504})
+#: How many attempts to make for transient blocklist download failures.
+_BLOCKLIST_HTTP_RETRY_ATTEMPTS: int = 2
+#: Base backoff in seconds used between retry attempts.
+_BLOCKLIST_HTTP_BACKOFF_BASE_SECONDS: float = 1.0
+
+
+async def _download_text_with_retries(
+    http_session: aiohttp.ClientSession,
+    url: str,
+    timeout: aiohttp.ClientTimeout,
+) -> tuple[int, str]:
+    """Download text from *url* with a small retry policy for transient failures."""
+    last_exception: Exception | None = None
+
+    for attempt in range(1, _BLOCKLIST_HTTP_RETRY_ATTEMPTS + 1):
+        try:
+            async with http_session.get(url, timeout=timeout) as resp:
+                text = await resp.text(errors="replace")
+                if resp.status in _BLOCKLIST_HTTP_RETRY_STATUSES and attempt < _BLOCKLIST_HTTP_RETRY_ATTEMPTS:
+                    backoff = _BLOCKLIST_HTTP_BACKOFF_BASE_SECONDS * (2 ** (attempt - 1))
+                    log.warning(
+                        "blocklist_download_retry",
+                        url=url,
+                        status=resp.status,
+                        attempt=attempt,
+                        backoff=backoff,
+                    )
+                    await asyncio.sleep(backoff)
+                    continue
+                return resp.status, text
+        except Exception as exc:  # noqa: BLE001
+            last_exception = exc
+            if attempt >= _BLOCKLIST_HTTP_RETRY_ATTEMPTS:
+                raise
+            backoff = _BLOCKLIST_HTTP_BACKOFF_BASE_SECONDS * (2 ** (attempt - 1))
+            log.warning(
+                "blocklist_download_retry_error",
+                url=url,
+                attempt=attempt,
+                error=repr(exc),
+                backoff=backoff,
+            )
+            await asyncio.sleep(backoff)
+
+    assert last_exception is not None
+    raise last_exception
+
 
 # ---------------------------------------------------------------------------
 # Source CRUD helpers
@@ -203,15 +254,18 @@ async def preview_source(
         ValueError: If the URL cannot be reached or returns a non-200 status.
     """
     try:
-        async with http_session.get(url, timeout=_aiohttp_timeout(10)) as resp:
-            if resp.status != 200:
-                raise ValueError(f"HTTP {resp.status} from {url}")
-            raw = await resp.content.read(_PREVIEW_MAX_BYTES)
+        status, raw = await _download_text_with_retries(
+            http_session,
+            url,
+            _aiohttp_timeout(10),
+        )
+        if status != 200:
+            raise ValueError(f"HTTP {status} from {url}")
     except Exception as exc:
         log.warning("blocklist_preview_failed", url=url, error=str(exc))
         raise ValueError(str(exc)) from exc
 
-    lines = raw.decode(errors="replace").splitlines()
+    lines = raw.splitlines()
     entries: list[str] = []
     valid = 0
     skipped = 0
@@ -272,21 +326,22 @@ async def import_source(
     """
     # --- Download ---
     try:
-        async with http_session.get(
-            source.url, timeout=_aiohttp_timeout(30)
-        ) as resp:
-            if resp.status != 200:
-                error_msg = f"HTTP {resp.status}"
-                await _log_result(db, source, 0, 0, error_msg)
-                log.warning("blocklist_import_download_failed", url=source.url, status=resp.status)
-                return ImportSourceResult(
-                    source_id=source.id,
-                    source_url=source.url,
-                    ips_imported=0,
-                    ips_skipped=0,
-                    error=error_msg,
-                )
-            content = await resp.text(errors="replace")
+        status, content = await _download_text_with_retries(
+            http_session,
+            source.url,
+            _aiohttp_timeout(30),
+        )
+        if status != 200:
+            error_msg = f"HTTP {status}"
+            await _log_result(db, source, 0, 0, error_msg)
+            log.warning("blocklist_import_download_failed", url=source.url, status=status)
+            return ImportSourceResult(
+                source_id=source.id,
+                source_url=source.url,
+                ips_imported=0,
+                ips_skipped=0,
+                error=error_msg,
+            )
     except Exception as exc:
         error_msg = str(exc)
         await _log_result(db, source, 0, 0, error_msg)

backend/app/startup.py

@@ -33,6 +33,22 @@ async def _ensure_database_schema(database_path: str) -> None:
         await db.close()
 
 
+def _create_http_session(settings: Settings) -> aiohttp.ClientSession:
+    """Build a shared aiohttp session with reasonable global limits and timeouts."""
+    timeout = aiohttp.ClientTimeout(
+        total=settings.http_request_timeout_seconds,
+        connect=settings.http_connect_timeout_seconds,
+        sock_read=settings.http_request_timeout_seconds,
+    )
+    connector = aiohttp.TCPConnector(
+        limit=settings.http_max_connections,
+        limit_per_host=settings.http_max_connections,
+        keepalive_timeout=settings.http_keepalive_timeout_seconds,
+        enable_cleanup_closed=True,
+    )
+    return aiohttp.ClientSession(timeout=timeout, connector=connector)
+
+
 async def startup_shared_resources(
     app: FastAPI,
     settings: Settings,
@@ -82,7 +98,7 @@ async def startup_shared_resources(
     if unresolved_count > 0:
         log.warning("geo_cache_unresolved_ips", unresolved=unresolved_count)
 
-    http_session: aiohttp.ClientSession = aiohttp.ClientSession()
+    http_session: aiohttp.ClientSession = _create_http_session(settings)
     geo_service.init_geoip(settings.geoip_db_path)
 
     scheduler: AsyncIOScheduler = AsyncIOScheduler(timezone="UTC")