Добавить защиту от одновременного запуска нескольких экземпляров приложения

[chelslava]

Описание

Приложение может быть запущено несколько раз параллельно, что приводит к конфликтам при захвате экрана. Windows Graphics Capture API поддерживает только один активный сеанс на приложение.

Текущее поведение

User 1: Запуск MIA → Захват активен
User 2: Запуск MIA → Захват конфликтует → Один из экземпляров падает
User 3: Запуск MIA → Конфигурация конфликтует
Result: Непредсказуемое поведение

Ожидаемое поведение

User: Запуск MIA → Проверка мьютекса → Другой экземпляр найден → Фокус на него
                                              ↓
                              Другого экземпляра нет → Нормальный запуск
Result: Только один экземпляр

Технические требования

1. Мьютекс для одиночного экземпляра

import sys
import win32event
import win32api
import winerror
from pathlib import Path

class SingleInstanceGuard:
    """Гарантирует запуск только одного экземпляра приложения.
    
    Использует Windows Named Event для межпроцессной синхронизации.
    """
    
    MUTEX_NAME = "MIA-ScreenCapture-SingleInstance-Mutex"
    IPC_CHANNEL = "MIA-ScreenCapture-IPC-Pipe"
    
    def __init__(self):
        self._handle: int | None = None
    
    def acquire(self) -> bool:
        """Пытается захватить мьютекс.
        
        Returns:
            True если захвачен (первый экземпляр)
            False если уже занят (второй экземпляр)
        """
        self._handle = win32event.CreateEventW(
            None,      # атрибуты безопасности
            True,      # manual reset
            False,     # initial state = not signaled
            self.MUTEX_NAME
        )
        
        # WAIT_TIMEOUT = 0, meaning already signaled
        result = win32event.WaitForSingleObject(
            self._handle,
            0  # не ждать
        )
        
        if result == winerror WAIT_TIMEOUT:
            # Другой экземпляр работает
            win32api.CloseHandle(self._handle)
            self._handle = None
            return False
        
        return True
    
    def release(self) -> None:
        """Освобождает мьютекс."""
        if self._handle:
            win32event.SetEvent(self._handle)
            win32api.CloseHandle(self._handle)
            self._handle = None
    
    def __enter__(self) -> "SingleInstanceGuard":
        if not self.acquire():
            raise AnotherInstanceRunningError()
        return self
    
    def __exit__(self, *args) -> None:
        self.release()

2. IPC для передачи команд

import win32pipe
import win32file
import json
import threading

class IPCBridge:
    """Канал связи между экземплярами.
    
    Позволяет:
    - Передать фокус другому экземпляру
    - Переслать команду (start/stop)
    - Получить статус
    """
    
    def __init__(self, instance_id: str):
        self._instance_id = instance_id
        self._pipe_name = f"\\\\.\\pipe\\{self.IPC_CHANNEL}"
        self._server_thread: threading.Thread | None = None
    
    def start_server(self, callback: Callable[[dict], dict]) -> None:
        """Запускает сервер для приёма команд."""
        self._callback = callback
        self._server_thread = threading.Thread(
            target=self._server_loop,
            daemon=True
        )
        self._server_thread.start()
    
    def connect_and_send(self, command: dict) -> dict | None:
        """Подключается к существующему экземпляру и отправляет команду."""
        try:
            handle = win32file.CreateFile(
                self._pipe_name,
                win32file.GENERIC_READ | win32file.GENERIC_WRITE,
                0,
                None,
                win32file.OPEN_EXISTING,
                0,
                None
            )
            
            # Отправить команду
            data = json.dumps(command).encode("utf-8")
            win32file.WriteFile(handle, data)
            
            # Получить ответ
            _, response = win32file.ReadFile(handle, 4096)
            return json.loads(response.decode("utf-8"))
            
        except Exception:
            return None
    
    def _server_loop(self) -> None:
        """Цикл сервера."""
        while True:
            try:
                handle = win32pipe.CreateNamedPipe(
                    self._pipe_name,
                    win32pipe.PIPE_ACCESS_DUPLEX,
                    win32pipe.PIPE_TYPE_MESSAGE | win32pipe.PIPE_READMODE_MESSAGE,
                    1,
                    65536,
                    65536,
                    0,
                    None
                )
                
                win32pipe.ConnectNamedPipe(handle, None)
                
                # Читать команду
                _, data = win32file.ReadFile(handle, 4096)
                command = json.loads(data.decode("utf-8"))
                
                # Обработать
                response = self._callback(command)
                
                # Ответить
                response_data = json.dumps(response).encode("utf-8")
                win32file.WriteFile(handle, response_data)
                
                win32file.CloseHandle(handle)
                
            except Exception:
                time.sleep(0.1)

3. Интеграция с main.py

def main():
    """Точка входа приложения."""
    try:
        with SingleInstanceGuard() as guard:
            # Это первый экземпляр - запускаем normally
            app = VideoRecorderApp()
            
            # Запускаем IPC сервер
            ipc = IPCBridge(get_instance_id())
            ipc.start_server(app.handle_ipc_command)
            
            app.run()
            
    except AnotherInstanceRunningError:
        # Второй экземпляр - переключаем фокус
        ipc = IPCBridge("")
        response = ipc.connect_and_send({
            "action": "bring_to_front"
        })
        
        if response is None:
            # Экземпляр завис, показываем диалог
            QMessageBox.warning(
                None,
                "MIA-ScreenCapture",
                "Приложение уже запущено."
            )
        sys.exit(0)

4. Обработка в главном экземпляре

class VideoRecorderApp:
    def handle_ipc_command(self, command: dict) -> dict:
        """Обработка команд от других экземпляров."""
        action = command.get("action")
        
        if action == "bring_to_front":
            # Переключить фокус на главное окно
            self.activateWindow()
            self.raise_()
            self.showNormal()
            return {"success": True, "focused": True}
        
        elif action == "start_recording":
            self.recording_controller.start()
            return {"success": True}
        
        elif action == "get_status":
            return {"success": True, "status": self.get_status()}
        
        return {"success": False, "error": "Unknown action"}

Критерии приёмки

• Только один экземпляр запускается одновременно
• Второй экземпляр переключает фокус на существующий
• Команды CLI корректно передаются существующему экземпляру
• Работает с системным треем (многократные клики)
• Graceful handling при зависании первого экземпляра
• Unit-тесты для SingleInstanceGuard

Файлы для изменения

• main.py — интеграция одиночного экземпляра
• utils.py — SingleInstanceGuard, IPCBridge
• app_runtime/ — обработка IPC команд
• tests/unit/test_main_entrypoint.py — тесты

Приоритет: P1

Тэги: reliability, security