plugin_executor.py

"""
Plugin Executor

Handles plugin execution (update() and display() calls) with timeout handling,
error isolation, and performance monitoring.
"""

import time
from typing import Any, Dict, Optional, Callable
from threading import Thread
import logging

from src.common.fetch_service import plugin_scope
from src.exceptions import PluginError
from src.logging_config import get_logger
from src.error_aggregator import record_error


class PluginTimeoutError(Exception):
    """Raised when a plugin operation times out."""


class PluginBusyError(PluginTimeoutError):
    """A plugin's lock stayed held past its bound.

    Not raised; recorded. The lock is held by the plugin's own display(),
    update(), on_config_change() or a Vegas content render -- slow, or hung
    -- so the caller skipped the plugin rather than wait on it. Report-only:
    it is kept as the plugin's state error info and counted as a busy skip in
    health, never as a failure, so it cannot open the circuit breaker.
    """


class PluginExecutor:
    """Handles plugin execution with timeout and error isolation."""

    #: A display() call at least this long is logged and counted as slow.
    #: A frame is milliseconds; two seconds is a plugin doing I/O in display().
    SLOW_DISPLAY_SECONDS = 2.0
    #: An update() call at least this long is logged as slow.
    SLOW_UPDATE_SECONDS = 5.0
    
    def __init__(
        self,
        default_timeout: float = 30.0,
        logger: Optional[logging.Logger] = None
    ) -> None:
        """
        Initialize the plugin executor.
        
        Args:
            default_timeout: Default timeout in seconds for plugin operations
            logger: Optional logger instance
        """
        self.default_timeout = default_timeout
        self.logger = logger or get_logger(__name__)
    
    def execute_with_timeout(
        self,
        operation: Callable[[], Any],
        timeout: Optional[float] = None,
        plugin_id: Optional[str] = None,
        thread_name: Optional[str] = None
    ) -> Any:
        """
        Execute a plugin operation with timeout.
        
        Args:
            operation: Function to execute
            timeout: Timeout in seconds (None = use default)
            plugin_id: Optional plugin ID for logging
            thread_name: Name for the thread the operation runs on (None
                keeps Python's default). Stack dumps list threads by name.
            
        Returns:
            Result of operation
            
        Raises:
            PluginTimeoutError: If operation times out
            PluginError: If operation raises an exception
        """
        timeout = timeout or self.default_timeout
        plugin_context = f"plugin {plugin_id}" if plugin_id else "plugin"
        
        # Use threading-based timeout (more reliable than signal-based)
        result_container: Dict[str, Any] = {'value': None, 'exception': None, 'completed': False}
        
        def target():
            try:
                # Fetches made by the operation (and by threads the core
                # starts from it) are counted against this plugin.
                with plugin_scope(plugin_id):
                    result_container['value'] = operation()
                result_container['completed'] = True
            except BaseException as e:  # pylint: disable=broad-except
                # asyncio.CancelledError and SystemExit too: uncaught, one
                # ended this thread with 'completed' unset, and an operation
                # that failed at once was reported as timing out.
                result_container['exception'] = e
                result_container['completed'] = True
        
        thread = Thread(target=target, daemon=True, name=thread_name)
        thread.start()
        thread.join(timeout=timeout)
        
        # NB: this timeout is advisory. Nothing cancels the thread -- Python
        # has no way to -- so on expiry the operation keeps running to
        # completion in the background and only this caller gives up waiting.
        # A plugin that hangs permanently leaks one daemon thread per attempt.
        # Callers that hold a resource across the call must release it from
        # inside the wrapped callable rather than after this returns; see the
        # _release_display_lock guard inside DisplayController.run().
        if not result_container['completed']:
            error_msg = f"{plugin_context} operation timed out after {timeout}s"
            self.logger.error(error_msg)
            timeout_error = PluginTimeoutError(error_msg)
            record_error(timeout_error, plugin_id=plugin_id, operation="timeout")
            raise timeout_error

        if result_container['exception']:
            error = result_container['exception']
            error_msg = f"{plugin_context} operation failed: {error}"
            self.logger.error(error_msg, exc_info=error)
            record_error(error, plugin_id=plugin_id, operation="execute")
            raise PluginError(error_msg, plugin_id=plugin_id) from error
        
        return result_container['value']
    
    def execute_update(
        self,
        plugin: Any,
        plugin_id: str,
        timeout: Optional[float] = None
    ) -> bool:
        """
        Execute plugin update() method with error handling.
        
        Args:
            plugin: Plugin instance
            plugin_id: Plugin identifier
            timeout: Timeout in seconds (None = use default)
            
        Returns:
            True if update succeeded, False otherwise
        """
        try:
            start_time = time.monotonic()
            self.execute_with_timeout(
                lambda: plugin.update(),
                timeout=timeout,
                plugin_id=plugin_id
            )
            duration = time.monotonic() - start_time
            
            if duration > self.SLOW_UPDATE_SECONDS:
                self.logger.warning(
                    "Plugin %s update() took %.2fs (consider optimizing)",
                    plugin_id,
                    duration
                )
            
            return True
        except PluginTimeoutError:
            self.logger.error("Plugin %s update() timed out", plugin_id)
            return False
        except PluginError:
            # Already logged and recorded in execute_with_timeout
            return False
        except Exception as e:
            self.logger.error(
                "Unexpected error executing update() for plugin %s: %s",
                plugin_id,
                e,
                exc_info=True
            )
            record_error(e, plugin_id=plugin_id, operation="update")
            return False
    
    def execute_display(
        self,
        plugin: Any,
        plugin_id: str,
        force_clear: bool = False,
        display_mode: Optional[str] = None,
        timeout: Optional[float] = None,
        accepts_display_mode: Optional[bool] = None,
        raise_errors: bool = False
    ) -> bool:
        """
        Execute plugin display() method with error handling.
        
        Args:
            plugin: Plugin instance
            plugin_id: Plugin identifier
            force_clear: Whether to force clear display
            display_mode: Optional display mode parameter
            timeout: Timeout in seconds (None = use default)
            accepts_display_mode: Whether plugin.display() takes a
                display_mode keyword. Pass it when the caller already knows;
                None falls back to inspecting the callable.
            raise_errors: Re-raise the PluginError wrapping an exception
                display() raised, instead of returning False. False alone
                cannot tell "no content" from "raised", and a caller that
                feeds the circuit breaker needs that difference. The error
                is still logged and recorded first. A timeout still returns
                False either way.
            
        Returns:
            True if display succeeded, False otherwise

        Raises:
            PluginError: Only with ``raise_errors``, when display() raised.
        """
        try:
            start_time = time.monotonic()
            
            # Does display() take a display_mode keyword? The caller usually
            # knows and caches the answer, so prefer what it passed.
            #
            # Inspecting here was not merely redundant, it could never be
            # cached: display_controller wraps the real plugin in a fresh
            # SimpleNamespace per call, so inspect.signature() saw a new
            # callable every time and paid ~55us on a Pi 4 to re-derive a
            # value the caller had computed one line earlier and stored in
            # self._plugin_accepts_display_mode.
            if accepts_display_mode is None:
                import inspect
                accepts_display_mode = (
                    'display_mode' in inspect.signature(plugin.display).parameters)
            has_display_mode = accepts_display_mode
            
            # Named for the plugin: this thread presents a screen's first
            # frame, so the frame-timing stall watchdog's stack dumps name it.
            thread_name = f"display-{plugin_id}"

            # Capture the return value from the plugin's display() method
            if has_display_mode and display_mode:
                result = self.execute_with_timeout(
                    lambda: plugin.display(display_mode=display_mode, force_clear=force_clear),
                    timeout=timeout,
                    plugin_id=plugin_id,
                    thread_name=thread_name
                )
            else:
                result = self.execute_with_timeout(
                    lambda: plugin.display(force_clear=force_clear),
                    timeout=timeout,
                    plugin_id=plugin_id,
                    thread_name=thread_name
                )
            
            duration = time.monotonic() - start_time
            
            if duration > self.SLOW_DISPLAY_SECONDS:
                self.logger.warning(
                    "Plugin %s display() took %.2fs (consider optimizing)",
                    plugin_id,
                    duration
                )
            
            # Return the actual result from the plugin's display() method
            # If it's a boolean, use it directly. Otherwise, treat None/other as True for backward compatibility
            if isinstance(result, bool):
                self.logger.debug(f"Plugin {plugin_id} display() returned boolean: {result}")
                return result
            # For backward compatibility: if plugin returns None or something else, treat as success
            self.logger.debug(f"Plugin {plugin_id} display() returned non-boolean: {result}, treating as True")
            return True
        except PluginTimeoutError:
            self.logger.error("Plugin %s display() timed out", plugin_id)
            return False
        except PluginError:
            # Already logged and recorded in execute_with_timeout
            if raise_errors:
                raise
            return False
        except Exception as e:
            self.logger.error(
                "Unexpected error executing display() for plugin %s: %s",
                plugin_id,
                e,
                exc_info=True
            )
            record_error(e, plugin_id=plugin_id, operation="display")
            return False