added concurrency limiting

src/monitor_utils/runner.py

@@ -293,7 +293,7 @@ async def execute_pagination(
     func: Callable[..., Awaitable[None]],
     session: aiohttp.ClientSession | AsyncSession,
     *,
-    max_concurrent_tasks: int = 5,
+    semaphore: asyncio.Semaphore,
     start_offset: int = 0,
     pagination_mode: str = "offset",
     **kwargs,
@@ -311,7 +311,11 @@ async def execute_pagination(
         func: Awaitable task invoked per page; receives session plus either
             offset or page_number depending on pagination_mode.
         session: HTTP session forwarded to each func call.
-        max_concurrent_tasks: Semaphore limit for concurrent page fetches.
+        max_concurrent_tasks: Semaphore limit for concurrent page fetches, used
+            only when semaphore is not supplied. Each execute_pagination call
+            builds its own semaphore, so this bounds one category in isolation;
+            with many categories running concurrently the effective limit is
+            this value times the category count.
         start_offset: Additive base applied to every offset, for APIs whose
             offset space does not begin at zero. Page N requests offset
             start_offset + (N - 1) * products_per_page.
@@ -327,7 +331,6 @@ async def execute_pagination(
         "executing pagination",
         extra={"category": category_identifier, "total": total_pages},
     )
-    semaphore = asyncio.Semaphore(max_concurrent_tasks)
     stop_event = asyncio.Event()
     pagination_label = "offset" if pagination_mode == "offset" else "page"
     page_range = range(2, total_pages + 1)

tests/test_pagination_semaphore.py

@@ -0,0 +1,87 @@
+"""unit test: execute_pagination honours a caller-supplied shared semaphore
+
+run from the repo root:
+    uv run python tests/test_pagination_semaphore.py
+
+offline — no network. A fake page func records peak concurrency so we can
+assert the fan-out limit actually holds, both for the shared-semaphore path
+(one instance capping many categories at once) and the default per-call path.
+"""
+
+import asyncio
+
+from monitor_utils.runner import execute_pagination
+
+
+def _make_tracking_func(state: dict):
+    """Return a page func that tracks peak simultaneous in-flight calls."""
+
+    async def func(session, offset, **kwargs):
+        state["live"] += 1
+        state["peak"] = max(state["peak"], state["live"])
+        await asyncio.sleep(0.01)
+        state["live"] -= 1
+
+    return func
+
+
+def test_shared_semaphore_caps_total_concurrency_across_categories():
+    """One shared semaphore bounds combined fan-out over concurrent categories.
+
+    Two categories paginate at once through the same semaphore(limit); without
+    sharing, each would build its own semaphore and the peak could reach twice
+    the limit. Peak must stay at or below the shared limit.
+    """
+
+    limit = 3
+    shared = asyncio.Semaphore(limit)
+    state = {"live": 0, "peak": 0}
+    func = _make_tracking_func(state)
+
+    async def run():
+        await asyncio.gather(
+            execute_pagination(
+                "catA",
+                total_products=300,
+                products_per_page=30,
+                func=func,
+                session=None,
+                semaphore=shared,
+            ),
+            execute_pagination(
+                "catB",
+                total_products=300,
+                products_per_page=30,
+                func=func,
+                session=None,
+                semaphore=shared,
+            ),
+        )
+
+    asyncio.run(run())
+    assert state["peak"] <= limit, f"peak {state['peak']} exceeded shared limit {limit}"
+
+
+def test_default_falls_back_to_max_concurrent_tasks():
+    """Without a semaphore, max_concurrent_tasks still bounds a single call."""
+
+    state = {"live": 0, "peak": 0}
+    func = _make_tracking_func(state)
+
+    async def run():
+        await execute_pagination(
+            "cat",
+            total_products=300,
+            products_per_page=30,
+            func=func,
+            session=None,
+            max_concurrent_tasks=2,
+        )
+
+    asyncio.run(run())
+    assert state["peak"] <= 2, f"peak {state['peak']} exceeded max_concurrent_tasks 2"
+
+
+test_shared_semaphore_caps_total_concurrency_across_categories()
+test_default_falls_back_to_max_concurrent_tasks()
+print("pagination semaphore test passed")