add concurrency test

Author: luke-beamcloud

fastloop/fastloop.py

@@ -349,11 +349,6 @@ async def _event_handler(request: dict[str, Any], func: Any = func):
                     loop_instance.ctx = context
                     await loop_instance.on_event(context, event)
 
-                if loop.status != LoopStatus.RUNNING:
-                    loop = await self.state_manager.update_loop_status(
-                        loop.loop_id, LoopStatus.RUNNING
-                    )
-
                 if loop_instance or created:
                     func_to_run = func
                 else:
@@ -584,11 +579,6 @@ async def _start_handler(request: dict[str, Any]):
                     state_manager=self.state_manager,
                 )
 
-                if workflow.status != LoopStatus.RUNNING:
-                    await self.state_manager.update_workflow_status(
-                        workflow.workflow_run_id, LoopStatus.RUNNING
-                    )
-
                 await self.workflow_manager.start(
                     func,
                     context,
@@ -712,15 +702,7 @@ async def restart_loop(self, loop_id: str) -> bool:
                 loop_delay=metadata["loop_delay"],
             )
             if started:
-                await self.state_manager.update_loop_status(
-                    loop.loop_id, LoopStatus.RUNNING
-                )
-                logger.info(
-                    "Restarted loop",
-                    extra={
-                        "loop_id": loop.loop_id,
-                    },
-                )
+                logger.info("Restarted loop", extra={"loop_id": loop.loop_id})
                 return True
             else:
                 logger.warning(
@@ -790,15 +772,9 @@ async def restart_workflow(self, workflow_run_id: str) -> bool:
             )
 
             if started:
-                await self.state_manager.update_workflow_status(
-                    workflow.workflow_run_id, LoopStatus.RUNNING
-                )
                 logger.info(
                     "Restarted workflow",
-                    extra={
-                        "workflow_run_id": workflow.workflow_run_id,
-                        "block_index": workflow.current_block_index,
-                    },
+                    extra={"workflow_run_id": workflow.workflow_run_id},
                 )
             return started
 

fastloop/loop.py

@@ -74,10 +74,17 @@ async def _run(
         context: Any,
         loop_id: str,
         delay: float,
+        loop_start_func: Callable[..., Any] | None,
         loop_stop_func: Callable[..., Any] | None,
     ) -> None:
         try:
             async with self.state_manager.with_claim(loop_id):  # type: ignore
+                await self.state_manager.update_loop_status(loop_id, LoopStatus.RUNNING)
+                if loop_start_func:
+                    if asyncio.iscoroutinefunction(loop_start_func):
+                        await loop_start_func(context)
+                    else:
+                        loop_start_func(context)  # type: ignore
                 idle_cycles = 0
 
                 while not context.should_stop and not context.should_pause:
@@ -177,15 +184,10 @@ async def start(
         if loop.loop_id in self.loop_tasks:
             return False
 
-        if loop_start_func:
-            if asyncio.iscoroutinefunction(loop_start_func):
-                await loop_start_func(context)
-            else:
-                loop_start_func(context)  # type: ignore
-
-        # TODO: switch out executor for thread/process based on config
         self.loop_tasks[loop.loop_id] = asyncio.create_task(
-            self._run(func, context, loop.loop_id, loop_delay, loop_stop_func)
+            self._run(
+                func, context, loop.loop_id, loop_delay, loop_start_func, loop_stop_func
+            )
         )
 
         return True

fastloop/monitor.py

@@ -158,7 +158,7 @@ async def _check_orphaned_tasks(self) -> None:
                 task_name=task.task_name,
                 retry_policy=metadata.get("retry"),
                 executor_type=metadata.get("executor"),
-            )
+                )
 
     async def _check_scheduled_workflows(self) -> None:
         now = time.time()

fastloop/workflow.py

@@ -216,6 +216,7 @@ async def _run(
         func: Callable[..., Any],
         context: Any,
         workflow_run_id: str,
+        on_start: Callable[..., Any] | None,
         on_stop: Callable[..., Any] | None,
         on_block_complete: Callable[..., Any] | None,
         on_error: Callable[..., Any] | None,
@@ -224,6 +225,10 @@ async def _run(
     ) -> None:
         try:
             async with self.state_manager.with_workflow_claim(workflow_run_id):
+                await self.state_manager.update_workflow_status(
+                    workflow_run_id, LoopStatus.RUNNING
+                )
+                await _call(on_start, context)
                 while True:
                     workflow = await self.state_manager.get_workflow(workflow_run_id)
                     if workflow.status in (LoopStatus.STOPPED, LoopStatus.FAILED):
@@ -438,13 +443,12 @@ async def start(
         if workflow.workflow_run_id in self.tasks:
             return False
 
-        await _call(on_start, context)
-
         self.tasks[workflow.workflow_run_id] = asyncio.create_task(
             self._run(
                 func,
                 context,
                 workflow.workflow_run_id,
+                on_start,
                 on_stop,
                 on_block_complete,
                 on_error,

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastloop"
-version = "0.1.100"
+version = "0.1.101"
 description = "A Python package for deploying stateful loops"
 readme = "README.md"
 requires-python = ">=3.12"