added bounded cycles for curl_cfii session wedges

samsung_shop/samsung_shop_monitor.py

@@ -8,12 +8,13 @@ from monitor_utils.constants import CYCLE_DELAY, EXPECTED_TASK_ERRORS
 from monitor_utils.database import add_products_to_database
 from monitor_utils.models import AvailabilityInfo, PricingInfo, Product, ProductMetadata
 from monitor_utils.runner import (
-    run_monitor_cycle,
     execute_pagination,
+    fan_out_group,
     init_monitor,
+    run_bounded_cycle,
     setup_monitor_logging,
 )
-from monitor_utils.sessions import get_default_session, send_request
+from monitor_utils.sessions import send_request
 from motor.motor_asyncio import AsyncIOMotorCollection
 
 setup_monitor_logging(monitor_name="samsung_shop")
@@ -26,20 +27,19 @@ CATEGORY_SEMAPHORE = asyncio.Semaphore(CATEGORY_CONCURRENCY)
 
 
 async def main():
-    """Run the samsung shop monitor loop over all category ids using a base session."""
+    """Run the samsung shop monitor loop over all category ids on bounded cycles."""
     category_paths, collection, proxies = await init_monitor(
         monitor_name="samsung_shop", development_mode=DEVELOPMENT_MODE
     )
-    async with get_default_session(proxies) as session:
-        while True:
-            await run_monitor_cycle(
-                category_paths,
-                task_func=get_category_products,
-                session=session,
-                collection=collection,
-            )
-            LOGGER.info("tasks completed sleeping", extra={"delay": CYCLE_DELAY})
-            await asyncio.sleep(CYCLE_DELAY)
+    while True:
+        await run_bounded_cycle(
+            category_paths,
+            get_category_products,
+            proxies,
+            collection,
+        )
+        LOGGER.info("tasks completed sleeping", extra={"delay": CYCLE_DELAY})
+        await asyncio.sleep(CYCLE_DELAY)
 
 
 async def get_category_products(
@@ -81,8 +81,10 @@ async def get_category_products(
         await process_page(data, category_id, offset, collection)
         return
 
-    page_task = asyncio.create_task(process_page(data, category_id, offset, collection))
-    try:
+    # offset zero owns the fan-out: the group binds this page's processing to the
+    # pagination it spawns, so a cancelled cycle takes both down together.
+    async with fan_out_group() as group:
+        group.create_task(process_page(data, category_id, offset, collection))
         total_products = data.get("searchTotalCount")
         if total_products is None:
             LOGGER.warning("missing searchTotalCount", extra={"category": category_id})
@@ -98,8 +100,6 @@ async def get_category_products(
             category_id=category_id,
             collection=collection,
         )
-    finally:
-        await page_task
 
 
 async def process_page(