From 4aa0e432895c541bd0f8f44d99b859dbc9e75fa8 Mon Sep 17 00:00:00 2001
From: TPV <TPVDev@proton.me>
Date: Thu, 16 Jul 2026 14:19:33 -0700
Subject: [PATCH] small bug fix

---
 src/monitor_utils/runner.py | 84 +++++++++++++++++++------------------
 1 file changed, 43 insertions(+), 41 deletions(-)

diff --git a/src/monitor_utils/runner.py b/src/monitor_utils/runner.py
index 8d58a84..a03d0f6 100644
--- a/src/monitor_utils/runner.py
+++ b/src/monitor_utils/runner.py
@@ -200,9 +200,9 @@ async def execute_monitor_logic(
 ):
     """Run the monitor's category tasks in a loop with a fresh session per cycle.
 
-    Each cycle opens a TLS or base session, dispatches task_func per category
-    concurrently, logs expected failures, raises unexpected ones, then sleeps
-    briefly before the next cycle.
+    Each cycle opens a TLS or base session, awaits pre_task_func if given,
+    dispatches task_func per category concurrently, logs expected failures,
+    raises unexpected ones, then sleeps briefly before the next cycle.
 
     Args:
         category_identifiers: List of category ids, or a mapping of category
@@ -223,8 +223,8 @@ async def execute_monitor_logic(
             Category tasks must not depend on its side effects (session
             headers, cookies, fetched state) since no ordering is guaranteed.
         use_tls: Use the TLS fingerprint session instead of the base session.
-        **kwargs: Forwarded verbatim to concurrent_pre_task and every
-            task_func call.
+        **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
+            and every task_func call.
     """
 
     while True:
@@ -237,48 +237,50 @@ async def execute_monitor_logic(
             else get_base_session(headers=headers, cookies=cookies, proxies=proxies)
         )
         async with session_ctx as session:
-            LOGGER.info("starting tasks")
-            if isinstance(category_identifiers, dict):
-                tasks = [
-                    task_func(category_id, session, query_category_id=query, **kwargs)
-                    for category_id, query in category_identifiers.items()
-                ]
-                category_keys = list(category_identifiers.keys())
-            else:
-                tasks = [
-                    task_func(category_identifier, session, **kwargs)
-                    for category_identifier in category_identifiers
-                ]
-                category_keys = list(category_identifiers)
-
+            skip_cycle = False
             if pre_task_func:
                 try:
                     await pre_task_func(session=session, **kwargs)
                 except EXPECTED_TASK_ERRORS as error:
-                    LOGGER.warning("Failed to get products", exc_info=error)
-                except Exception:
-                    LOGGER.exception("Unexpected error")
-                    raise
+                    LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
+                    skip_cycle = True
 
-            if concurrent_pre_task_func:
-                tasks.append(concurrent_pre_task_func(session=session, **kwargs))
-                category_keys.append("pre_task")
+            if not skip_cycle:
+                LOGGER.info("starting tasks")
+                if isinstance(category_identifiers, dict):
+                    tasks = [
+                        task_func(
+                            category_id, session, query_category_id=query, **kwargs
+                        )
+                        for category_id, query in category_identifiers.items()
+                    ]
+                    category_keys = list(category_identifiers.keys())
+                else:
+                    tasks = [
+                        task_func(category_identifier, session, **kwargs)
+                        for category_identifier in category_identifiers
+                    ]
+                    category_keys = list(category_identifiers)
 
-            results = await asyncio.gather(*tasks, return_exceptions=True)
-            for category_identifier, result in zip(category_keys, results):
-                if isinstance(result, EXPECTED_TASK_ERRORS):
-                    LOGGER.warning(
-                        "failed to get products",
-                        extra={"category": category_identifier},
-                        exc_info=result,
-                    )
-                elif isinstance(result, Exception):
-                    LOGGER.error(
-                        "unexpected error",
-                        extra={"category": category_identifier},
-                        exc_info=result,
-                    )
-                    raise result
+                if concurrent_pre_task_func:
+                    tasks.append(concurrent_pre_task_func(session=session, **kwargs))
+                    category_keys.append("pre_task")
+
+                results = await asyncio.gather(*tasks, return_exceptions=True)
+                for category_identifier, result in zip(category_keys, results):
+                    if isinstance(result, EXPECTED_TASK_ERRORS):
+                        LOGGER.warning(
+                            "failed to get products",
+                            extra={"category": category_identifier},
+                            exc_info=result,
+                        )
+                    elif isinstance(result, Exception):
+                        LOGGER.error(
+                            "unexpected error",
+                            extra={"category": category_identifier},
+                            exc_info=result,
+                        )
+                        raise result
 
         LOGGER.info("sleeping before next tasks", extra={"delay": 0.30})
         await asyncio.sleep(0.30)