From d9793286d13b87f41a32b21a500ab9aac3b993f8 Mon Sep 17 00:00:00 2001 From: Loren Eteval Date: Sun, 30 Aug 2026 17:09:05 +0800 Subject: [PATCH] refactor: make connection startup asynchronous Replace GUI connection startup waits with a cancellable, staged Qt transaction. Observe core endpoints, DNS resolution, TUN device readiness, and runtime survival without nested event-loop waits while preserving the synchronous plugin compatibility path. Commit runtimes only after every required stage succeeds, reject stale generations, and cover readiness, cancellation, rollback, platform ordering, and controller integration. Signed-off-by: Loren Eteval --- Furious/Backends/Hysteria1/Plugin.py | 1 + Furious/Backends/Hysteria2/Plugin.py | 1 + Furious/Backends/Xray/Plugin.py | 7 +- Furious/Controllers/ConnectionController.py | 127 ++- Furious/Core/CoreProcessWorker.py | 28 +- Furious/Interface/Runtime.py | 8 + Furious/Plugins/API.py | 18 +- Furious/Plugins/__init__.py | 2 + Furious/Service/AGENTS.md | 7 +- Furious/Service/ConnectionManager.py | 1034 ++++++++++++++++++- Furious/Service/DnsResolver.py | 128 ++- Furious/Service/__init__.py | 8 +- tests/README.md | 6 +- tests/test_connection_startup_async.py | 786 ++++++++++++++ tests/test_controllers.py | 146 +++ 15 files changed, 2280 insertions(+), 27 deletions(-) create mode 100644 tests/test_connection_startup_async.py diff --git a/Furious/Backends/Hysteria1/Plugin.py b/Furious/Backends/Hysteria1/Plugin.py index 17a368d..99b3749 100644 --- a/Furious/Backends/Hysteria1/Plugin.py +++ b/Furious/Backends/Hysteria1/Plugin.py @@ -94,6 +94,7 @@ class Hysteria1CoreRuntimeFactory(CoreRuntimeFactory): Hysteria1.mmdb(routingObject.get('mmdb', '')), ), options=request.options, + startup=CoreRuntimeStartup(endpoint=config.httpProxy()), ) def prepareDownloadTest(self, config, port: int): diff --git a/Furious/Backends/Hysteria2/Plugin.py b/Furious/Backends/Hysteria2/Plugin.py index 2ecf13f..c4b034f 100644 --- a/Furious/Backends/Hysteria2/Plugin.py +++ b/Furious/Backends/Hysteria2/Plugin.py @@ -211,6 +211,7 @@ class Hysteria2CoreRuntimeFactory(CoreRuntimeFactory): runtime, request.configuration, options=request.options, + startup=CoreRuntimeStartup(endpoint=request.configuration.httpProxy()), ) def prepareDownloadTest(self, config, port: int): diff --git a/Furious/Backends/Xray/Plugin.py b/Furious/Backends/Xray/Plugin.py index b9add89..738d06a 100644 --- a/Furious/Backends/Xray/Plugin.py +++ b/Furious/Backends/Xray/Plugin.py @@ -327,7 +327,12 @@ class XrayCoreRuntimeFactory(CoreRuntimeFactory): ) runtime.xrayStatsTarget = statsTarget - return CoreRuntimeLaunch(runtime, config, options=request.options) + return CoreRuntimeLaunch( + runtime, + config, + options=request.options, + startup=CoreRuntimeStartup(endpoint=config.httpProxy()), + ) def prepareDownloadTest(self, config, port: int): """Create an Xray configuration with one local HTTP test inbound.""" diff --git a/Furious/Controllers/ConnectionController.py b/Furious/Controllers/ConnectionController.py index 3424f96..bcd44d7 100644 --- a/Furious/Controllers/ConnectionController.py +++ b/Furious/Controllers/ConnectionController.py @@ -24,6 +24,7 @@ from Furious.Interface import * from Furious.Models import ServerProfile from Furious.Plugins import getPluginRegistry from Furious.Qt.DynamicTranslate import gettext as _ +from Furious.Qt.Signals import connectWeakly from Furious.Repository import Storage from Furious.Service import ( CORE_LOG_CATEGORY, @@ -102,6 +103,8 @@ class ConnectionController(QtCore.QObject): self._state = ConnectionState.Disconnected self._activeProfile = None self._lastError = None + self._startOperation = None + self._pendingHttpProxy = '' self._actionTimer = QtCore.QTimer(self) self._actionTimer.timeout.connect(self._callActionFromQueue) @@ -187,6 +190,8 @@ class ConnectionController(QtCore.QObject): def _reset(self): """Restore disconnected state after all runtime resources stop.""" + self._startOperation = None + self._pendingHttpProxy = '' self.progressFinished.emit(True) self._setActiveProfile(None) @@ -269,12 +274,57 @@ class ConnectionController(QtCore.QObject): self._lastError = None self._setActiveProfile(configuration) + self._pendingHttpProxy = httpProxy self._startConnecting() logManager = AppLogManager() # Retain application diagnostics while starting a fresh runtime log. logManager.clear(runtimeOnly=True) + startAsync = getattr(self._coreManager, 'startAsync', None) + + if callable(startAsync): + try: + operation = startAsync( + configuration, + routing=AppSettings.get('Routing'), + exitCallback=self.coreExitCallback, + msgCallbackCore=logManager.callback(CORE_LOG_CATEGORY), + msgCallbackTUN_=logManager.callback( + TUN2SOCKS_LOG_CATEGORY, + source='Tun2socks', + ), + ) + except Exception as ex: + logger.error(f'failed to schedule core manager startup: {ex}') + + return self._failConnection( + f'{configuration.coreName()}: ' + _('Unknown error'), + str(ex), + ) + + self._startOperation = operation + connectWeakly( + operation.succeeded, + self, + '_connectionStartSucceeded', + sender=operation, + ) + connectWeakly( + operation.failed, + self, + '_connectionStartFailed', + sender=operation, + ) + connectWeakly( + operation.cancelled, + self, + '_connectionStartCancelled', + sender=operation, + ) + + return True + startExceptionDetails = '' try: @@ -315,6 +365,10 @@ class ConnectionController(QtCore.QObject): startExceptionDetails, ) + return self._finishConnection(configuration, httpProxy) + + def _finishConnection(self, configuration, httpProxy) -> bool: + """Commit system integration after manager runtime ownership commits.""" settings = AppSettings.get('CustomProxyBypass') proxyServerBypass = ( @@ -352,14 +406,80 @@ class ConnectionController(QtCore.QObject): return True + @QtCore.Slot(object) + def _connectionStartSucceeded(self, operation): + """Finish the exact manager generation that committed successfully.""" + if operation is not self._startOperation or not self.isConnecting(): + return + + self._startOperation = None + self._emitRuntimesChanged() + + while not self._actionQueue.empty(): + self._callActionFromQueue() + + if not self.isConnecting(): + return + + configuration = self.activeProfile + + if configuration is None: + self._failConnection(_('Unknown error')) + + return + + self._finishConnection(configuration, self._pendingHttpProxy) + + @QtCore.Slot(object, str, str) + def _connectionStartFailed(self, operation, message, details): + """Return one failed manager generation to stable disconnected state.""" + if operation is not self._startOperation: + return + + self._startOperation = None + self._emitRuntimesChanged() + + configuration = self.activeProfile + coreName = configuration.coreName() if configuration is not None else '' + startError = message or getattr(self._coreManager, 'lastStartError', '') + displayMessage = (f'{coreName}: ' if coreName else '') + ( + _(startError) if startError else _('Unknown error') + ) + + self._failConnection(displayMessage, details) + + @QtCore.Slot(object) + def _connectionStartCancelled(self, operation): + """Ignore stale cancellation or reset an externally cancelled start.""" + if operation is not self._startOperation: + return + + self._startOperation = None + self._emitRuntimesChanged() + self._reset() + def startDisconnection(self, notification: str = '') -> bool: """Stop the active runtime and optionally request a notification.""" if self.state is ConnectionState.Disconnected: return False + operation = self._startOperation + + self._startOperation = None self._setState(ConnectionState.Disconnecting) self._actionTimer.stop() + if operation is not None: + cancelStart = getattr(self._coreManager, 'cancelStart', None) + + if callable(cancelStart): + try: + cancelStart(operation) + except Exception as ex: + # Any non-exit exceptions + + logger.error(f'failed to cancel connection startup: {ex}') + try: SystemProxy.off() except Exception as ex: @@ -390,13 +510,10 @@ class ConnectionController(QtCore.QObject): def startReconnection(self, notification: str = '') -> bool: """Restart the active repository profile when lifecycle state permits.""" - if self.state in ( - ConnectionState.Connecting, - ConnectionState.Disconnecting, - ): + if self.state is ConnectionState.Disconnecting: return False - if self.isConnected(): + if self.isConnected() or self.isConnecting(): self.startDisconnection(notification) return self.startConnection() diff --git a/Furious/Core/CoreProcessWorker.py b/Furious/Core/CoreProcessWorker.py index 4460d00..58dd114 100644 --- a/Furious/Core/CoreProcessWorker.py +++ b/Furious/Core/CoreProcessWorker.py @@ -64,7 +64,12 @@ class CoreProcessState(Enum): @dataclass class CoreLaunchSpec: - """Describe the parameters required by a core launch operation.""" + """Describe the parameters required by a core launch operation. + + 'waitCore' preserves the synchronous plugin compatibility path. Normal GUI + connections launch with it disabled and confirm readiness through the + connection manager's event-driven startup transaction. + """ DefaultWaitTime: ClassVar[int] = 2500 @@ -437,20 +442,35 @@ class CoreProcessWorker(CoreProcessMonitor, ABC): self.msgQueue.setTimeout(MsgQueue.ACTIVE_DRAIN_INTERVAL) self.msgQueue.startTimer() + # Monitor the process during asynchronous startup as well. A child that + # exits before readiness must fail the owning transaction instead of + # waiting for a later connected-state health check. + self.daemon.start(CORE_CHECK_ALIVE_INTERVAL) + if launchSpec.waitCore: # Wait for the core to start up completely PySide6Legacy.eventLoopWait(launchSpec.waitTime) if self.queryIsAlive(): + # Preserve the low-level compatibility contract: a successfully + # spawned process is Running even when its caller disables the + # historical grace wait. The connection transaction separately + # owns semantic readiness and final commit. self.setState(CoreProcessState.Running) - # Start core daemon - self.daemon.start(CORE_CHECK_ALIVE_INTERVAL) - return True else: return False + def confirmStartup(self) -> bool: + """Move a live externally observed process into the running state.""" + if not self.isAlive(): + return False + + self.setState(CoreProcessState.Running) + + return True + def stop(self): """Stop the core process worker.""" self.msgQueue.stopTimer() diff --git a/Furious/Interface/Runtime.py b/Furious/Interface/Runtime.py index c0bbe3c..77cbe45 100644 --- a/Furious/Interface/Runtime.py +++ b/Furious/Interface/Runtime.py @@ -69,6 +69,14 @@ class CoreRuntime(ABC): if callable(self._exitCallback): self._exitCallback(self, exitcode) + def setExitCallback(self, callback): + """Transfer termination reporting to the runtime's current owner.""" + self._exitCallback = callback + + def confirmStartup(self) -> bool: + """Confirm readiness after an external startup observer succeeds.""" + return True + def startError(self) -> str: """Return the most recent concise startup failure, if any.""" return self._startError diff --git a/Furious/Plugins/API.py b/Furious/Plugins/API.py index fc3491c..c1064dd 100644 --- a/Furious/Plugins/API.py +++ b/Furious/Plugins/API.py @@ -30,6 +30,7 @@ __all__ = [ 'CoreRuntimeFactory', 'CoreRuntimeLaunch', 'CoreRuntimeRequest', + 'CoreRuntimeStartup', 'FuriousPlugin', 'NavigationPageDescriptor', 'NavigationPageProvider', @@ -401,6 +402,15 @@ class CoreRuntimeRequest: options: Mapping[str, Any] = field(default_factory=dict) +@dataclass(frozen=True) +class CoreRuntimeStartup: + """Describe an optional event-driven readiness contract for one runtime.""" + + endpoint: str = '' + timeout: int = 2500 + retryInterval: int = 50 + + @dataclass(frozen=True) class CoreRuntimeLaunch: """Bind a constructed core runtime to its prepared start arguments.""" @@ -409,14 +419,18 @@ class CoreRuntimeLaunch: configuration: Any arguments: Tuple[Any, ...] = tuple() options: Mapping[str, Any] = field(default_factory=dict) + startup: Optional[CoreRuntimeStartup] = None - def start(self) -> bool: + def start(self, **optionOverrides) -> bool: """Start the prepared core runtime.""" + options = dict(self.options) + options.update(optionOverrides) + return bool( self.runtime.start( self.configuration, *self.arguments, - **dict(self.options), + **options, ) ) diff --git a/Furious/Plugins/__init__.py b/Furious/Plugins/__init__.py index d468c6d..1c163a0 100644 --- a/Furious/Plugins/__init__.py +++ b/Furious/Plugins/__init__.py @@ -26,6 +26,7 @@ from .API import ( CoreRuntimeFactory, CoreRuntimeLaunch, CoreRuntimeRequest, + CoreRuntimeStartup, FuriousPlugin, NavigationPageDescriptor, NavigationPageProvider, @@ -74,6 +75,7 @@ __all__ = [ 'CoreRuntimeFactory', 'CoreRuntimeLaunch', 'CoreRuntimeRequest', + 'CoreRuntimeStartup', 'FuriousPlugin', 'NavigationPageDescriptor', 'NavigationPageProvider', diff --git a/Furious/Service/AGENTS.md b/Furious/Service/AGENTS.md index bd0ea7b..a6f5fdc 100644 --- a/Furious/Service/AGENTS.md +++ b/Furious/Service/AGENTS.md @@ -12,8 +12,11 @@ ## Runtime and asynchronous invariants - `ConnectionManager` consumes an attempt-scoped copy, asks the selected factory for native-TUN/application-tun2socks - policy, and owns exact runtimes only after commit. On failure it rolls back only resources acquired by that attempt and - restores host routing/DNS state through those owners. + policy, and owns exact runtimes only after commit. Normal built-in GUI startup belongs to one generation-checked + `ConnectionStartOperation`: launch, local-endpoint readiness, asynchronous DNS, TUN-device observation, host mutation, + rollback, and commit advance through owned Qt timers/signals without a nested event loop. The synchronous `start()` + path is a compatibility boundary for legacy plugins/non-interactive callers, not the preferred GUI path. On failure + or cancellation, roll back only resources acquired by that attempt and restore host routing/DNS through those owners. - Every async workflow defines supersession: generation/version checks for stale completion or exact-object identity for independent requests. Terminal paths release context, finish/abort once, and schedule each Qt reply for deletion. Callbacks cannot retain a shut-down manager; worker results cross into the manager's Qt thread before mutation. diff --git a/Furious/Service/ConnectionManager.py b/Furious/Service/ConnectionManager.py index 173b2b4..cd9df7e 100644 --- a/Furious/Service/ConnectionManager.py +++ b/Furious/Service/ConnectionManager.py @@ -31,24 +31,33 @@ from Furious.Frozenlib import ( SystemRoutingTable, SystemRuntime, isValidIPAddress, + parseHostPort, ) from Furious.Interface import CoreRuntime from Furious.Models import CoreConfiguration, ServerProfile from Furious.Repository import Storage -from Furious.Plugins import TUNPreparationError, getPluginRegistry +from Furious.Plugins import ( + CoreRuntimeLaunch, + CoreRuntimeStartup, + TUNPreparationError, + getPluginRegistry, +) +from Furious.Qt.Signals import connectWeakly, singleShotWeakly from Furious.Service.DnsResolver import DnsResolver from Furious.Core.CoreProcessWorker import CoreProcessWorker from Furious.Core.Tun2socks import Tun2socks from typing import Callable, Tuple, Union from dataclasses import dataclass, field +from enum import Enum +from PySide6 import QtCore, QtNetwork import os import logging import tempfile import functools -__all__ = ['ConnectionManager'] +__all__ = ['ConnectionManager', 'ConnectionStartOperation', 'ConnectionStartStage'] logger = logging.getLogger(__name__) @@ -80,6 +89,11 @@ class _ConnectionStartAttempt: return False for runtime in reversed(self.runtimes): + setExitCallback = getattr(runtime, 'setExitCallback', None) + + if callable(setExitCallback): + setExitCallback(None) + self.manager._stopAndDisposeRuntime(runtime) self.runtimes.clear() @@ -125,6 +139,960 @@ def getUserTUNSettings(*args, **kwargs): ) +class ConnectionStartStage(Enum): + """Describe the active stage of one connection startup transaction.""" + + Pending = 'pending' + Preparing = 'preparing' + StartingPrimary = 'starting-primary' + WaitingPrimary = 'waiting-primary' + PreparingTUN = 'preparing-tun' + ResolvingTUNAddress = 'resolving-tun-address' + WaitingTUNDevice = 'waiting-tun-device' + StartingTUNRuntime = 'starting-tun-runtime' + WaitingTUNRuntime = 'waiting-tun-runtime' + ApplyingHostNetwork = 'applying-host-network' + Committing = 'committing' + Succeeded = 'succeeded' + Failed = 'failed' + Cancelled = 'cancelled' + + +class _RuntimeReadinessProbe(QtCore.QObject): + """Observe process survival and an optional local TCP endpoint.""" + + ready = QtCore.Signal() + failed = QtCore.Signal(str) + + def __init__(self, runtime, startup, parent=None): + """Initialize a bounded readiness observer.""" + super().__init__(parent) + + self._runtime = runtime + self._startup = startup + self._terminal = False + self._host = '' + self._port = 0 + + self._elapsed = QtCore.QElapsedTimer() + self._timer = QtCore.QTimer(self) + self._timer.setInterval(max(int(startup.retryInterval), 1)) + + connectWeakly(self._timer.timeout, self, '_poll') + + self._socket = QtNetwork.QTcpSocket(self) + + connectWeakly(self._socket.connected, self, '_connected') + + def start(self): + """Begin observing without blocking or nesting the Qt event loop.""" + if self._terminal or self._timer.isActive(): + return + + if self._startup.endpoint: + try: + self._host, port = parseHostPort(self._startup.endpoint) + self._port = int(port) + except Exception: + # Any non-exit exceptions + + self._finishFailed( + f'invalid runtime readiness endpoint: ' + f'{self._startup.endpoint!r}' + ) + + return + + self._elapsed.start() + self._timer.start() + self._poll() + + def _runtimeAlive(self) -> bool: + """Return whether the observed runtime still owns a live process.""" + isAlive = getattr(self._runtime, 'isAlive', None) + + return bool(callable(isAlive) and isAlive()) + + def _poll(self): + """Retry endpoint connection and enforce the startup deadline.""" + if self._terminal: + return + + if not self._runtimeAlive(): + self._finishFailed('core process exited during startup') + + return + + timeout = max(int(self._startup.timeout), 1) + + if self._elapsed.isValid() and self._elapsed.elapsed() >= timeout: + if self._startup.endpoint: + self._finishFailed('core readiness check timed out') + else: + self._finishReady() + + return + + if ( + self._startup.endpoint + and self._socket.state() + is QtNetwork.QAbstractSocket.SocketState.UnconnectedState + ): + self._socket.connectToHost(self._host, self._port) + + def _connected(self): + """Accept the first successful local endpoint connection.""" + self._finishReady() + + def _finishReady(self): + """Publish readiness exactly once.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._socket.abort() + self.ready.emit() + + def _finishFailed(self, message): + """Publish startup failure exactly once.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._socket.abort() + self.failed.emit(str(message)) + + def cancel(self): + """Stop all probe resources without publishing a stale result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._socket.abort() + + +class _ConditionProbe(QtCore.QObject): + """Poll one host condition through a bounded parent-owned Qt timer.""" + + finished = QtCore.Signal(bool) + + def __init__(self, predicate, description, timeout=10000, parent=None): + """Initialize an idle condition observer.""" + super().__init__(parent) + + self._predicate = predicate + self._description = description + self._timeout = max(int(timeout), 1) + self._terminal = False + + self._elapsed = QtCore.QElapsedTimer() + self._timer = QtCore.QTimer(self) + self._timer.setInterval(100) + + connectWeakly(self._timer.timeout, self, '_poll') + + def start(self): + """Begin polling the condition.""" + if self._terminal or self._timer.isActive(): + return + + self._elapsed.start() + self._timer.start() + self._poll() + + def _poll(self): + """Publish the first successful observation or one timeout.""" + if self._terminal: + return + + try: + ready = bool(self._predicate()) + except Exception as ex: + # Any non-exit exceptions + + logger.error(f'failed to inspect {self._description}: {ex}') + + ready = False + + if ready: + logger.info(f'find {self._description} success') + + self._finish(True) + elif self._elapsed.isValid() and self._elapsed.elapsed() >= self._timeout: + logger.error(f'find {self._description} failed') + + self._finish(False) + + def _finish(self, success): + """Publish exactly one terminal condition result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self.finished.emit(bool(success)) + + def cancel(self): + """Stop polling without publishing a stale result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + + +class ConnectionStartOperation(QtCore.QObject): + """Own one cancellable, staged connection startup transaction.""" + + succeeded = QtCore.Signal(object) + failed = QtCore.Signal(object, str, str) + cancelled = QtCore.Signal(object) + stageChanged = QtCore.Signal(object) + + def __init__( + self, + manager, + generation, + config, + routing, + *, + exitCallback=None, + msgCallbackCore=None, + msgCallbackTUN_=None, + deepcopy=True, + proxyModeOnly=False, + log=True, + options=None, + parent=None, + ): + """Initialize an operation without starting any resource.""" + super().__init__(parent) + + self.manager = manager + self.generation = generation + self.routing = routing + self.exitCallback = exitCallback + self.msgCallbackCore = msgCallbackCore + self.msgCallbackTUN_ = msgCallbackTUN_ + self.proxyModeOnly = proxyModeOnly + self.log = log + self.options = dict(options or {}) + self.attempt = _ConnectionStartAttempt( + manager, manager._runtimeConfiguration(config, deepcopy) + ) + + self.stage = ConnectionStartStage.Pending + self._terminal = False + self._readinessProbe = None + self._conditionProbe = None + self._conditionContinuation = '' + self._dnsOperation = None + self._tun = None + self._startTUN = None + self._gateway = None + self._interface = None + + def _isCurrent(self): + """Return whether this generation still owns manager startup.""" + return ( + not self._terminal + and self.manager._activeStartOperation is self + and self.manager._startGeneration == self.generation + ) + + def _setStage(self, stage): + """Publish one observable startup stage.""" + if self.stage is stage: + return + + self.stage = stage + self.stageChanged.emit(stage) + + def _resume(self, methodName): + """Run one named continuation and translate exceptions into rollback.""" + if not self._isCurrent(): + return + + try: + getattr(self, methodName)() + except Exception as ex: + # Any non-exit exceptions + + self._fail('', str(ex)) + + def start(self): + """Start the semantic transaction after observers are connected.""" + if not self._isCurrent() or self.stage is not ConnectionStartStage.Pending: + return + + self._setStage(ConnectionStartStage.Preparing) + + configcopy = self.attempt.runtimeConfiguration + + try: + ( + self.attempt.nativeTUNHandled, + self.attempt.applicationTun2socks, + ) = self.manager._prepareTUNPolicy(configcopy, self.proxyModeOnly) + except TUNPreparationError as ex: + self.manager._lastStartError = str(ex) + self._fail(str(ex), 'native TUN preparation failed') + + return + except Exception as ex: + # Any non-exit exceptions + + self._fail('', str(ex)) + + return + + self._resume('_startPrimary') + + def _startPrimary(self): + """Construct and launch the primary runtime under attempt ownership.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.StartingPrimary) + + try: + launch = getPluginRegistry().createCoreRuntime( + self.attempt.runtimeConfiguration, + self.routing, + exitCallback=self.runtimeExitCallback, + messageCallback=self.msgCallbackCore, + proxyModeOnly=self.proxyModeOnly, + log=self.log, + **self.options, + ) + except Exception as ex: + # Any non-exit exceptions + + self._fail('', str(ex)) + + return + + if not isinstance(launch, CoreRuntimeLaunch): + self._fail('', 'no core runtime is available for this configuration') + + return + + runtime = launch.runtime + + self.attempt.ownRuntime(runtime) + + try: + success = ( + launch.start(waitCore=False) + if launch.startup is not None + else launch.start() + ) + except Exception as ex: + # Any non-exit exceptions + + self._fail(self._runtimeStartError(runtime), str(ex)) + + return + + if not success: + self._fail(self._runtimeStartError(runtime)) + + return + + if launch.startup is None: + self._resume('_afterPrimaryReady') + else: + self._observeRuntime( + runtime, + launch.startup, + '_afterPrimaryReady', + ConnectionStartStage.WaitingPrimary, + ) + + @staticmethod + def _runtimeStartError(runtime): + """Return one concise error published by a failed runtime.""" + startError = getattr(runtime, 'startError', None) + + return str(startError() or '') if callable(startError) else '' + + def _observeRuntime(self, runtime, startup, continuation, stage): + """Observe runtime readiness and resume through a named method.""" + if not self._isCurrent(): + return + + self._setStage(stage) + self._conditionContinuation = continuation + + probe = _RuntimeReadinessProbe(runtime, startup, parent=self) + + self._readinessProbe = probe + + connectWeakly(probe.ready, self, '_runtimeReady') + connectWeakly(probe.failed, self, '_runtimeReadinessFailed') + + probe.start() + + def _runtimeReady(self): + """Confirm the observed runtime and resume its continuation.""" + if not self._isCurrent() or self._readinessProbe is None: + return + + probe = self._readinessProbe + runtime = probe._runtime + continuation = self._conditionContinuation + + self._readinessProbe = None + self._conditionContinuation = '' + + probe.deleteLater() + + confirmStartup = getattr(runtime, 'confirmStartup', None) + + if callable(confirmStartup) and not confirmStartup(): + self._fail(self._runtimeStartError(runtime)) + + return + + self._resume(continuation) + + def _runtimeReadinessFailed(self, message): + """Fail the transaction when a readiness observer reaches terminal.""" + if not self._isCurrent(): + return + + self._readinessProbe = None + self._conditionContinuation = '' + self._fail('', message) + + def _afterPrimaryReady(self): + """Continue into application TUN or commit a proxy-only startup.""" + if not self._isCurrent(): + return + + if self.attempt.applicationTun2socks: + self._beginApplicationTun() + else: + self._commit() + + def runtimeExitCallback(self, runtime, exitcode): + """Abort when any attempt-owned runtime exits before commit.""" + if not self._isCurrent(): + return + + if exitcode == CoreRuntime.ExitCode.ConfigurationError.value: + message = 'Invalid server configuration' + elif exitcode == CoreRuntime.ExitCode.ServerStartFailure.value: + message = 'Failed to start core' + else: + try: + pluginMessage = getPluginRegistry().coreExitMessage(runtime, exitcode) + except Exception: + # Any non-exit exceptions + + pluginMessage = None + + message = pluginMessage or 'Core terminated unexpectedly' + + self._fail( + self._runtimeStartError(runtime) or message, + f'{runtime.name()} exited during startup with code {exitcode}', + ) + + def _beginApplicationTun(self): + """Acquire tun2socks and preserve the platform's startup ordering.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.PreparingTUN) + + configcopy = self.attempt.runtimeConfiguration + + if PLATFORM == 'Windows': + SystemRoutingTable.delete( + '0.0.0.0', + APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS, + ) + + userGateway = userDefaultPrimaryGatewayIP() + userInterfaceIP = userPrimaryAdapterInterfaceIP() + + if userGateway and userInterfaceIP: + logger.info( + f'got user defined TUN settings. ' + f'\'DefaultPrimaryGatewayIP\': {userGateway}. ' + f'\'PrimaryAdapterInterfaceIP\': {userInterfaceIP}' + ) + self._gateway, self._interface = userGateway, userInterfaceIP + else: + logger.info( + 'automatically fetching TUN settings: ' + '\'DefaultPrimaryGatewayIP\' and ' + '\'PrimaryAdapterInterfaceIP\'' + ) + defaultGateway = SystemRoutingTable.getDefaultGateway() + + if PLATFORM == 'Darwin': + defaultGateway = list( + filter( + lambda item: (item != APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS), + defaultGateway, + ) + ) + + if len(defaultGateway) != 1: + self._fail('', f'bad default gateway: {defaultGateway}') + + return + + if PLATFORM in ('Windows', 'Linux'): + self._gateway, self._interface = defaultGateway[0] + elif PLATFORM == 'Darwin': + self._gateway, self._interface = defaultGateway[0], None + else: + self._fail('', f'unrecognized platform: {PLATFORM}') + + return + + tun = Tun2socks( + exitCallback=self.runtimeExitCallback, + msgCallback=self.msgCallbackTUN_, + ) + + self._tun = tun + self.attempt.ownRuntime(tun) + + tcpSendBufferSize = userTcpSendBufferSize() + tcpReceiveBufferSize = userTcpReceiveBufferSize() + tcpAutoTuning = userTcpAutoTuning() == 'True' + interfaceArg = ( + APPLICATION_TUN2SOCKS_NETWORK_INTERFACE_NAME + if PLATFORM != 'Linux' + else self._interface + ) + + self._startTUN = functools.partial( + tun.start, + APPLICATION_TUN2SOCKS_DEVICE_NAME, + interfaceArg, + 'error', + f'socks5://{configcopy.socksProxy()}', + '', + f'{tcpSendBufferSize}MB', + f'{tcpReceiveBufferSize}MB', + tcpAutoTuning, + waitCore=False, + ) + + if PLATFORM != 'Linux': + self._setStage(ConnectionStartStage.StartingTUNRuntime) + + if not self._startTUN(): + self._fail(self._runtimeStartError(tun)) + + return + + if PLATFORM == 'Darwin': + self._observeRuntime( + tun, + CoreRuntimeStartup(), + '_prepareTunBypass', + ConnectionStartStage.WaitingTUNRuntime, + ) + else: + self._resume('_prepareTunBypass') + + def _prepareTunBypass(self): + """Prepare remote-server bypass routes without blocking for DNS.""" + if not self._isCurrent(): + return + + bypassTUN = userBypassTUNAdapterInterfaceIP() + + if bypassTUN: + for bypass in bypassTUN.split(','): + bypass = bypass.strip() + + if not isValidIPAddress(bypass): + SystemRoutingTable.managedRoutes.clear() + + self._fail( + '', + f'invalid IP address when processing user TUN ' + f'bypass settings: {bypass}', + ) + + return + + logger.info(f'processing user TUN bypass IP: {bypass}') + + SystemRoutingTable.managedRoutes.append([bypass, self._gateway]) + + self._resume('_continueTunPlatform') + + return + + address = self.attempt.runtimeConfiguration.remoteAddress() + + if isValidIPAddress(address): + SystemRoutingTable.managedRoutes.append([address, self._gateway]) + + self._resume('_continueTunPlatform') + + return + + self._setStage(ConnectionStartStage.ResolvingTUNAddress) + + resolver = self.manager._connectionDnsResolver() + resolver.configureHttpProxy(self.attempt.runtimeConfiguration.httpProxy()) + operation = resolver.resolveAsync(address, parent=self) + + self._dnsOperation = operation + + connectWeakly(operation.finished, self, '_tunAddressResolved') + + operation.start() + + def _tunAddressResolved(self, error, resolved): + """Resume TUN setup after event-driven remote-address resolution.""" + if not self._isCurrent(): + return + + operation = self._dnsOperation + + self._dnsOperation = None + + if operation is not None: + operation.deleteLater() + + if error: + SystemRoutingTable.managedRoutes.clear() + + self._fail( + '', + f'DNS resolution failed: ' + f'{self.attempt.runtimeConfiguration.remoteAddress()}', + ) + + return + + for address in resolved: + SystemRoutingTable.managedRoutes.append([address, self._gateway]) + + self._resume('_continueTunPlatform') + + def _continueTunPlatform(self): + """Continue with the platform-specific TUN stage.""" + if PLATFORM == 'Windows': + self._waitForTunDevice( + functools.partial( + SystemRoutingTable.WIN32IpconfigFindContent, + APPLICATION_TUN2SOCKS_DEVICE_NAME, + ), + '_applyWindowsTun', + ) + elif PLATFORM == 'Darwin': + self._applyDarwinTun() + elif PLATFORM == 'Linux': + self._prepareLinuxTun() + else: + self._fail('', f'unrecognized platform: {PLATFORM}') + + def _waitForTunDevice(self, predicate, continuation): + """Wait for one platform TUN device through a Qt timer.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.WaitingTUNDevice) + self._conditionContinuation = continuation + + probe = _ConditionProbe( + predicate, + f'TUN device {APPLICATION_TUN2SOCKS_DEVICE_NAME!r}', + parent=self, + ) + + self._conditionProbe = probe + + connectWeakly(probe.finished, self, '_tunDeviceObserved') + + probe.start() + + def _tunDeviceObserved(self, success): + """Resume after one device observation reaches terminal.""" + if not self._isCurrent() or self._conditionProbe is None: + return + + probe = self._conditionProbe + continuation = self._conditionContinuation + + self._conditionProbe = None + self._conditionContinuation = '' + + probe.deleteLater() + + if not success: + self._fail('', 'TUN device did not become ready') + + return + + self._resume(continuation) + + def _applyWindowsTun(self): + """Apply Windows DNS, gateway, and route mutations after readiness.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.ApplyingHostNetwork) + + userInterfaceName = userPrimaryAdapterInterfaceName() + alias = ( + userInterfaceName + if userInterfaceName + else SystemRoutingTable.WIN32GetInterfaceAliasByIP(self._interface) + ) + + if alias: + + def _windowsCleanup(_alias): + SystemRoutingTable.WIN32SetInterfaceDNS(_alias) + SystemRoutingTable.WIN32FlushDNSCache() + + self._tun.cleanup = functools.partial(_windowsCleanup, alias) + + if userDisablePrimaryAdapterInterfaceDNS() != 'False': + SystemRoutingTable.WIN32SetInterfaceDNS( + alias, + '127.0.0.1', + False, + ) + + interfaceDNS = ( + userTunAdapterInterfaceDNS() or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS + ) + + SystemRoutingTable.addRelations() + SystemRoutingTable.WIN32SetInterfaceDNS( + APPLICATION_TUN2SOCKS_DEVICE_NAME, + interfaceDNS, + False, + ) + SystemRoutingTable.setDeviceGateway( + APPLICATION_TUN2SOCKS_DEVICE_NAME, + APPLICATION_TUN2SOCKS_IP_ADDRESS, + APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS, + ) + SystemRoutingTable.WIN32FlushDNSCache() + + self._commit() + + def _applyDarwinTun(self): + """Apply macOS DNS, gateway, and route mutations after startup.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.ApplyingHostNetwork) + + for address in [ + *list(f'{2 ** (8 - index)}.0.0.0/{index}' for index in range(8, 0, -1)), + '198.18.0.0/15', + ]: + SystemRoutingTable.managedRoutes.append( + [address, APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS] + ) + + servers = SystemRoutingTable.DarwinGetDNSServers() + + def _darwinCleanup(_servers): + for service, dnsserver in _servers: + SystemRoutingTable.DarwinSetDNSServers(service, dnsserver) + + self._tun.cleanup = functools.partial(_darwinCleanup, servers) + + interfaceDNS = ( + userTunAdapterInterfaceDNS() or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS + ) + + for service, _dnsserver in servers: + SystemRoutingTable.DarwinSetDNSServers(service, interfaceDNS) + + SystemRoutingTable.setDeviceGateway( + APPLICATION_TUN2SOCKS_DEVICE_NAME, + APPLICATION_TUN2SOCKS_IP_ADDRESS, + APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS, + ) + SystemRoutingTable.addRelations() + + self._commit() + + def _prepareLinuxTun(self): + """Create Linux TUN and routes before starting tun2socks.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.ApplyingHostNetwork) + + def _linuxCleanup(): + SystemRoutingTable.LinuxDeleteTUNDevice(APPLICATION_TUN2SOCKS_DEVICE_NAME) + + self._tun.cleanup = _linuxCleanup + + commandBringUpTUN = '' + + if not SystemRoutingTable.LinuxFindTUNDevice(APPLICATION_TUN2SOCKS_DEVICE_NAME): + commandBringUpTUN = ( + f'ip tuntap add mode tun dev ' + f'{APPLICATION_TUN2SOCKS_DEVICE_NAME}\n' + f'ip addr add 10.10.10.10/24 dev ' + f'{APPLICATION_TUN2SOCKS_DEVICE_NAME}\n' + f'ip link set dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} up' + ) + + commandAddDefaultRoute = ( + f'ip route add default dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} ' 'metric 5' + ) + + def route(source, destination): + return f'{source} via {destination} dev {self._interface}' + + iproute = SystemRoutingTable.LinuxGetIpRoute() + commandBypass = '\n'.join( + f'ip route add {route(sourceIP, destinationIP)}' + for sourceIP, destinationIP in SystemRoutingTable.managedRoutes + if iproute.find(route(sourceIP, destinationIP)) == -1 + ) + tempdir = os.environ.get('TMPDIR') if SystemRuntime.flatpakID() else None + + with tempfile.NamedTemporaryFile( + mode='w', + encoding='utf-8', + suffix='.sh', + dir=tempdir, + delete=True, + ) as file: + file.write( + '\n'.join( + filter( + bool, + [ + commandBringUpTUN, + commandAddDefaultRoute, + commandBypass, + ], + ) + ) + ) + file.flush() + + if not SystemRoutingTable.LinuxExecutePrivilegedScript( + file.name, + shell='bash', + ): + self._fail('', 'failed to configure Linux TUN routes') + + return + + self._waitForTunDevice( + functools.partial( + SystemRoutingTable.LinuxFindTUNDevice, + APPLICATION_TUN2SOCKS_DEVICE_NAME, + ), + '_startLinuxTun', + ) + + def _startLinuxTun(self): + """Start Linux tun2socks after the host device exists.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.StartingTUNRuntime) + + if not self._startTUN(): + self._fail(self._runtimeStartError(self._tun)) + + return + + self._observeRuntime( + self._tun, + CoreRuntimeStartup(), + '_commit', + ConnectionStartStage.WaitingTUNRuntime, + ) + + def _commit(self): + """Transfer exact runtime ownership only after every stage succeeds.""" + if not self._isCurrent(): + return + + self._setStage(ConnectionStartStage.Committing) + + for runtime in self.attempt.runtimes: + setExitCallback = getattr(runtime, 'setExitCallback', None) + + if callable(setExitCallback): + setExitCallback(self.exitCallback) + + self.attempt.commit() + self._terminal = True + self._setStage(ConnectionStartStage.Succeeded) + self.succeeded.emit(self) + self.manager._finishStartOperation(self) + self.deleteLater() + + def _cancelObservers(self): + """Cancel every child observer owned by this operation.""" + if self._readinessProbe is not None: + self._readinessProbe.cancel() + self._readinessProbe = None + + if self._conditionProbe is not None: + self._conditionProbe.cancel() + self._conditionProbe = None + + if self._dnsOperation is not None: + self._dnsOperation.cancel() + self._dnsOperation = None + + self._conditionContinuation = '' + + def _fail(self, message='', details=''): + """Roll back and publish one terminal failure.""" + if self._terminal: + return + + self._terminal = True + self._cancelObservers() + + concise = str(message or '') + + if concise: + self.manager._lastStartError = concise + + self.attempt.rollback(f'connection startup failed: {details or concise}') + self._setStage(ConnectionStartStage.Failed) + self.failed.emit(self, concise, str(details or '')) + self.manager._finishStartOperation(self) + self.deleteLater() + + def cancel(self): + """Cancel this generation and roll back only its acquired resources.""" + if self._terminal: + return False + + self._terminal = True + self._cancelObservers() + self.attempt.rollback('connection startup cancelled') + self._setStage(ConnectionStartStage.Cancelled) + self.cancelled.emit(self) + self.manager._finishStartOperation(self) + self.deleteLater() + + return True + + class ConnectionManager(Mixins.CleanupOnExit): """Coordinate proxy cores, TUN setup, DNS changes, and routing cleanup.""" @@ -137,6 +1105,8 @@ class ConnectionManager(Mixins.CleanupOnExit): self.uniqueCleanup = False self.runtimes = list() self._lastStartError = '' + self._startGeneration = 0 + self._activeStartOperation = None def _connectionDnsResolver(self) -> DnsResolver: """Return the resolver owned by this connection-manager lifecycle.""" @@ -273,7 +1243,12 @@ class ConnectionManager(Mixins.CleanupOnExit): log=True, **kwargs, ) -> bool: - """Run one staged connection attempt and roll it back on any failure.""" + """Run the legacy synchronous startup compatibility path. + + Normal GUI connections use :meth:`startAsync`. This path remains for + third-party plugins and non-interactive callers that rely on historical + synchronous launch semantics. + """ self._lastStartError = '' attempt = _ConnectionStartAttempt( @@ -298,6 +1273,56 @@ class ConnectionManager(Mixins.CleanupOnExit): raise + def startAsync( + self, + config: CoreConfiguration | ServerProfile, + routing: str, + exitCallback=None, + msgCallbackCore=None, + msgCallbackTUN_=None, + deepcopy=True, + proxyModeOnly=False, + log=True, + **kwargs, + ): + """Create and schedule one manager-owned startup transaction.""" + if self._activeStartOperation is not None: + self._activeStartOperation.cancel() + + self._lastStartError = '' + self._startGeneration += 1 + operation = ConnectionStartOperation( + self, + self._startGeneration, + config, + routing, + exitCallback=exitCallback, + msgCallbackCore=msgCallbackCore, + msgCallbackTUN_=msgCallbackTUN_, + deepcopy=deepcopy, + proxyModeOnly=proxyModeOnly, + log=log, + options=kwargs, + ) + self._activeStartOperation = operation + singleShotWeakly(0, operation, 'start') + + return operation + + def _finishStartOperation(self, operation): + """Release manager ownership of one exact terminal generation.""" + if self._activeStartOperation is operation: + self._activeStartOperation = None + + def cancelStart(self, operation=None): + """Cancel the current startup generation when it matches *operation*.""" + current = self._activeStartOperation + + if current is None or (operation is not None and current is not operation): + return False + + return current.cancel() + def _startAttempt( self, attempt: _ConnectionStartAttempt, @@ -751,11 +1776,14 @@ class ConnectionManager(Mixins.CleanupOnExit): def stopAll(self): """Stop every managed proxy-core and TUN runtime.""" + self.cancelStart() + for runtime in reversed(list(self.runtimes)): self._releaseRuntime(runtime) def cleanup(self): """Release resources owned by the core manager.""" + self.cancelStart() self.stopAll() if self._dnsResolver is not None: diff --git a/Furious/Service/DnsResolver.py b/Furious/Service/DnsResolver.py index c6cc30b..ec5f796 100644 --- a/Furious/Service/DnsResolver.py +++ b/Furious/Service/DnsResolver.py @@ -22,6 +22,7 @@ from __future__ import annotations from Furious.Frozenlib import * from Furious.Models import * from Furious.Qt.HttpGetManager import * +from Furious.Qt.Signals import connectWeakly from PySide6 import QtCore from PySide6.QtNetwork import * @@ -30,11 +31,104 @@ from typing import Tuple import logging -__all__ = ['DnsResolver'] +__all__ = ['DnsResolutionOperation', 'DnsResolver'] logger = logging.getLogger(__name__) +class DnsResolutionOperation(QtCore.QObject): + """Observe one recursive DNS request without nesting the Qt event loop.""" + + finished = QtCore.Signal(bool, object) + + def __init__(self, resolver, domain, timeout=30000, parent=None): + """Initialize an idle resolution operation.""" + super().__init__(parent) + + self._resolver = resolver + self._domain = domain + self._timeout = max(int(timeout), 1) + self._resultMap = resolver._newResultMap(domain) + self._terminal = False + + self._elapsed = QtCore.QElapsedTimer() + self._timer = QtCore.QTimer(self) + self._timer.setInterval(20) + + connectWeakly(self._timer.timeout, self, '_poll') + + def start(self): + """Start the DNS request and its event-driven completion observer.""" + if self._terminal or self._timer.isActive(): + return + + try: + self._resolver._beginResolve(self._resultMap) + except Exception as ex: + # Any non-exit exceptions + + logger.error(f'failed to start DNS resolution for {self._domain!r}: {ex}') + + self._resultMap['error'] = True + self._finish() + + return + + self._elapsed.start() + self._timer.start() + self._poll() + + def _poll(self): + """Finish when recursion drains, or abort this request at its deadline.""" + if self._terminal: + return + + if self._resultMap['depth'] == 0: + self._finish() + + return + + if self._elapsed.isValid() and self._elapsed.elapsed() >= self._timeout: + logger.error( + f'DNS resolution for {self._domain!r} reached timeout ' + f'{self._timeout // 1000}s' + ) + + self._resultMap['error'] = True + self._abortReplies() + self._finish() + + def _abortReplies(self): + """Abort only network replies acquired by this resolution.""" + for networkReply in self._resultMap['reference']: + if ( + isinstance(networkReply, QNetworkReply) + and not networkReply.isFinished() + ): + networkReply.abort() + + def _finish(self): + """Publish exactly one terminal result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self.finished.emit( + bool(self._resultMap['error']), + list(self._resultMap['result'].keys()), + ) + + def cancel(self): + """Cancel without publishing a stale result.""" + if self._terminal: + return + + self._terminal = True + self._timer.stop() + self._abortReplies() + + class DnsResolver(HttpGetManager): """Represent DNS resolver.""" @@ -199,17 +293,25 @@ class DnsResolver(HttpGetManager): resultMap['error'] = True resultMap['depth'] -= 1 - def resolve(self, domain, timeout=30000) -> Tuple[bool, list[str]]: - """Resolve the DNS resolver.""" - resultMap = { + @staticmethod + def _newResultMap(domain): + """Return mutable state for one recursive DNS resolution.""" + normalizedDomain = str(domain).rstrip('.').strip().casefold() + + return { 'domain': domain, 'depth': 0, 'error': False, 'reference': [], 'result': {}, - 'visited': {str(domain).rstrip('.').strip().casefold()}, + 'visited': {normalizedDomain}, } + def _beginResolve(self, resultMap): + """Start the root request for one prepared resolution state.""" + domain = resultMap['domain'] + normalizedDomain = str(domain).rstrip('.').strip().casefold() + resultMap['depth'] += 1 networkReply = self.webGET( @@ -218,15 +320,29 @@ class DnsResolver(HttpGetManager): domain=domain, resultMap=resultMap, referenceDepth=0, - ancestry=(str(domain).rstrip('.').strip().casefold(),), + ancestry=(normalizedDomain,), ) resultMap['reference'].append(networkReply) + def resolve(self, domain, timeout=30000) -> Tuple[bool, list[str]]: + """Resolve the DNS resolver.""" + resultMap = self._newResultMap(domain) + + self._beginResolve(resultMap) self.wait(resultMap, timeout=timeout) return resultMap['error'], list(resultMap['result'].keys()) + def resolveAsync(self, domain, timeout=30000, parent=None): + """Return an event-driven DNS operation; the caller starts and owns it.""" + return DnsResolutionOperation( + self, + domain, + timeout=timeout, + parent=parent, + ) + @staticmethod def wait(resultMap, startCounter=0, timeout=30000, step=100): """Wait for the DNS resolver operation to complete.""" diff --git a/Furious/Service/__init__.py b/Furious/Service/__init__.py index 75ea5a2..2bdb8b0 100644 --- a/Furious/Service/__init__.py +++ b/Furious/Service/__init__.py @@ -19,7 +19,11 @@ from __future__ import annotations -from .ConnectionManager import ConnectionManager +from .ConnectionManager import ( + ConnectionManager, + ConnectionStartOperation, + ConnectionStartStage, +) from .ConnectivityManager import ConnectivityManager from .DnsResolver import DnsResolver from .EndpointInfoService import ( @@ -84,6 +88,8 @@ from .UpdateManager import UpdateManager __all__ = [ 'ConnectionManager', + 'ConnectionStartOperation', + 'ConnectionStartStage', 'ConnectivityManager', 'DnsResolver', 'EndpointInfo', diff --git a/tests/README.md b/tests/README.md index 34a932c..25693db 100644 --- a/tests/README.md +++ b/tests/README.md @@ -40,7 +40,7 @@ strategy in an individual test. | Configuration, profiles, migration, repositories | `test_models_and_services.py`, `test_repository_contracts.py` | | Generation log invariants, model fuzzing, concurrency, reclamation, complexity, and opt-in soak/latency probes | `test_log_manager_generation.py` | | Low-level application, runtime, editor, and storage contracts | `test_interface.py` | -| Application composition, startup rollback, connection ownership, entry-point and crash boundaries | `test_architecture_refactors.py`, `test_application_process.py` | +| Application composition, asynchronous readiness/TUN startup, rollback, connection ownership, entry-point and crash boundaries | `test_architecture_refactors.py`, `test_connection_startup_async.py`, `test_application_process.py` | | Plugin registration, capability dispatch, factories, rollback, and Hysteria1 ownership | `test_plugin_architecture.py`, `test_hysteria1_protocol.py` | | Controller state and error transitions with injected runtimes | `test_controllers.py` | | SOCKS and SIP002 Shadowsocks codecs and generated round trips | `test_socks_uri.py`, `test_shadowsocks_uri.py` | @@ -109,7 +109,7 @@ Then run the desired test tier. python -m unittest discover -s tests -v # Regular logic, persistence, plugin, controller, codec, and UI regressions -python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_architecture_refactors tests.test_plugin_architecture tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_controllers tests.test_subscription_manager tests.test_subscription_sync tests.test_profile_test_jobs tests.test_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract tests.test_xray_asset_download tests.test_native_tun_semantics tests.test_metrics_behavior tests.test_endpoint_info tests.test_service_runtime tests.test_frozenlib tests.test_isolation_and_navigation tests.test_main_window_geometry tests.test_dialog_geometry tests.test_ui_behavior tests.test_qt_interactions tests.test_stylesheet_states tests.test_theme_transition tests.test_public_api -v +python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_architecture_refactors tests.test_connection_startup_async tests.test_plugin_architecture tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_controllers tests.test_subscription_manager tests.test_subscription_sync tests.test_profile_test_jobs tests.test_socks_uri tests.test_shadowsocks_uri tests.test_backend_editor_contract tests.test_xray_asset_download tests.test_native_tun_semantics tests.test_metrics_behavior tests.test_endpoint_info tests.test_service_runtime tests.test_frozenlib tests.test_isolation_and_navigation tests.test_main_window_geometry tests.test_dialog_geometry tests.test_ui_behavior tests.test_qt_interactions tests.test_stylesheet_states tests.test_theme_transition tests.test_public_api -v # Direct Qt/process integration and destruction/lifetime checks python -m unittest tests.test_application_process tests.test_external_core tests.test_layout_matrix tests.test_qt_lifetime -v @@ -126,7 +126,7 @@ python -m unittest tests.test_very_heavy -v python -m unittest tests.test_log_manager_generation.VeryHeavyGenerationLogManagerTest -v # Shared-state order-independence spot check -python -m unittest tests.test_public_api tests.test_theme_transition tests.test_stylesheet_states tests.test_qt_interactions tests.test_ui_behavior tests.test_dialog_geometry tests.test_main_window_geometry tests.test_isolation_and_navigation tests.test_frozenlib tests.test_service_runtime tests.test_endpoint_info tests.test_metrics_behavior tests.test_native_tun_semantics tests.test_xray_asset_download tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri tests.test_profile_test_jobs tests.test_subscription_sync tests.test_subscription_manager tests.test_controllers tests.test_hysteria2_compatibility tests.test_hysteria1_protocol tests.test_plugin_architecture tests.test_architecture_refactors tests.test_repository_contracts tests.test_models_and_services tests.test_interface -v +python -m unittest tests.test_public_api tests.test_theme_transition tests.test_stylesheet_states tests.test_qt_interactions tests.test_ui_behavior tests.test_dialog_geometry tests.test_main_window_geometry tests.test_isolation_and_navigation tests.test_frozenlib tests.test_service_runtime tests.test_endpoint_info tests.test_metrics_behavior tests.test_native_tun_semantics tests.test_xray_asset_download tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri tests.test_profile_test_jobs tests.test_subscription_sync tests.test_subscription_manager tests.test_controllers tests.test_hysteria2_compatibility tests.test_hysteria1_protocol tests.test_plugin_architecture tests.test_connection_startup_async tests.test_architecture_refactors tests.test_repository_contracts tests.test_models_and_services tests.test_interface -v python -m unittest discover -s tests -v ``` diff --git a/tests/test_connection_startup_async.py b/tests/test_connection_startup_async.py new file mode 100644 index 0000000..40b5480 --- /dev/null +++ b/tests/test_connection_startup_async.py @@ -0,0 +1,786 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# This file is part of Furious. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""Exercise event-driven connection startup with real Qt event delivery.""" + +from __future__ import annotations + +from Furious.Interface import CoreRuntime +from Furious.Plugins import CoreRuntimeLaunch, CoreRuntimeStartup +from Furious.Qt.Signals import singleShotWeakly +from Furious.Service.ConnectionManager import ( + ConnectionManager, + ConnectionStartStage, +) +from Furious.Service.DnsResolver import DnsResolutionOperation + +from PySide6 import QtCore, QtNetwork + +from tests.support import application, processQtEvents, waitFor + +import importlib +import unittest + +from unittest import TestCase, mock + + +class _Configuration: + """Provide the endpoint values used by manager-only tests.""" + + def httpProxy(self): + """Return a deterministic local proxy endpoint.""" + return '127.0.0.1:18080' + + def socksProxy(self): + """Return a deterministic local SOCKS endpoint.""" + return '127.0.0.1:18081' + + def remoteAddress(self): + """Avoid DNS during mocked TUN tests.""" + return '192.0.2.1' + + +class _Runtime(CoreRuntime): + """Provide an observable in-process stand-in for a core runtime.""" + + def __init__(self, exitCallback=None): + """Initialize an idle runtime and lifecycle counters.""" + super().__init__(exitCallback) + + self.alive = False + self.startOptions = [] + self.stopCount = 0 + self.disposeCount = 0 + self.confirmCount = 0 + + @staticmethod + def name(): + """Return the fixture runtime name.""" + return 'Async Fixture' + + @staticmethod + def version(): + """Return a fixture version.""" + return '1.0' + + def start(self, _configuration=None, *_args, **kwargs): + """Become live and retain launch options.""" + self.startOptions.append(dict(kwargs)) + self.alive = True + + return True + + def stop(self): + """Stop this exact runtime.""" + self.stopCount += 1 + self.alive = False + + def dispose(self): + """Record final resource disposal.""" + self.disposeCount += 1 + self._exitCallback = None + + def isAlive(self): + """Return the controlled process-survival state.""" + return self.alive + + def confirmStartup(self): + """Record readiness confirmation.""" + if not self.alive: + return False + + self.confirmCount += 1 + + return True + + def fail(self, exitcode=61): + """Simulate an early child-process exit.""" + self.alive = False + self.callExitCallback(exitcode) + + +class _Registry: + """Return prepared launches in deterministic order.""" + + def __init__(self, launches): + """Retain a launch queue.""" + self.launches = list(launches) + + def createCoreRuntime(self, *_args, **_kwargs): + """Return the next prepared runtime launch.""" + return self.launches.pop(0) + + +class _ResolverFixture(QtCore.QObject): + """Complete one recursive-resolution state through a real Qt timer.""" + + def __init__(self): + """Initialize without a pending result.""" + super().__init__() + + self.resultMap = None + + @staticmethod + def _newResultMap(domain): + """Return the minimum state required by the observer.""" + return { + 'domain': domain, + 'depth': 0, + 'error': False, + 'reference': [], + 'result': {}, + } + + def _beginResolve(self, resultMap): + """Schedule asynchronous completion.""" + self.resultMap = resultMap + resultMap['depth'] = 1 + singleShotWeakly(0, self, '_complete') + + def _complete(self): + """Publish one deterministic address through shared state.""" + self.resultMap['result']['192.0.2.10'] = True + self.resultMap['depth'] = 0 + + +class ConnectionStartupAsyncTest(TestCase): + """Verify readiness, cancellation, rollback, and compatibility.""" + + def setUp(self): + """Ensure a Qt application exists for real timer/socket delivery.""" + self.app = application() + self.servers = [] + self.managers = [] + + def tearDown(self): + """Release every exact manager and local listener.""" + for manager in self.managers: + manager.cleanup() + + for server in self.servers: + server.close() + server.deleteLater() + + processQtEvents() + + def _manager(self): + """Return a manager with host TUN disabled at the test boundary.""" + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, False)) + self.managers.append(manager) + + return manager + + def _server(self): + """Listen on one real loopback endpoint.""" + server = QtNetwork.QTcpServer(self.app) + self.assertTrue( + server.listen( + QtNetwork.QHostAddress.SpecialAddress.LocalHost, + 0, + ) + ) + self.servers.append(server) + + return server + + def _operation(self, manager, launch): + """Start one operation through an injected registry.""" + registry = _Registry([launch]) + patcher = mock.patch( + 'Furious.Service.ConnectionManager.getPluginRegistry', + return_value=registry, + ) + patcher.start() + self.addCleanup(patcher.stop) + + return manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + + def testLocalEndpointReadinessCompletesEarlyAndKeepsQtResponsive(self): + """Commit as soon as a real listener accepts while timers still run.""" + server = self._server() + endpoint = f'127.0.0.1:{server.serverPort()}' + runtime = _Runtime() + launch = CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup(endpoint=endpoint, timeout=1500), + ) + manager = self._manager() + operation = self._operation(manager, launch) + succeeded = [] + timerEvents = [] + operation.succeeded.connect(succeeded.append) + QtCore.QTimer.singleShot(0, lambda: timerEvents.append(True)) + + self.assertEqual(operation.stage, ConnectionStartStage.Pending) + self.assertTrue(waitFor(lambda: bool(succeeded), timeout=0.5)) + self.assertEqual(timerEvents, [True]) + self.assertEqual(manager.runtimes, [runtime]) + self.assertEqual(runtime.confirmCount, 1) + self.assertFalse(runtime.startOptions[0]['waitCore']) + + def testReadinessTimeoutRollsBackTheExactRuntime(self): + """Fail a live process whose promised local endpoint never appears.""" + server = self._server() + port = server.serverPort() + server.close() + runtime = _Runtime() + manager = self._manager() + operation = self._operation( + manager, + CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{port}', + timeout=60, + retryInterval=5, + ), + ), + ) + failures = [] + operation.failed.connect(lambda *_args: failures.append(_args)) + + self.assertTrue(waitFor(lambda: bool(failures))) + self.assertEqual(manager.runtimes, []) + self.assertEqual(runtime.stopCount, 1) + self.assertEqual(runtime.disposeCount, 1) + + def testEarlyRuntimeExitFailsOnceAndIgnoresLateProbeEvents(self): + """Let process termination own the terminal result before timeout.""" + runtime = _Runtime() + manager = self._manager() + operation = self._operation( + manager, + CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint='127.0.0.1:1', + timeout=500, + ), + ), + ) + failures = [] + operation.failed.connect(lambda *_args: failures.append(_args)) + + processQtEvents() + singleShotWeakly(0, runtime, 'fail') + + self.assertTrue(waitFor(lambda: bool(failures))) + processQtEvents(5) + self.assertEqual(len(failures), 1) + self.assertEqual(runtime.stopCount, 1) + self.assertEqual(runtime.disposeCount, 1) + + def testCancellationAndReplacementCannotCommitAStaleGeneration(self): + """Cancel the first generation before a second ready launch commits.""" + server = self._server() + first = _Runtime() + second = _Runtime() + manager = self._manager() + registry = _Registry( + [ + CoreRuntimeLaunch( + first, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint='127.0.0.1:1', + timeout=5000, + ), + ), + CoreRuntimeLaunch( + second, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ), + ] + ) + + with mock.patch( + 'Furious.Service.ConnectionManager.getPluginRegistry', + return_value=registry, + ): + firstOperation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + cancelled = [] + firstOperation.cancelled.connect(cancelled.append) + processQtEvents() + + secondOperation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + secondOperation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual(cancelled, [firstOperation]) + self.assertEqual(first.stopCount, 1) + self.assertEqual(first.disposeCount, 1) + self.assertEqual(manager.runtimes, [second]) + + def testLegacyLaunchRetainsSynchronousStartOptions(self): + """Do not inject waitCore into a plugin without async capability.""" + runtime = _Runtime() + manager = self._manager() + operation = self._operation( + manager, + CoreRuntimeLaunch(runtime, _Configuration()), + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + self.assertEqual(runtime.startOptions, [{}]) + self.assertEqual(runtime.confirmCount, 0) + self.assertEqual(manager.runtimes, [runtime]) + + def testDnsResolutionOperationCompletesAndCancelsWithoutNestedWait(self): + """Observe recursive DNS state through timers and suppress stale cancel.""" + resolver = _ResolverFixture() + operation = DnsResolutionOperation( + resolver, + 'example.test', + timeout=200, + ) + results = [] + operation.finished.connect( + lambda error, addresses: results.append((error, addresses)) + ) + operation.start() + + self.assertTrue(waitFor(lambda: bool(results))) + self.assertEqual(results, [(False, ['192.0.2.10'])]) + + cancelled = DnsResolutionOperation( + resolver, + 'cancelled.test', + timeout=200, + ) + staleResults = [] + cancelled.finished.connect( + lambda error, addresses: staleResults.append((error, addresses)) + ) + cancelled.start() + cancelled.cancel() + processQtEvents(5) + + self.assertEqual(staleResults, []) + operation.deleteLater() + cancelled.deleteLater() + resolver.deleteLater() + + def testLinuxTunPreservesDeviceBeforeRuntimeOrderingWithoutNestedWait(self): + """Create and observe the Linux device before launching tun2socks.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + server = self._server() + primary = _Runtime() + tun = _Runtime() + tun.cleanup = None + events = [] + deviceChecks = 0 + + def tunFactory(**kwargs): + tun.setExitCallback(kwargs.get('exitCallback')) + + return tun + + originalTunStart = tun.start + + def startTun(*args, **kwargs): + events.append('tun-start') + + return originalTunStart(*args, **kwargs) + + tun.start = startTun + + def findDevice(_name): + nonlocal deviceChecks + + deviceChecks += 1 + events.append('find-device') + + return deviceChecks >= 2 + + def executeScript(*_args, **_kwargs): + events.append('script') + + return True + + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) + self.managers.append(manager) + registry = _Registry( + [ + CoreRuntimeLaunch( + primary, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ) + ] + ) + shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5) + + with ( + mock.patch.object(module, 'PLATFORM', 'Linux'), + mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), + mock.patch.object( + module, + 'CoreRuntimeStartup', + return_value=shortSurvival, + ), + mock.patch.object( + module, + 'getPluginRegistry', + return_value=registry, + ), + mock.patch.object( + module.SystemRoutingTable, + 'managedRoutes', + [], + ), + mock.patch.object( + module.SystemRoutingTable, + 'getDefaultGateway', + return_value=[('192.0.2.254', 'eth0')], + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxFindTUNDevice', + side_effect=findDevice, + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxGetIpRoute', + return_value='', + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxExecutePrivilegedScript', + side_effect=executeScript, + ), + mock.patch.object( + module.SystemRoutingTable, + 'LinuxDeleteTUNDevice', + ), + mock.patch.object( + module.SystemRoutingTable, + 'deleteRelations', + ), + mock.patch.object( + module.SystemRuntime, + 'flatpakID', + return_value='', + ), + mock.patch.object( + module.PySide6Legacy, + 'eventLoopWait', + side_effect=AssertionError('nested event loop used'), + ), + ): + operation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual( + events[:4], + ['find-device', 'script', 'find-device', 'tun-start'], + ) + self.assertEqual(manager.runtimes, [primary, tun]) + self.assertFalse(tun.startOptions[0]['waitCore']) + + def testWindowsTunStartsRuntimeBeforeObservingAndMutatingDevice(self): + """Keep the Windows launch, device, then host-mutation sequence.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + server = self._server() + primary = _Runtime() + tun = _Runtime() + tun.cleanup = None + events = [] + + def tunFactory(**kwargs): + tun.setExitCallback(kwargs.get('exitCallback')) + + return tun + + originalTunStart = tun.start + + def startTun(*args, **kwargs): + events.append('tun-start') + + return originalTunStart(*args, **kwargs) + + tun.start = startTun + + def findDevice(_name): + events.append('find-device') + + return True + + def addRelations(): + events.append('add-relations') + + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) + self.managers.append(manager) + registry = _Registry( + [ + CoreRuntimeLaunch( + primary, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ) + ] + ) + + with ( + mock.patch.object(module, 'PLATFORM', 'Windows'), + mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), + mock.patch.object( + module, + 'getPluginRegistry', + return_value=registry, + ), + mock.patch.object( + module.SystemRoutingTable, + 'managedRoutes', + [], + ), + mock.patch.object( + module.SystemRoutingTable, + 'delete', + ), + mock.patch.object( + module.SystemRoutingTable, + 'getDefaultGateway', + return_value=[('192.0.2.254', '192.0.2.10')], + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32IpconfigFindContent', + side_effect=findDevice, + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32GetInterfaceAliasByIP', + return_value='Ethernet', + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32SetInterfaceDNS', + ), + mock.patch.object( + module.SystemRoutingTable, + 'WIN32FlushDNSCache', + ), + mock.patch.object( + module.SystemRoutingTable, + 'setDeviceGateway', + ), + mock.patch.object( + module.SystemRoutingTable, + 'addRelations', + side_effect=addRelations, + ), + mock.patch.object( + module.SystemRoutingTable, + 'deleteRelations', + ), + mock.patch.object( + module.PySide6Legacy, + 'eventLoopWait', + side_effect=AssertionError('nested event loop used'), + ), + ): + operation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual( + events[:3], + ['tun-start', 'find-device', 'add-relations'], + ) + self.assertEqual(manager.runtimes, [primary, tun]) + + def testDarwinTunSurvivalPrecedesDnsAndRouteMutation(self): + """Observe tun2socks survival before applying macOS host networking.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + server = self._server() + primary = _Runtime() + tun = _Runtime() + tun.cleanup = None + events = [] + + def tunFactory(**kwargs): + tun.setExitCallback(kwargs.get('exitCallback')) + + return tun + + originalTunStart = tun.start + + def startTun(*args, **kwargs): + events.append('tun-start') + + return originalTunStart(*args, **kwargs) + + tun.start = startTun + + def dnsServers(): + events.append('read-dns') + + return [('Wi-Fi', ['192.0.2.53'])] + + manager = ConnectionManager() + manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) + self.managers.append(manager) + registry = _Registry( + [ + CoreRuntimeLaunch( + primary, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint=f'127.0.0.1:{server.serverPort()}', + timeout=500, + ), + ) + ] + ) + shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5) + + with ( + mock.patch.object(module, 'PLATFORM', 'Darwin'), + mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), + mock.patch.object( + module, + 'CoreRuntimeStartup', + return_value=shortSurvival, + ), + mock.patch.object( + module, + 'getPluginRegistry', + return_value=registry, + ), + mock.patch.object( + module.SystemRoutingTable, + 'managedRoutes', + [], + ), + mock.patch.object( + module.SystemRoutingTable, + 'getDefaultGateway', + return_value=['192.0.2.254'], + ), + mock.patch.object( + module.SystemRoutingTable, + 'DarwinGetDNSServers', + side_effect=dnsServers, + ), + mock.patch.object( + module.SystemRoutingTable, + 'DarwinSetDNSServers', + ), + mock.patch.object( + module.SystemRoutingTable, + 'setDeviceGateway', + ), + mock.patch.object( + module.SystemRoutingTable, + 'addRelations', + ), + mock.patch.object( + module.SystemRoutingTable, + 'deleteRelations', + ), + mock.patch.object( + module.PySide6Legacy, + 'eventLoopWait', + side_effect=AssertionError('nested event loop used'), + ), + ): + operation = manager.startAsync( + _Configuration(), + 'Global', + deepcopy=False, + ) + succeeded = [] + operation.succeeded.connect(succeeded.append) + + self.assertTrue(waitFor(lambda: bool(succeeded))) + + self.assertEqual(events, ['tun-start', 'read-dns']) + self.assertEqual(manager.runtimes, [primary, tun]) + + def testRepeatedCancellationReleasesOperationTimersAndRuntimes(self): + """Keep repeated startup cancellation bounded and independently owned.""" + manager = self._manager() + + for _index in range(12): + runtime = _Runtime() + operation = self._operation( + manager, + CoreRuntimeLaunch( + runtime, + _Configuration(), + startup=CoreRuntimeStartup( + endpoint='127.0.0.1:1', + timeout=5000, + ), + ), + ) + processQtEvents() + self.assertTrue(manager.cancelStart(operation)) + processQtEvents() + self.assertEqual(runtime.stopCount, 1) + self.assertEqual(runtime.disposeCount, 1) + self.assertIsNone(manager._activeStartOperation) + + +if __name__ == '__main__': + application() + unittest.main() diff --git a/tests/test_controllers.py b/tests/test_controllers.py index 5e2d0f6..73745ac 100644 --- a/tests/test_controllers.py +++ b/tests/test_controllers.py @@ -29,6 +29,8 @@ from Furious.Frozenlib import AppBinarySettings, AppSettings from Furious.Models import CoreConfiguration, ServerProfile from Furious.Service.LogManager import LogManager +from PySide6 import QtCore + from tests.support import application, isolatedSettings, processQtEvents import unittest @@ -87,6 +89,53 @@ class FixtureCoreManager: raise self.stopException +class FixtureStartOperation(QtCore.QObject): + """Publish controllable asynchronous startup outcomes.""" + + succeeded = QtCore.Signal(object) + failed = QtCore.Signal(object, str, str) + cancelled = QtCore.Signal(object) + + def succeed(self): + """Publish one successful manager commit.""" + self.succeeded.emit(self) + + def fail(self, message='fixture failure', details=''): + """Publish one failed manager transaction.""" + self.failed.emit(self, message, details) + + +class FixtureAsyncCoreManager: + """Record controller use of the asynchronous manager boundary.""" + + def __init__(self): + """Initialize empty operation and runtime history.""" + self.lastStartError = '' + self.runtimes = [] + self.operations = [] + self.cancelCalls = [] + self.stopCalls = 0 + + def startAsync(self, configuration, **kwargs): + """Return one idle operation for the controller to observe.""" + operation = FixtureStartOperation() + self.operations.append((operation, configuration, kwargs)) + + return operation + + def cancelStart(self, operation): + """Cancel one exact operation.""" + self.cancelCalls.append(operation) + operation.cancelled.emit(operation) + + return True + + def stopAll(self): + """Record stable-runtime cleanup.""" + self.stopCalls += 1 + self.runtimes.clear() + + class FixtureUpdatesManager: """Record update hooks without contacting any update service.""" @@ -313,6 +362,103 @@ class ConnectionControllerTest(unittest.TestCase): controller.deleteLater() + def testAsyncManagerCommitsSystemProxyOnlyAfterSuccess(self): + """Remain Connecting until the manager transaction commits.""" + with isolatedSettings(): + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + mock.patch.object(controller, '_runPostConnectTasksOnce'), + ): + self.assertTrue(controller.startConnection(self.profile)) + self.assertTrue(controller.isConnecting()) + proxySet.assert_not_called() + + operation = core.operations[0][0] + core.runtimes.append(object()) + operation.succeed() + + self.assertTrue(controller.isConnected()) + proxySet.assert_called_once() + self.assertTrue(controller.startDisconnection()) + + controller.deleteLater() + + def testDisconnectCancelsAnInFlightStartupGeneration(self): + """Cancel partial startup without ever enabling the system proxy.""" + with isolatedSettings(): + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + ): + self.assertTrue(controller.startConnection(self.profile)) + operation = core.operations[0][0] + self.assertTrue(controller.startDisconnection()) + + self.assertEqual(core.cancelCalls, [operation]) + self.assertTrue(controller.state is ConnectionState.Disconnected) + proxySet.assert_not_called() + + controller.deleteLater() + + def testReconnectCancelsConnectingGenerationBeforeReplacement(self): + """Replace a Connecting attempt without accepting stale completion.""" + with isolatedSettings(): + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, + updatesManager=FixtureUpdatesManager(), + ) + + with ( + mock.patch( + 'Furious.Controllers.ConnectionController.Storage.UserServers', + return_value=[self.profile], + ), + mock.patch( + 'Furious.Controllers.ConnectionController.Storage.UserActivatedItemIndex', + return_value=0, + ), + mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, + mock.patch('Furious.Controllers.ConnectionController.SystemProxy.off'), + ): + self.assertTrue(controller.startConnection(self.profile)) + first = core.operations[0][0] + self.assertTrue(controller.startReconnection()) + second = core.operations[1][0] + + first.succeed() + self.assertTrue(controller.isConnecting()) + proxySet.assert_not_called() + + core.runtimes.append(object()) + second.succeed() + self.assertTrue(controller.isConnected()) + proxySet.assert_called_once() + controller.startDisconnection() + + self.assertEqual(core.cancelCalls, [first]) + + controller.deleteLater() + def testStartAndProxyExceptionsReturnToStableDisconnectedState(self): """Clean every partially acquired resource after injected failures.""" for core, proxySideEffect in (