abstracted out bounded cycles

src/monitor_utils/runner.py

@@ -15,8 +15,15 @@ from monitor_utils.config import (
     is_production,
     set_logger_levels,
 )
-from monitor_utils.constants import EXPECTED_TASK_ERRORS, LOGGING_LEVEL
+from monitor_utils.constants import (
+    CYCLE_TIMEOUT_SECONDS,
+    EXPECTED_TASK_ERRORS,
+    LOGGING_LEVEL,
+    MAX_CURL_HANDLES,
+    TIMEOUT_SECONDS,
+)
 from monitor_utils.database import get_collection
+from monitor_utils.sessions import get_default_session
 from monitor_utils.models import (
     AvailabilityInfo,
     CycleResult,
@@ -233,8 +240,10 @@ async def run_monitor_cycle(
     concurrently, logs expected failures, and raises unexpected ones. The
     caller owns the loop, the session lifetime, and the inter-cycle delay: wrap
     this in a `while True` that opens a fresh session per cycle and sleeps
-    between cycles. The returned CycleResult is pure counts for the caller to
-    log or act on; this function applies no policy to it.
+    between cycles — or use run_bounded_cycle, which owns the session and a
+    hard cycle budget and leaves only the loop and the sleep to the caller.
+    The returned CycleResult is pure counts for the caller to log or act on;
+    this function applies no policy to it.
 
     session and collection are the two per-cycle resources, passed to every
     callback (pre_task_func, concurrent_pre_task_func, and each task_func) just
@@ -329,6 +338,69 @@ async def run_monitor_cycle(
     return CycleResult(successes=successes, failures=failures)
 
 
+async def run_bounded_cycle(
+    category_identifiers: list[str] | dict[str, str],
+    task_func: Callable[..., Awaitable[None]],
+    proxies: AioProxies,
+    collection: AsyncIOMotorCollection,
+    *,
+    session_timeout: int = TIMEOUT_SECONDS,
+    max_clients: int = MAX_CURL_HANDLES,
+    cycle_timeout: int = CYCLE_TIMEOUT_SECONDS,
+    **kwargs,
+) -> CycleResult | None:
+    """Run one monitor cycle on a fresh session under a hard time budget.
+
+    The session is opened and closed inside the call, so curl multi-handle
+    state cannot outlive one cycle: a wedged transfer dies with its session
+    instead of parking the process. The budget turns an engine-level lost
+    wakeup into a caught TimeoutError and an abandoned cycle, never a
+    permanent wedge; the session's own close is bounded by
+    get_default_session, so the exit cannot park past the budget either.
+
+    The caller still owns the loop and the inter-cycle delay:
+
+        while True:
+            await run_bounded_cycle(ids, task_func, proxies, collection)
+            await asyncio.sleep(CYCLE_DELAY)
+
+    Args:
+        category_identifiers: Categories, passed through to run_monitor_cycle.
+        task_func: Awaitable invoked per category.
+        proxies: Pool bound to each cycle's fresh session.
+        collection: Mongo collection forwarded to every callback.
+        session_timeout: Per-request timeout on the cycle's session.
+        max_clients: Curl handle pool size. Keep at or above the monitor's
+            whole fan-out; see get_default_session for why.
+        cycle_timeout: Hard budget in seconds for the whole cycle.
+        **kwargs: Forwarded to run_monitor_cycle (pre_task_func,
+            concurrent_pre_task_func, and per-task kwargs).
+
+    Returns:
+        The cycle's CycleResult, or None when the budget expired and the
+        cycle was abandoned.
+    """
+
+    async with get_default_session(
+        proxies, timeout=session_timeout, max_clients=max_clients
+    ) as session:
+        try:
+            async with asyncio.timeout(cycle_timeout):
+                return await run_monitor_cycle(
+                    category_identifiers,
+                    task_func,
+                    session=session,
+                    collection=collection,
+                    **kwargs,
+                )
+        except TimeoutError:
+            LOGGER.error(
+                "cycle exceeded budget, abandoning cycle",
+                extra={"timeout": cycle_timeout},
+            )
+            return None
+
+
 async def execute_pagination(
     category_identifier: str,
     total_products: int,