Audit and fix scheduling and state management (#15)

* Refactor: Improve sleep functionality and state management

This commit introduces several improvements:

- **Enhanced `LoopContext.sleep_for` and `LoopContext.sleep_until`**:
    - Added docstrings explaining functionality, arguments, and potential errors.
    - Implemented validation for sleep durations and timestamps to prevent invalid values.
    - `sleep_for` now raises a `ValueError` if the duration is not positive.
    - `sleep_until` now raises a `ValueError` if the timestamp is in the past.

- **State Management Updates**:
    - **`LoopEvent` and `LoopState`**:
        - Renamed `to_json` methods to `to_dict` for clarity and consistency.
        - Added `to_string` methods for JSON serialization.
        - Kept `to_json` as a deprecated alias for `to_dict` for backward compatibility.
        - Changed `created_at` type in `LoopState` from `int` to `float` for better precision.
    - **`RedisStateManager`**:
        - Improved `_run_wake_monitoring` with better error handling and logging.
        - Modified `set_wake_time` to use milliseconds (`px`) for Redis TTL, allowing for more precise wake-up scheduling.
        - Updated `set_wake_time` to use the full wake key in `srem` for consistency.
        - Added a docstring to `has_claim` for better understanding.
    - **`S3StateManager`**:
        - Explicitly raised `NotImplementedError` in `set_wake_time` with a clear message indicating that S3 does not support wake time scheduling.
        - Added a private `_delete_object` method.

- **Utility Function Update**:
    - **`infer_application_path`**:
        - Updated the function signature to accept `app_instance` instead of `self`.
        - Improved the docstring to clarify its purpose and arguments.
        - Enhanced introspection logic to correctly identify the application instance.

Co-authored-by: luke <luke@smartshare.io>

* Refactor: Use to_dict() instead of to_json()

Co-authored-by: luke <luke@smartshare.io>

* Changes made by Agent

Co-authored-by: luke <luke@smartshare.io>

* Add Ruff linting and formatting, and CI workflow

Co-authored-by: luke <luke@smartshare.io>

* Checkpoint before follow-up message

Co-authored-by: luke <luke@smartshare.io>

* Refactor: Use real Redis for scheduling tests

This commit replaces fakeredis with a real Redis instance for scheduling tests. It also updates the CI to include a Redis service and modifies the Makefile to support starting/stopping a local Redis instance. The integration tests have been removed as they are now covered by the refactored unit tests.

Co-authored-by: luke <luke@smartshare.io>

* Refactor Redis wake scheduling for reliability

Co-authored-by: luke <luke@smartshare.io>

* Refactor: Improve state management and documentation

This commit refactors the state management system, particularly for Redis, to enhance reliability and simplify the code. It also updates documentation and tests for better clarity and robustness.

Co-authored-by: luke <luke@smartshare.io>

* Refactor: Improve Redis state manager and tests

Add stop method to RedisStateManager for graceful shutdown of the wake monitoring thread.
Update tests to use unique app names and ensure proper cleanup.
Add pytest-asyncio and ruff to dev dependencies.
Update fastapi version to 0.115.14.
Update fastloop version to 0.1.71.

Co-authored-by: luke <luke@smartshare.io>

* Refactor loop monitor and state management for efficiency

Co-authored-by: luke <luke@smartshare.io>

* Remove unused get_running_loop_ids method

Co-authored-by: luke <luke@smartshare.io>

---------

Co-authored-by: Cursor Agent <cursoragent@cursor.com>

Author: luke-lombardi

.github/workflows/ci.yml

@@ -0,0 +1,69 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    
+    services:
+      redis:
+        image: redis:7
+        ports:
+          - 6379:6379
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+        with:
+          version: "latest"
+
+      - name: Set up Python
+        run: uv python install 3.12
+
+      - name: Install dependencies
+        run: uv sync --all-extras --dev
+
+      - name: Run linter
+        run: uv run ruff check .
+
+      - name: Run formatter check
+        run: uv run ruff format --check .
+
+      - name: Run tests
+        run: uv run pytest tests/ -v --tb=short
+        env:
+          REDIS_TEST_HOST: localhost
+          REDIS_TEST_PORT: 6379
+
+  type-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+        with:
+          version: "latest"
+
+      - name: Set up Python
+        run: uv python install 3.12
+
+      - name: Install dependencies
+        run: uv sync --all-extras --dev
+
+      - name: Check Python syntax
+        run: |
+          uv run python -m py_compile fastloop/*.py
+          uv run python -m py_compile fastloop/**/*.py

Makefile

@@ -1,5 +1,40 @@
 VERSION=$(shell grep '^version =' pyproject.toml | head -1 | cut -d'"' -f2)
 
+.PHONY: test test-verbose install-dev lint format check publish redis-start redis-stop
+
+install-dev:
+	uv sync --all-extras --dev
+
+# Start Redis for local testing
+redis-start:
+	docker run -d --name fastloop-test-redis -p 6379:6379 redis:7
+	@echo "Redis started. Run 'make test' to run tests."
+
+redis-stop:
+	docker stop fastloop-test-redis && docker rm fastloop-test-redis
+
+# Run tests (requires Redis - set REDIS_TEST_HOST=localhost or run 'make redis-start')
+test:
+	REDIS_TEST_HOST=localhost uv run pytest tests/ -v
+
+test-verbose:
+	REDIS_TEST_HOST=localhost uv run pytest tests/ -v -s
+
+test-scheduling:
+	REDIS_TEST_HOST=localhost uv run pytest tests/test_scheduling.py -v -s
+
+lint:
+	uv run ruff check .
+
+format:
+	uv run ruff format .
+
+format-check:
+	uv run ruff format --check .
+
+check: lint format-check test
+	@echo "All checks passed!"
+
 publish:
 	rm -rf dist/
 	uv build

fastloop/context.py

@@ -21,9 +21,19 @@
     from .integrations import Integration
 
 logger = setup_logger(__name__)
-
 T = TypeVar("T", bound="LoopContext")
 
+_DURATION_RE = re.compile(
+    r"^(\d+(?:\.\d+)?)\s*(seconds?|secs?|minutes?|mins?|hours?|hrs?|days?)$"
+)
+_UNIT_MULTIPLIERS = {
+    "sec": 1,
+    "min": 60,
+    "hour": 3600,
+    "hr": 3600,
+    "day": 86400,
+}
+
 
 class LoopContext:
     def __init__(
@@ -72,9 +82,13 @@ def switch_to(self: T, func: Callable[[T], Awaitable[None]]):
         raise LoopContextSwitchError(func, self)
 
     async def sleep_for(self, duration: float | str) -> None:
+        """Sleep the loop for a duration (float seconds or string like "5 seconds")."""
         if isinstance(duration, str):
             duration = self._parse_duration(duration)
 
+        if duration <= 0:
+            raise ValueError("Sleep duration must be positive")
+
         logger.info(
             f"Loop sleeping for {duration} seconds",
             extra={"loop_id": self.loop_id, "duration": duration},
@@ -84,6 +98,10 @@ async def sleep_for(self, duration: float | str) -> None:
         self.pause()
 
     async def sleep_until(self, timestamp: float) -> None:
+        """Sleep the loop until a specific Unix timestamp."""
+        if timestamp <= time.time():
+            raise ValueError("Cannot sleep until a time in the past")
+
         logger.info(
             f"Loop sleeping until {timestamp}",
             extra={"loop_id": self.loop_id, "timestamp": timestamp},
@@ -223,26 +241,12 @@ def should_pause(self) -> bool:
         return self._pause_requested
 
     def _parse_duration(self, duration_str: str) -> float:
-        duration_str = duration_str.lower().strip()
-
-        match = re.match(
-            r"^(\d+(?:\.\d+)?)\s*(seconds?|secs?|minutes?|mins?|hours?|hrs?|days?)$",
-            duration_str,
-        )
+        match = _DURATION_RE.match(duration_str.lower().strip())
         if not match:
             raise ValueError(f"Invalid duration format: {duration_str}")
 
-        value = float(match.group(1))
-        unit = match.group(2)
-
-        # Convert to seconds
-        if unit.startswith("sec"):
-            return value
-        elif unit.startswith("min"):
-            return value * 60
-        elif unit.startswith("hour") or unit.startswith("hr"):
-            return value * 3600
-        elif unit.startswith("day"):
-            return value * 86400
-        else:
-            raise ValueError(f"Unknown time unit: {unit}")
+        value, unit = float(match.group(1)), match.group(2)
+        for prefix, mult in _UNIT_MULTIPLIERS.items():
+            if unit.startswith(prefix):
+                return value * mult
+        return value

fastloop/fastloop.py

@@ -22,7 +22,7 @@
 from .integrations import Integration
 from .logging import configure_logging, setup_logger
 from .loop import LoopEvent, LoopManager
-from .state.state import LoopState, StateManager, create_state_manager
+from .state.state import StateManager, create_state_manager
 from .types import BaseConfig, LoopStatus
 from .utils import get_func_import_path, import_func_from_path, infer_application_path
 
@@ -338,7 +338,7 @@ async def _retrieve_handler(loop_id: str):
                     ) from e
 
                 return JSONResponse(
-                    content=loop.to_json(), media_type="application/json"
+                    content=loop.to_dict(), media_type="application/json"
                 )
 
             async def _stop_handler(loop_id: str):
@@ -499,120 +499,76 @@ def __init__(
         wake_queue: Queue[str],
         fastloop_instance: FastLoop,
     ):
-        self.state_manager: StateManager = state_manager
-        self.loop_manager: LoopManager = loop_manager
-        self.restart_callback: Callable[[str], Coroutine[Any, Any, bool]] = (
-            restart_callback
-        )
-        self.wake_queue: Queue[str] = wake_queue
-        self.fastloop_instance: FastLoop = fastloop_instance
-        self._stop_event: asyncio.Event = asyncio.Event()
+        self.state_manager = state_manager
+        self.loop_manager = loop_manager
+        self.restart_callback = restart_callback
+        self.wake_queue = wake_queue
+        self.fastloop_instance = fastloop_instance
+        self._stop_event = asyncio.Event()
 
     def stop(self) -> None:
         self._stop_event.set()
 
-    async def run(self):
-        while not self._stop_event.is_set():
-            try:
-                if not self.wake_queue.empty():
-                    loop_id = self.wake_queue.get()
-                    if await self.state_manager.has_claim(loop_id):
-                        continue
-
-                    logger.info(
-                        "Loop woke up, restarting",
-                        extra={"loop_id": loop_id},
-                    )
-                    if not await self.restart_callback(loop_id):
-                        await self.state_manager.update_loop_status(
-                            loop_id, LoopStatus.STOPPED
-                        )
-                        await self.loop_manager.stop(loop_id)
-
-                    continue
-
-                loop_ids: set[str] = await self.state_manager.get_all_loop_ids()
-                active_loop_ids: set[str] = await self.loop_manager.active_loop_ids()
-                loops_running: set[str] = active_loop_ids.intersection(loop_ids)
-
-                for loop_id in loops_running:
-                    loop = await self.state_manager.get_loop(loop_id)
-
-                    if (
-                        loop.status in LoopStatus.IDLE
-                        or loop.status == LoopStatus.STOPPED
-                    ):
-                        if await self.state_manager.has_claim(loop_id):
-                            continue
-
-                        logger.info(
-                            "Loop is idle or stopped, stopping",
-                            extra={"loop_id": loop_id},
-                        )
+    async def _process_wake(self, loop_id: str) -> None:
+        if await self.state_manager.has_claim(loop_id):
+            return
+        logger.info("Loop woke up, restarting", extra={"loop_id": loop_id})
+        if not await self.restart_callback(loop_id):
+            await self.state_manager.update_loop_status(loop_id, LoopStatus.STOPPED)
 
-                        await self.loop_manager.stop(loop_id)
-                        continue
+    async def _check_orphaned_loops(self) -> None:
+        running_loops = await self.state_manager.get_all_loops(
+            status=LoopStatus.RUNNING
+        )
+        for loop in running_loops:
+            if await self.state_manager.has_claim(loop.loop_id):
+                continue
+            logger.info(
+                "Loop has no claim, restarting", extra={"loop_id": loop.loop_id}
+            )
+            if not await self.restart_callback(loop.loop_id):
+                await self.state_manager.update_loop_status(
+                    loop.loop_id, LoopStatus.STOPPED
+                )
 
-                loops: list[LoopState] = await self.state_manager.get_all_loops(
-                    status=LoopStatus.RUNNING
+    async def _check_disconnect_stops(self) -> None:
+        active_ids = await self.loop_manager.active_loop_ids()
+        for loop_id in active_ids:
+            try:
+                loop = await self.state_manager.get_loop(loop_id)
+            except LoopNotFoundError:
+                continue
+            if not loop.loop_name:
+                continue
+            metadata = self.fastloop_instance._loop_metadata.get(loop.loop_name)
+            if not metadata or not metadata.get("stop_on_disconnect"):
+                continue
+            if not await self.fastloop_instance.has_active_clients(loop_id):
+                logger.info(
+                    "Loop has no clients, stopping",
+                    extra={"loop_id": loop_id, "loop_name": loop.loop_name},
                 )
-                for loop in loops:
-                    # Restart loop if it has no claim and is not idle (maybe the task crashed or was interrupted)
-                    if not await self.state_manager.has_claim(loop.loop_id):
-                        logger.info(
-                            "Loop has no claim, restarting",
-                            extra={
-                                "loop_id": loop.loop_id,
-                            },
-                        )
+                await self.state_manager.update_loop_status(loop_id, LoopStatus.STOPPED)
+                await self.loop_manager.stop(loop_id)
+
+    async def run(self):
+        while not self._stop_event.is_set():
+            try:
+                while not self.wake_queue.empty():
+                    await self._process_wake(self.wake_queue.get_nowait())
 
-                        if not await self.restart_callback(loop.loop_id):
-                            await self.state_manager.update_loop_status(
-                                loop.loop_id, LoopStatus.STOPPED
-                            )
-                            await self.loop_manager.stop(loop.loop_id)
-
-                        continue
-
-                # Check for loops with stop_on_disconnect=true that have no active clients
-                for loop in loops:
-                    if (
-                        loop.loop_name
-                        and loop.loop_name in self.fastloop_instance._loop_metadata
-                    ):
-                        metadata = self.fastloop_instance._loop_metadata[loop.loop_name]
-                        if metadata.get("stop_on_disconnect", False):
-                            has_clients = (
-                                await self.fastloop_instance.has_active_clients(
-                                    loop.loop_id
-                                )
-                            )
-                            if not has_clients:
-                                logger.info(
-                                    "Loop has stop_on_disconnect=true and no active clients, stopping",
-                                    extra={
-                                        "loop_id": loop.loop_id,
-                                        "loop_name": loop.loop_name,
-                                    },
-                                )
-                                await self.state_manager.update_loop_status(
-                                    loop.loop_id, LoopStatus.STOPPED
-                                )
-                                await self.loop_manager.stop(loop.loop_id)
+                await self._check_orphaned_loops()
+                await self._check_disconnect_stops()
 
                 try:
                     await asyncio.wait_for(
                         self._stop_event.wait(), timeout=WATCHDOG_INTERVAL_S
                     )
                     break
                 except TimeoutError:
-                    continue
-
+                    pass
             except asyncio.CancelledError:
                 break
-            except BaseException as e:
-                logger.error(
-                    "Error in loop monitor",
-                    extra={"error": str(e)},
-                )
+            except Exception as e:
+                logger.error("Error in loop monitor", extra={"error": str(e)})
                 await asyncio.sleep(WATCHDOG_INTERVAL_S)