Skip to content

Hot Reloading Development Guide

StarStreamer's hot reloading system provides a seamless development experience by intelligently reloading modules without losing connections or state. This guide covers the architecture, usage patterns, and best practices for effective development with hot reloading.

Overview

Hot reloading in StarStreamer uses a hybrid approach that balances development speed with system safety:

  • Module changes → Hot reload (fast, preserves state)
  • Core changes → Full restart (safe, clean state)

This approach allows rapid iteration on module development while ensuring framework changes are handled safely.

Architecture

File Watcher System

The hot reloading system is built on Python's watchdog library with custom event handling and thread-safe event loop integration:

class HybridReloadHandler(FileSystemEventHandler):
    """Handler for hybrid reload - hot reload modules, full restart for core changes"""

    def __init__(self, starstreamer_app: "StarStreamer", loop: asyncio.AbstractEventLoop):
        self.app = starstreamer_app
        self.loop = loop  # Main event loop for thread-safe scheduling

    def on_modified(self, event: FileSystemEvent) -> None:
        if "src/modules/" in path or "/modules/" in path:
            self._schedule_module_reload(path)
        elif "src/starstreamer/" in path:
            self._schedule_full_restart()

Reload Strategy Decision Tree

flowchart TD
    A[File Changed] --> B{File Type?}
    B -->|Python .py| C{Directory?}
    B -->|Other| D[Ignore]
    C -->|modules/| E[Hot Reload]
    C -->|src/starstreamer/| F[Full Restart]
    C -->|tests/| D
    C -->|__pycache__| D
    E --> G[Extract Module Name]
    G --> H[Reload Specific Module]
    F --> I[Graceful Shutdown]
    I --> J[Complete Restart]

Module Discovery and Loading

The system includes automatic module discovery:

class ModuleRegistry:
    async def discover_modules(self) -> list[str]:
        """Discover available modules in the modules directory"""

    async def reload_module(self, module_name: str) -> bool:
        """Hot reload a specific module using importlib.reload()"""

    async def unregister_module_handlers(self, module_name: str) -> None:
        """Clean up handlers for a module before reload"""

Usage

Starting Development Mode

Enable hot reloading with the --reload flag:

# Basic hot reload
uv run python src/main.py --reload

# With custom web port
uv run python src/main.py --reload --web-port 8080

# With debug logging
uv run python src/main.py --reload --log-level DEBUG

# Combined with other options
uv run python src/main.py --reload --web-port 8080 --log-level DEBUG

What Gets Monitored

Included Files: - *.py files in the project directory - Changes in modules/ subdirectories - Changes in src/starstreamer/ subdirectories

Excluded Files: - Test files (test_*.py, *_test.py) - Python cache (__pycache__/, *.pyc) - Hidden files (.git/, .venv/) - Database files (*.db, *.db-wal, *.db-shm)

File Change Handling

The system includes intelligent debouncing and thread-safe event loop integration:

self.restart_delay = 0.3  # 300ms debounce

def _schedule_module_reload(self, path: str) -> None:
    def reload_module() -> None:
        # Check if we have an event loop
        if not self.loop:
            self.logger.error("Cannot hot reload: no event loop available")
            return

        # Schedule coroutine on the main event loop (thread-safe)
        future = asyncio.run_coroutine_threadsafe(
            self._hot_reload_module(module_name), 
            self.loop
        )
        future.add_done_callback(self._handle_reload_result)

    # Cancel any pending reload
    if self.restart_timer:
        self.restart_timer.cancel()

    # Schedule new reload with delay
    self.restart_timer = threading.Timer(self.restart_delay, reload_module)
    self.restart_timer.start()

Development Workflows

Module Development Workflow

  1. Start development mode: 

    uv run python src/main.py --reload

  2. Create or modify a module: 

    # modules/custom/actions/commands.py
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!test"))
async def test_command(event: Event, twitch: TwitchClient):
    await twitch.send_message("Testing hot reload!")

  3. Save the file → Module reloads automatically in ~200ms

  4. Test immediately in Twitch chat without restarting

Core Development Workflow

  1. Modify core framework: 

    # src/starstreamer/core/event_bus.py
class EventBus:
    async def emit(self, event_type: str, data: dict) -> None:
        # Add new functionality
        pass

  2. Save the file → Full restart occurs automatically

  3. All modules reload cleanly with new core functionality

New Module Creation Workflow

  1. Create module structure: 

    mkdir -p modules/mymodule/actions
touch modules/mymodule/__init__.py
touch modules/mymodule/module.py
touch modules/mymodule/actions/__init__.py
touch modules/mymodule/actions/commands.py

  2. Implement module class: 

    # modules/mymodule/module.py
from modules.base import BaseModule

class MyModule(BaseModule):
    @property
    def module_name(self) -> str:
        return "mymodule"

    async def register_actions(self) -> None:
        from modules.mymodule.actions import commands  # noqa: F401

  3. Add commands: 

    # modules/mymodule/actions/commands.py
from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger
from starstreamer.plugins.twitch import TwitchClient
from starstreamer.runtime.types import Event

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!mycommand"))
async def my_command(event: Event, twitch: TwitchClient):
    await twitch.send_message("My new module works!")

  4. Save files → Module is discovered and loaded automatically

Hot Reload Internals

Module Reloading Process

async def _hot_reload_module(self, module_name: str) -> None:
    """Perform hot reload of a specific module"""
    try:
        # 1. Get existing module instance
        module = self.app.module_registry._modules.get(module_name)
        if not module:
            self.logger.warning(f"Module {module_name} not found for reload")
            return

        # 2. Unregister existing handlers
        await self.app.event_bus.unregister_module_handlers(module_name)

        # 3. Reload module using importlib
        import importlib
        module_path = f"modules.{module_name}.module"
        if module_path in sys.modules:
            importlib.reload(sys.modules[module_path])

        # 4. Re-register actions
        await module.register_actions()

        self.logger.info(f"🔥 Hot reloaded module: {module_name}")

    except Exception as e:
        self.logger.error(f"Failed to hot reload module {module_name}: {e}")

Handler Tracking

The EventBus tracks handlers by module for clean unregistration:

class EventBus:
    def __init__(self):
        self.handler_registry = HandlerRegistry()
        self._module_handlers: dict[str, list[str]] = {}

    async def unregister_module_handlers(self, module_name: str) -> None:
        """Remove all handlers registered by a specific module"""
        if module_name in self._module_handlers:
            for handler_id in self._module_handlers[module_name]:
                self.handler_registry.unregister(handler_id)
            del self._module_handlers[module_name]

State Preservation

During hot reloads, critical state is preserved:

  • Twitch WebSocket connections remain active
  • Database connections stay open
  • Service container maintains registered services
  • Event bus configuration is preserved
  • Web server continues running

Best Practices

Module Design for Hot Reloading

✅ Do:

# Stateless handlers
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!counter"))
async def counter_command(event: Event, twitch: TwitchClient, db: Database):
    # Get count from database each time
    count = await db.get_variable("counter", 0)
    await db.set_variable("counter", count + 1)
    await twitch.send_message(f"Count: {count + 1}")

# Use dependency injection
async def my_handler(event: Event, economy: EconomyService, logger: logging.Logger):
    # Services are injected fresh on each reload
    pass

# Module-level configuration
class MyModule(BaseModule):
    def __init__(self):
        super().__init__()
        self.config = self.load_config()  # Reloaded with module

❌ Avoid:

# Module-level mutable state
user_cache = {}  # Won't reset on hot reload

# Global service imports
from starstreamer.services.economy import economy_service
async def handler(event: Event):
    # This reference won't update on core changes
    pass

# Background tasks without cleanup
asyncio.create_task(background_worker())  # Orphaned on reload

Error Handling

Always handle reload failures gracefully:

try:
    await registry.reload_module("problematic_module")
except ImportError as e:
    logger.error(f"Syntax error in module: {e}")
    # Previous version continues running
except Exception as e:
    logger.error(f"Reload failed: {e}")
    # Module remains in previous working state

Testing Hot Reloads

  1. Make small, safe changes first: 

    # Add a timestamp to verify reload
await twitch.send_message(f"Hello! (reloaded at {time.time()})")

  2. Test error handling: 

    # Introduce a syntax error, save, then fix it
await twitch.send_message("Test"  # Missing closing parenthesis

  3. Verify state preservation: 

    # Check that connections remain active during reload
curl http://localhost:8888/api/status

Performance Considerations

Hot Reload Performance

  • Reload time: ~200ms for typical modules
  • Memory usage: Minimal increase during reload
  • Connection stability: No interruption to external connections
  • Event loop integration: Uses asyncio.run_coroutine_threadsafe() for thread-safe scheduling

File Watcher Overhead

  • CPU usage: <1% during file monitoring
  • Memory usage: ~10MB for watchdog processes
  • Debouncing: Prevents excessive reloads during rapid edits

Module Size Impact

Module Size Reload Time Memory Impact
Small (1-5 files) ~100ms <1MB
Medium (5-15 files) ~200ms 1-5MB
Large (15+ files) ~500ms 5-10MB

Troubleshooting

Common Issues

Module not reloading: 

# Check if file is being watched
uv run python src/main.py --reload --log-level DEBUG
# Look for "📝 Detected change in..." messages

Syntax errors preventing reload: 

# Error is logged, previous version continues
ERROR: Failed to hot reload module chat: invalid syntax (commands.py, line 15)

Handler not responding after reload: 

# Check if handler registration failed
WARNING: Handler registration failed for module chat

Memory leaks during development: 

# Restart periodically during heavy development
# Hot reloads preserve memory, but imports accumulate

Debug Information

Enable detailed logging for troubleshooting:

# Full debug output
uv run python src/main.py --reload --log-level DEBUG

# Watch specific directories
uv run python src/main.py --reload --watch-dirs modules

# Monitor file system events
uv run python src/main.py --reload --log-level DEBUG 2>&1 | grep "📝 Detected"

Recovery Procedures

If hot reload gets stuck: 1. Make a trivial change to force a reload 2. Check logs for error messages 3. Restart manually if needed: Ctrl+C → uv run python src/main.py --reload

If module state becomes inconsistent: 1. Force full restart by modifying a core file 2. Or restart manually to clean state

Advanced Usage

Web Interface Module Management

StarStreamer v0.6.1+ provides a web interface for interactive module management:

  1. Access the Modules page:
  2. Open http://localhost:8888/modules

  3. View real-time module statistics and status

  4. Available actions:

  5. Reload - Hot reload module code (same as automatic file watching)
  6. Enable/Disable - Toggle module functionality without reloading

  7. Load/Unload - Add/remove modules from memory

  8. Development benefits:

  9. Visual feedback - See which modules are loaded and enabled
  10. Manual control - Override automatic reloading when needed
  11. Status monitoring - Track reload success/failure in real-time

Selective Module Reloading

You can reload specific modules programmatically:

from modules.registry import ModuleRegistry

# In a web endpoint or CLI command
await registry.reload_module("chat")
await registry.reload_module("economy")

Custom File Watching

Extend the file watcher for custom needs:

class CustomReloadHandler(HybridReloadHandler):
    def on_modified(self, event: FileSystemEvent) -> None:
        if "config.yaml" in event.src_path:
            self._reload_configuration()
        else:
            super().on_modified(event)

Integration with IDEs

VS Code settings for optimal experience: 

{
    "python.defaultInterpreterPath": ".venv/bin/python",
    "python.terminal.activateEnvironment": false,
    "files.watcherExclude": {
        "**/__pycache__/**": true,
        "**/data/*.db*": true
    }
}

PyCharm configuration: - Exclude __pycache__ directories from indexing - Set up run configuration with --reload flag - Enable auto-save for immediate feedback

See Also