From aa6ed4cb06797004d736abc5cd5e0f7f2527944b Mon Sep 17 00:00:00 2001 From: TPV Date: Thu, 30 Jul 2026 11:52:36 -0700 Subject: [PATCH] added pagination fan out pagination abstraction --- src/monitor_utils/runner.py | 37 ++++++++++- tests/test_fan_out_group.py | 126 ++++++++++++++++++++++++++++++++++++ 2 files changed, 162 insertions(+), 1 deletion(-) create mode 100644 tests/test_fan_out_group.py diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py index 8684b1b..9bc4c60 100644 --- a/src/monitor_utils/runner.py +++ b/src/monitor_utils/runner.py @@ -1,7 +1,7 @@ import asyncio import logging from collections.abc import Awaitable, Callable, Iterable -from contextlib import nullcontext +from contextlib import asynccontextmanager, nullcontext import aiohttp from aioproxies import AioProxies @@ -568,3 +568,38 @@ def first_error(group_exc: BaseExceptionGroup) -> BaseException: error = error.exceptions[0] return error + + +@asynccontextmanager +async def fan_out_group(): + """A TaskGroup for page-one fan-out whose failure surfaces unwrapped. + + Use it where one page's processing must be bound to the pagination that + page spawns: create the processing task on the yielded group, then await + execute_pagination in the body. The group cancels whichever sibling is + still running when the other fails, so a dead category stops doing work + nobody will use, and a cancelled cycle takes both down together. + + The unwrapping is the reason this is shared rather than inlined. A bare + TaskGroup always raises an ExceptionGroup, and run_monitor_cycle classifies + task results with isinstance against EXPECTED_TASK_ERRORS -- an + ExceptionGroup matches neither branch as intended, so a routine page + failure would be reclassified as an unexpected one and kill the monitor. + first_error unwraps to a leaf, preferring an unexpected exception so a real + bug is never masked by a network error. + + Note the caller's own body is inside the group, so a plain exception raised + there (not from a child) propagates unchanged rather than being wrapped, + and an early `return` still awaits the child tasks. + + Example: + async with fan_out_group() as group: + group.create_task(process_page(data, ...)) + await execute_pagination(...) + """ + + try: + async with asyncio.TaskGroup() as group: + yield group + except BaseExceptionGroup as group_exc: + raise first_error(group_exc) from None diff --git a/tests/test_fan_out_group.py b/tests/test_fan_out_group.py new file mode 100644 index 0000000..c624a70 --- /dev/null +++ b/tests/test_fan_out_group.py @@ -0,0 +1,126 @@ +"""unit test: fan_out_group binds sibling tasks and surfaces errors unwrapped + +run from the repo root: + uv run pytest tests/test_fan_out_group.py + +offline — no network. These lock the contract monitors depend on: a bare +TaskGroup always raises an ExceptionGroup, which run_monitor_cycle would +misclassify (it matches neither the EXPECTED_TASK_ERRORS branch nor the +unexpected-Exception branch as intended), so the unwrapping is load-bearing and +not cosmetic. +""" + +import asyncio + +import aiohttp +import pytest + +from monitor_utils.runner import fan_out_group + + +async def _boom(error: BaseException): + raise error + + +def test_child_error_surfaces_unwrapped(): + """A failing child arrives as itself, not wrapped in an ExceptionGroup.""" + + async def scenario(): + async with fan_out_group() as group: + group.create_task(_boom(RuntimeError("child"))) + await asyncio.sleep(0) + + with pytest.raises(RuntimeError) as exc_info: + asyncio.run(scenario()) + + assert not isinstance(exc_info.value, BaseExceptionGroup) + assert str(exc_info.value) == "child" + + +def test_body_error_surfaces_unwrapped(): + """An error raised by the caller's own body propagates unchanged.""" + + async def ok(): + await asyncio.sleep(0) + + async def scenario(): + async with fan_out_group() as group: + group.create_task(ok()) + raise ValueError("body") + + with pytest.raises(ValueError) as exc_info: + asyncio.run(scenario()) + + assert not isinstance(exc_info.value, BaseExceptionGroup) + + +def test_unexpected_error_is_preferred_over_an_expected_one(): + """first_error must not let a routine network error mask a real bug.""" + + async def scenario(): + async with fan_out_group() as group: + group.create_task(_boom(aiohttp.ClientError("routine"))) + await asyncio.sleep(0) + raise ValueError("real bug") + + with pytest.raises(ValueError): + asyncio.run(scenario()) + + +def test_early_return_still_awaits_children(): + """A `return` inside the body must not abandon an in-flight child task.""" + + finished = [] + + async def child(): + await asyncio.sleep(0.01) + finished.append("done") + + async def scenario(): + async with fan_out_group() as group: + group.create_task(child()) + return "returned" + + assert asyncio.run(scenario()) == "returned" + assert finished == ["done"] + + +def test_parked_sibling_is_cancelled_when_the_body_fails(): + """The binding is the point: a child that never finishes must not hang. + + Without the group, a create_task + `finally: await` shape would park here + forever. The deadline is the assertion — a hang fails this test. + """ + + async def parked(): + await asyncio.Event().wait() + + async def scenario(): + async with asyncio.timeout(2): + async with fan_out_group() as group: + group.create_task(parked()) + await asyncio.sleep(0) + raise RuntimeError("body") + + with pytest.raises(RuntimeError): + asyncio.run(scenario()) + + +def test_children_run_concurrently_with_the_body(): + """Both siblings make progress; the group is not a sequencing device.""" + + order = [] + + async def child(): + order.append("child-start") + await asyncio.sleep(0.01) + order.append("child-end") + + async def scenario(): + async with fan_out_group() as group: + group.create_task(child()) + order.append("body") + await asyncio.sleep(0.02) + + asyncio.run(scenario()) + assert order == ["body", "child-start", "child-end"]