From 41da0417fe44c84acc9d3d45b1ba7d099ea73bf1 Mon Sep 17 00:00:00 2001 From: Loren Eteval Date: Sat, 19 Sep 2026 20:48:53 +0800 Subject: [PATCH] Preserve runtime ownership through cancellation and cleanup Signed-off-by: Loren Eteval --- Furious/Backends/ExternalCore/Process.py | 24 ++-- Furious/Controllers/ConnectionController.py | 128 ++++++++++++++++++-- Furious/Core/MultiprocessingRuntime.py | 35 ++++-- Furious/Service/AGENTS.md | 6 +- Furious/Service/ConnectionManager.py | 71 +++++++++-- Furious/Service/RuntimeLease.py | 49 ++++++-- tests/README.md | 6 +- tests/test_connection_startup_async.py | 37 ++++++ tests/test_controllers.py | 105 ++++++++++++++++ tests/test_external_core.py | 48 ++++++++ tests/test_runtime_lifecycle.py | 99 ++++++++++++++- 11 files changed, 552 insertions(+), 56 deletions(-) diff --git a/Furious/Backends/ExternalCore/Process.py b/Furious/Backends/ExternalCore/Process.py index 97e5a74..7b55fcd 100644 --- a/Furious/Backends/ExternalCore/Process.py +++ b/Furious/Backends/ExternalCore/Process.py @@ -204,8 +204,6 @@ class ExternalCoreProcess(CoreRuntime): with self._lock: readers = tuple(self._readerThreads) - self._readerThreads.clear() - current = threading.current_thread() for thread in readers: @@ -228,6 +226,11 @@ class ExternalCoreProcess(CoreRuntime): if thread is not current: thread.join(self.ForcedShutdownTimeout) + with self._lock: + self._readerThreads = [ + thread for thread in self._readerThreads if thread.is_alive() + ] + def _watch(self, process: subprocess.Popen): """Reap the process and report an unexpected exit exactly once.""" exitCode = process.wait() @@ -509,21 +512,28 @@ class ExternalCoreProcess(CoreRuntime): exitCode = process.poll() if exitCode is None: - logger.error('external core process could not be reaped') - else: - logger.info(f'external core process stopped with code {exitCode}') + # Readers and the watcher still need this exact child and its pipes. + raise RuntimeError('External core process could not be reaped') + + logger.info(f'external core process stopped with code {exitCode}') self._joinReaders(process) with self._lock: watcher = self._watcherThread - self._watcherThread = None - if watcher is not None and watcher is not threading.current_thread(): watcher.join(self.ForcedShutdownTimeout) with self._lock: + if watcher is not None and not watcher.is_alive(): + self._watcherThread = None + + if self._readerThreads or self._watcherThread is not None: + self.setState(RuntimeState.Stopping) + + raise RuntimeError('External core reader or watcher did not stop') + self._lastExitCode = exitCode self.setState(RuntimeState.Exited) self._process = None diff --git a/Furious/Controllers/ConnectionController.py b/Furious/Controllers/ConnectionController.py index 2319e23..d6df655 100644 --- a/Furious/Controllers/ConnectionController.py +++ b/Furious/Controllers/ConnectionController.py @@ -107,6 +107,10 @@ class ConnectionController(QtCore.QObject): self._lastError = None self._startOperation = None self._pendingHttpProxy = '' + self._connectionGeneration = 0 + self._startAdmissionPending = False + self._shuttingDown = False + self._runtimeCleanupFailed = False self._actionTimer = QtCore.QTimer(self) self._actionTimer.timeout.connect(self._callActionFromQueue) @@ -158,9 +162,14 @@ class ConnectionController(QtCore.QObject): interactionWasEnabled = self.interactionEnabled + generation = self._connectionGeneration + self._state = state self.stateChanged.emit(state) + if generation != self._connectionGeneration or self._state is not state: + return + if self.interactionEnabled != interactionWasEnabled: self.interactionEnabledChanged.emit(self.interactionEnabled) @@ -202,26 +211,48 @@ class ConnectionController(QtCore.QObject): self._setState(ConnectionState.Disconnected) + def _isCurrentConnection(self, generation): + """Reject continuations from an earlier, synchronously replaced lifecycle.""" + return generation == self._connectionGeneration and not self._shuttingDown + def _startConnecting(self): """Enter the connecting state and request progress presentation.""" + generation = self._connectionGeneration + self._setState(ConnectionState.Connecting) - self.progressStarted.emit() + + if self._isCurrentConnection(generation) and self.isConnecting(): + self.progressStarted.emit() def _finishConnecting(self): - """Enter the connected state and notify connection-aware consumers.""" + """Publish connection completion only while this lifecycle stays current.""" + generation = self._connectionGeneration + self.progressFinished.emit(True) + if not self._isCurrentConnection(generation) or not self.isConnecting(): + return False + AppSettings.turnON_('Connect') self._setState(ConnectionState.Connected) + if not self._isCurrentConnection(generation) or not self.isConnected(): + return False + Mixins.ConnectionAware.callConnectedCallback() + return self._isCurrentConnection(generation) and self.isConnected() + def startConnection(self, configuration=None) -> bool: """Start *configuration* or the active repository profile.""" # QObject already exposes a legacy ``connect`` attribute in PySide. # Using an explicit operation name avoids shadowing Qt signal plumbing. - if self.state is not ConnectionState.Disconnected: + if ( + self.state is not ConnectionState.Disconnected + or self._startAdmissionPending + or self._shuttingDown + ): return False if configuration is None: @@ -275,17 +306,33 @@ class ConnectionController(QtCore.QObject): return False + self._connectionGeneration += 1 + + generation = self._connectionGeneration + + self._startAdmissionPending = True self._lastError = None - self._setActiveProfile(configuration) self._pendingHttpProxy = httpProxy + self._setActiveProfile(configuration) + + if not self._isCurrentConnection(generation): + return False self._startConnecting() + if not self._isCurrentConnection(generation) or not self.isConnecting(): + return False + + self._startAdmissionPending = False + logManager = AppLogManager() # Retain application diagnostics while starting a fresh runtime log. logManager.clear(runtimeOnly=True) + if not self._isCurrentConnection(generation) or not self.isConnecting(): + return False + startAsync = getattr(self._coreManager, 'startAsync', None) if callable(startAsync): @@ -310,6 +357,11 @@ class ConnectionController(QtCore.QObject): str(ex), ) + if not self._isCurrentConnection(generation) or not self.isConnecting(): + self._coreManager.cancelStart(operation) + + return False + self._startOperation = operation connectWeakly( @@ -362,6 +414,9 @@ class ConnectionController(QtCore.QObject): return False + if not self._isCurrentConnection(generation) or not self.isConnecting(): + return False + if not success: logger.error('failed to start core manager') @@ -377,6 +432,8 @@ class ConnectionController(QtCore.QObject): def _finishConnection(self, configuration, httpProxy) -> bool: """Commit system integration after manager runtime ownership commits.""" + generation = self._connectionGeneration + settings = AppSettings.get('CustomProxyBypass') proxyServerBypass = ( @@ -401,12 +458,19 @@ class ConnectionController(QtCore.QObject): f'{configuration.coreName()}: ' + _('Unknown error'), str(ex) ) - self._finishConnecting() + if not self._isCurrentConnection(generation) or not self.isConnecting(): + return False + + if not self._finishConnecting(): + return False self.notificationRequested.emit( f'{configuration.coreName()}: ' + _('Connected') ) + if not self._isCurrentConnection(generation) or not self.isConnected(): + return False + interval = CORE_CHECK_ALIVE_INTERVAL if AppSettings.isStateON_('PowerSaveMode'): @@ -424,6 +488,8 @@ class ConnectionController(QtCore.QObject): if operation is not self._startOperation or not self.isConnecting(): return + generation = self._connectionGeneration + self._startOperation = None self._emitRuntimesChanged() @@ -431,7 +497,7 @@ class ConnectionController(QtCore.QObject): while not self._actionQueue.empty(): self._callActionFromQueue() - if not self.isConnecting(): + if not self._isCurrentConnection(generation) or not self.isConnecting(): return configuration = self.activeProfile @@ -449,10 +515,15 @@ class ConnectionController(QtCore.QObject): if operation is not self._startOperation: return + generation = self._connectionGeneration + self._startOperation = None self._emitRuntimesChanged() + if not self._isCurrentConnection(generation): + return + configuration = self.activeProfile coreName = configuration.coreName() if configuration is not None else '' startError = message or getattr(self._coreManager, 'lastStartError', '') @@ -468,17 +539,31 @@ class ConnectionController(QtCore.QObject): if operation is not self._startOperation: return + generation = self._connectionGeneration + self._startOperation = None self._emitRuntimesChanged() + if not self._isCurrentConnection(generation): + return + self._reset() def startDisconnection(self, notification: str = '') -> bool: """Stop the active runtime and optionally request a notification.""" - if self.state is ConnectionState.Disconnected: + if self.isDisconnecting() or ( + self.state is ConnectionState.Disconnected + and not self._startAdmissionPending + ): return False + self._connectionGeneration += 1 + + generation = self._connectionGeneration + + self._startAdmissionPending = False + operation = self._startOperation self._startOperation = None @@ -504,6 +589,8 @@ class ConnectionController(QtCore.QObject): logger.error(f'failed to turn off system proxy: {ex}') + self._runtimeCleanupFailed = False + try: self._coreManager.stopAll() except Exception as ex: @@ -511,11 +598,16 @@ class ConnectionController(QtCore.QObject): # Always complete the state transition. A cleanup failure must not # strand every connection UI in the disabled Disconnecting state. + self._runtimeCleanupFailed = True + logger.error(f'failed to stop connection runtime: {ex}') self._emitRuntimesChanged() self._reset() + if generation != self._connectionGeneration: + return True + while not self._actionQueue.empty(): try: self._actionQueue.get_nowait() @@ -552,11 +644,25 @@ class ConnectionController(QtCore.QObject): """Stop runtime resources without changing the next-start preference.""" reconnectOnStartup = AppSettings.isStateON_('Connect') - if self.state is not ConnectionState.Disconnected: - self.startDisconnection() + self._shuttingDown = True - if reconnectOnStartup: - AppSettings.turnON_('Connect') + try: + if ( + self.state is not ConnectionState.Disconnected + or self._startAdmissionPending + ): + self.startDisconnection() + + if not self._runtimeCleanupFailed: + return + + # Disconnect keeps the UI usable on failure. Final shutdown must + # surface retained resources before its owner is destroyed. + self._coreManager.stopAll() + self._runtimeCleanupFailed = False + finally: + if reconnectOnStartup: + AppSettings.turnON_('Connect') def toggle(self) -> bool: """Perform the operation represented by the current stable state.""" diff --git a/Furious/Core/MultiprocessingRuntime.py b/Furious/Core/MultiprocessingRuntime.py index 2a0580e..cce25a7 100644 --- a/Furious/Core/MultiprocessingRuntime.py +++ b/Furious/Core/MultiprocessingRuntime.py @@ -134,7 +134,8 @@ class MultiprocessingRuntime(CoreRuntime): if self.isRunning(): raise RuntimeStartError('Runtime is already running') - self._closeProcess() + if not self._closeProcess(): + raise RuntimeStartError('Previous runtime handle could not be closed') self._stopRequested = False self._exitPublished = False @@ -212,22 +213,28 @@ class MultiprocessingRuntime(CoreRuntime): self.publishExit(event) def _closeProcess(self): - """Close and forget the exact inactive multiprocessing handle.""" + """Release a handle only after close succeeds; keep failures retryable.""" process = self._process - if process is not None: - try: - process.join(0) - except (AssertionError, OSError, ValueError): - pass + if process is None: + return True - try: - process.close() - except (OSError, ValueError): - pass + try: + process.join(0) + except (AssertionError, OSError, ValueError): + pass + + try: + process.close() + except (OSError, ValueError) as ex: + logger.error(f'{self.name()} process handle could not be closed: {ex}') + + return False self._process = None + return True + def stop(self): """Idempotently terminate execution without disposing this object.""" self._stopRequested = True @@ -258,8 +265,14 @@ class MultiprocessingRuntime(CoreRuntime): process.kill() process.join(self.StopJoinTimeout) + if process.is_alive(): + raise RuntimeError(f'{self.name()} process did not stop after escalation') + event = self._consumeExit() + if not self._closeProcess(): + raise RuntimeError(f'{self.name()} process handle could not be closed') + logger.info(f'{self.name()} stopped with exitcode {event.code}') def dispose(self): diff --git a/Furious/Service/AGENTS.md b/Furious/Service/AGENTS.md index 669126b..65cea72 100644 --- a/Furious/Service/AGENTS.md +++ b/Furious/Service/AGENTS.md @@ -37,8 +37,10 @@ for execution, and Qt for lifetime primitives. This scope owns multi-stage workf replacing the runtime callback. Worker-thread exits are queued to the router's Qt thread, delivered at most once, and suppressed after release. Execution liveness and endpoint/TUN readiness remain separate observations. A readiness timeout never replaces a typed exit after execution has already stopped, even when that exit is still queued. - Lease release currently logs stop/dispose errors and completes logical callback release; this is not evidence that - the underlying resource was reaped. Changes to cleanup-failure reporting must cover both runtime and lease owners. + Failed stop/dispose keeps the lease in Releasing with terminal delivery suppressed. The manager retains it for + retry and refuses new startup while release remains incomplete. A failed attempt transfers unreleased leases + back to that durable owner; deleting the attempt must not abandon them. Independent DNS cleanup still runs. + Verify runtime liveness and actual handle/thread release separately from the lease's logical state. - `HttpGetManager` owns reply/error/timeout cleanup. DNS recursion and external-input caches are bounded. Update, connectivity, endpoint, subscription, and asset requests own their exact reply and reject stale generations. - Subscription stages remain separate: decoders return neutral items; import constructs profiles/metadata; diff --git a/Furious/Service/ConnectionManager.py b/Furious/Service/ConnectionManager.py index 72957b9..8ad7840 100644 --- a/Furious/Service/ConnectionManager.py +++ b/Furious/Service/ConnectionManager.py @@ -99,7 +99,8 @@ class _ConnectionStartAttempt: return False for lease in reversed(self.leases): - lease.release() + if not lease.release(): + self.manager._pendingReleases.append(lease) self.leases.clear() @@ -432,6 +433,9 @@ class ConnectionStartOperation(QtCore.QObject): self._setStage(ConnectionStartStage.Preparing) + if not self._isCurrent(): + return + configcopy = self.attempt.runtimeConfiguration try: @@ -461,6 +465,9 @@ class ConnectionStartOperation(QtCore.QObject): self._setStage(ConnectionStartStage.StartingPrimary) + if not self._isCurrent(): + return + router = RuntimeEventRouter() try: @@ -491,6 +498,11 @@ class ConnectionStartOperation(QtCore.QObject): self.attempt.ownRuntime(runtime, router) + if not self._isCurrent(): + self.attempt.rollback() + + return + try: launch.start() except RuntimeStartError as ex: @@ -520,6 +532,10 @@ class ConnectionStartOperation(QtCore.QObject): return self._setStage(stage) + + if not self._isCurrent(): + return + self._conditionContinuation = continuation probe = _RuntimeReadinessProbe(runtime, startup, parent=self) @@ -594,6 +610,9 @@ class ConnectionStartOperation(QtCore.QObject): self._setStage(ConnectionStartStage.PreparingTUN) + if not self._isCurrent(): + return + configcopy = self.attempt.runtimeConfiguration if PLATFORM == 'Windows': @@ -676,6 +695,9 @@ class ConnectionStartOperation(QtCore.QObject): if PLATFORM != 'Linux': self._setStage(ConnectionStartStage.StartingTUNRuntime) + if not self._isCurrent(): + return + try: self._startTUN() except RuntimeStartError as ex: @@ -796,6 +818,10 @@ class ConnectionStartOperation(QtCore.QObject): return self._setStage(ConnectionStartStage.WaitingTUNDevice) + + if not self._isCurrent(): + return + self._conditionContinuation = continuation probe = _ConditionProbe( @@ -1000,6 +1026,9 @@ class ConnectionStartOperation(QtCore.QObject): self._setStage(ConnectionStartStage.StartingTUNRuntime) + if not self._isCurrent(): + return + try: self._startTUN() except RuntimeStartError as ex: @@ -1021,6 +1050,9 @@ class ConnectionStartOperation(QtCore.QObject): self._setStage(ConnectionStartStage.Committing) + if not self._isCurrent(): + return + self.attempt.commit(self.exitCallback) self._terminal = True @@ -1099,6 +1131,7 @@ class ConnectionManager(Mixins.CleanupOnExit): self.uniqueCleanup = False self._leases = list() + self._pendingReleases = [] self._lastStartError = '' self._startGeneration = 0 self._activeStartOperation = None @@ -1245,6 +1278,7 @@ class ConnectionManager(Mixins.CleanupOnExit): third-party plugins and non-interactive callers that rely on historical synchronous launch semantics. """ + self._retryPendingReleases() self._lastStartError = '' attempt = _ConnectionStartAttempt( @@ -1285,6 +1319,7 @@ class ConnectionManager(Mixins.CleanupOnExit): if self._activeStartOperation is not None: self._activeStartOperation.cancel() + self._retryPendingReleases() self._lastStartError = '' self._startGeneration += 1 @@ -1757,7 +1792,8 @@ class ConnectionManager(Mixins.CleanupOnExit): if lease is None: return - lease.release() + if not lease.release(): + self._pendingReleases.append(lease) self._leases.remove(lease) @@ -1765,20 +1801,29 @@ class ConnectionManager(Mixins.CleanupOnExit): """Stop every managed proxy-core and TUN runtime.""" self.cancelStart() - for lease in reversed(self._leases): - lease.release() - + self._pendingReleases.extend(reversed(self._leases)) self._leases.clear() + self._retryPendingReleases() + + def _retryPendingReleases(self): + """Keep exact failed resources and refuse replacement until release succeeds.""" + self._pendingReleases[:] = [ + lease for lease in self._pendingReleases if not lease.release() + ] + + if self._pendingReleases: + raise RuntimeError('Connection runtime cleanup is incomplete') def cleanup(self): """Release resources owned by the core manager.""" - self.cancelStart() - self.stopAll() + try: + self.stopAll() + finally: + # A failed runtime release must not strand independent DNS replies. + if self._dnsResolver is not None: + dispose = getattr(self._dnsResolver, 'dispose', None) - if self._dnsResolver is not None: - dispose = getattr(self._dnsResolver, 'dispose', None) + if callable(dispose): + dispose() - if callable(dispose): - dispose() - - self._dnsResolver = None + self._dnsResolver = None diff --git a/Furious/Service/RuntimeLease.py b/Furious/Service/RuntimeLease.py index c138fa1..0ae0afe 100644 --- a/Furious/Service/RuntimeLease.py +++ b/Furious/Service/RuntimeLease.py @@ -171,6 +171,7 @@ class RuntimeLease: self.runtime = runtime self.router = router + self._releaseInProgress = False @property def state(self): @@ -182,23 +183,55 @@ class RuntimeLease: self.router.commit(callback) def release(self): - """Idempotently stop and dispose the exact owned runtime.""" - if not self.router.beginRelease(): - return + """Release execution, retaining failed cleanup for the owner to retry.""" + if self.state is RuntimeLeaseState.Released: + return True + + if self._releaseInProgress: + return False + + self.router.beginRelease() + self._releaseInProgress = True + + stopped, disposed = True, True try: - self.runtime.stop() - except Exception as ex: - # Any non-exit exceptions + try: + self.runtime.stop() + except Exception as ex: + # Any non-exit exceptions + + stopped = False + + logger.error(f'error stopping core runtime: {ex}') - logger.error(f'error stopping core runtime: {ex}') - finally: try: self.runtime.dispose() except Exception as ex: # Any non-exit exceptions + disposed = False + logger.error(f'error disposing core runtime: {ex}') + if not stopped or not disposed: + return False + + try: + if self.runtime.isRunning(): + logger.error('core runtime remains alive after disposal') + + return False + except Exception as ex: + # Any non-exit exceptions + + logger.error(f'could not verify core runtime shutdown: {ex}') + + return False + self.router.finishRelease() self.router.deleteLater() + + return True + finally: + self._releaseInProgress = False diff --git a/tests/README.md b/tests/README.md index c1363ff..755f8db 100644 --- a/tests/README.md +++ b/tests/README.md @@ -89,9 +89,9 @@ worker. Choose tests by the changed contract rather than by filename alone. | [test_application_process.py](test_application_process.py) | Exact application-child ownership, shared crash flag, exception/signal handling, temporary crash logs, command-line dispatch. | | [test_architecture_refactors.py](test_architecture_refactors.py) | Startup acquisition/rollback including partial controller construction and cleanup retry, singleton election and real isolated IPC race, tray/exit policy, host integration ownership, bounded core-log transport, connection transactions, stylesheet composition. | | [test_connection_startup_async.py](test_connection_startup_async.py) | Real local-listener readiness, timeout/cancel/replacement, semantic exits, DNS reply lifetime, mocked platform-specific TUN sequencing. | -| [test_controllers.py](test_controllers.py) | Connection state/error/reconnect transitions, startup restoration, shared settings, routing fallback persistence and tray/selector agreement after custom-routing disable/re-enable. | -| [test_runtime_lifecycle.py](test_runtime_lifecycle.py) | Qt-thread exit dispatch, commit/exit races, duplicate and late exits, idempotent release, spawn failure, queue/timer disposal on preparation failure. | -| [test_external_core.py](test_external_core.py) | Harmless real process launch/output/shutdown, partial thread-start rollback, non-finite timeout rejection, readiness/TUN metadata, Windows paths with spaces, subscription rejection of executable profiles, bounded DNS references. | +| [test_controllers.py](test_controllers.py) | Connection state/error/reconnect transitions, reentrant cancellation/replacement before launch and during completion, startup restoration, shared settings, routing fallback persistence and tray/selector agreement after custom-routing disable/re-enable. | +| [test_runtime_lifecycle.py](test_runtime_lifecycle.py) | Qt-thread exit dispatch, commit/exit races, duplicate and late exits, idempotent release and retained failure/retry ownership, failed reap/handle close, spawn failure, queue/timer disposal on preparation failure. | +| [test_external_core.py](test_external_core.py) | Harmless real process launch/output/shutdown, partial thread-start rollback, non-finite timeout rejection, failed reap/thread-join retry, readiness/TUN metadata, Windows paths with spaces, subscription rejection of executable profiles, bounded DNS references. | | [test_frozenlib.py](test_frozenlib.py) | Nested state guards, cleanup isolation, bounded caches/throttling, dual-stack probe selection, mocked proxy/DNS/routes/startup/session boundaries and failure handling. | | [test_native_tun_semantics.py](test_native_tun_semantics.py) | Xray/Hysteria2 runtime-copy TUN preservation/replacement, managed-TUN failures, download-test stripping, prevention of a second tun2socks owner. | | [test_subscription_sync.py](test_subscription_sync.py) | Group-local preparation/commit, stable duplicate identity, atomic failure, preservation of newer local metadata, rejection of changed source state. | diff --git a/tests/test_connection_startup_async.py b/tests/test_connection_startup_async.py index 5951d25..ac7e77a 100644 --- a/tests/test_connection_startup_async.py +++ b/tests/test_connection_startup_async.py @@ -158,6 +158,43 @@ class _ResolverFixture(QtCore.QObject): class ConnectionStartupAsyncTest(TestCase): """Verify readiness, cancellation, rollback, and compatibility.""" + def testStageListenerCancellationStopsNextAcquisitionOrCommit(self): + """Cancel through real Qt stage signals before the next ownership boundary.""" + for stage in ( + ConnectionStartStage.Preparing, + ConnectionStartStage.StartingPrimary, + ConnectionStartStage.Committing, + ): + with self.subTest(stage=stage): + manager = ConnectionManager() + self.managers.append(manager) + runtime = _Runtime() + registry = _Registry([PreparedRuntime(runtime)]) + with mock.patch.object( + manager, '_prepareTUNPolicy', return_value=(False, False) + ) as prepare, mock.patch( + 'Furious.Service.ConnectionManager.getPluginRegistry', + return_value=registry, + ), mock.patch( + 'sys.excepthook' + ) as qtErrors: + operation = manager.startAsync(_Configuration(), '', deepcopy=False) + operation.stageChanged.connect( + lambda current: operation.cancel() if current is stage else None + ) + operation.start() + self.assertEqual(manager.runtimes, []) + self.assertIsNone(manager._activeStartOperation) + if stage is ConnectionStartStage.Preparing: + prepare.assert_not_called() + if stage is not ConnectionStartStage.Committing: + self.assertEqual(runtime.startOptions, []) + else: + self.assertEqual(runtime.stopCount, 1) + self.assertEqual(runtime.disposeCount, 1) + processQtEvents() + qtErrors.assert_not_called() + def setUp(self): """Ensure a Qt application exists for real timer/socket delivery.""" self.app = application() diff --git a/tests/test_controllers.py b/tests/test_controllers.py index 62d5dc2..56767ca 100644 --- a/tests/test_controllers.py +++ b/tests/test_controllers.py @@ -177,6 +177,111 @@ class ConnectionControllerTest(unittest.TestCase): processQtEvents() + def testReentrantDisconnectPreventsStartupAndCompletion(self): + """Real signal callbacks cancel admission and later host/publication work.""" + for phase in ( + 'profile', + 'connecting', + 'progress', + 'runtimes', + 'finish', + 'connected', + 'notification', + ): + with self.subTest(phase=phase), isolatedSettings(), mock.patch( + 'sys.excepthook' + ) as qtErrors: + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, updatesManager=FixtureUpdatesManager() + ) + if phase == 'profile': + controller.activeProfileChanged.connect( + lambda profile: ( + controller.startDisconnection() + if profile is self.profile + else None + ) + ) + elif phase in ('connecting', 'connected'): + target = ( + ConnectionState.Connecting + if phase == 'connecting' + else ConnectionState.Connected + ) + controller.stateChanged.connect( + lambda state: ( + controller.startDisconnection() if state is target else None + ) + ) + else: + signal = { + 'progress': controller.progressStarted, + 'runtimes': controller.runtimesChanged, + 'finish': controller.progressFinished, + 'notification': controller.notificationRequested, + }[phase] + signal.connect(lambda *_: controller.startDisconnection()) + + with mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.off' + ), mock.patch.object( + controller, '_runPostConnectTasksOnce' + ) as postConnect: + admitted = controller.startConnection(self.profile) + if phase in ('profile', 'connecting', 'progress'): + self.assertFalse(admitted) + self.assertEqual(core.operations, []) + else: + self.assertTrue(admitted) + core.operations[0][0].succeed() + self.assertEqual(controller.state, ConnectionState.Disconnected) + self.assertIsNone(controller.activeProfile) + self.assertFalse(controller._actionTimer.isActive()) + self.assertEqual(AppSettings.get('Connect'), AppBinarySettings.OFF) + postConnect.assert_not_called() + if phase in ('profile', 'connecting', 'progress', 'runtimes'): + proxySet.assert_not_called() + controller.deleteLater() + processQtEvents() + qtErrors.assert_not_called() + + def testReentrantReplacementOfSameProfileKeepsNewStartPending(self): + """State/profile equality must not let an old success finish a new generation.""" + with isolatedSettings(): + core = FixtureAsyncCoreManager() + controller = ConnectionController( + coreManager=core, updatesManager=FixtureUpdatesManager() + ) + replaced = [] + + def replace(*_): + if replaced: + return + replaced.append(True) + controller.startDisconnection() + controller.startConnection(self.profile) + + with mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.set' + ) as proxySet, mock.patch( + 'Furious.Controllers.ConnectionController.SystemProxy.off' + ), mock.patch.object( + controller, '_runPostConnectTasksOnce' + ): + controller.startConnection(self.profile) + controller.runtimesChanged.connect(replace) + core.operations[0][0].succeed() + self.assertEqual(len(core.operations), 2) + self.assertIs(controller._startOperation, core.operations[1][0]) + self.assertTrue(controller.isConnecting()) + proxySet.assert_not_called() + controller.shutdown() + controller.deleteLater() + processQtEvents() + def testSuccessfulConnectionAndDisconnectionStateMachine(self): """Publish stable states while using only injected runtime resources.""" with isolatedSettings(): diff --git a/tests/test_external_core.py b/tests/test_external_core.py index f00b0b5..1b3d17d 100644 --- a/tests/test_external_core.py +++ b/tests/test_external_core.py @@ -117,6 +117,53 @@ class ExternalCoreProcessTest(unittest.TestCase): spawn.assert_not_called() runtime.dispose() + def testFailedReapRetainsChildUntilRetrySucceeds(self): + """Keep independently observed live execution after every escalation fails.""" + runtime = ExternalCoreProcess( + self.configuration(['-c', 'pass'], str(Path.cwd())) + ) + child = mock.Mock(pid=1234) + child.poll.return_value = None + runtime._process = child + runtime.setState(RuntimeState.Alive) + with mock.patch.object(runtime, '_requestStop'), mock.patch.object( + runtime, '_terminate' + ), mock.patch.object(runtime, '_kill'), mock.patch.object( + runtime, '_waitForExit', return_value=False + ): + with self.assertRaisesRegex(RuntimeError, 'could not be reaped'): + runtime.dispose() + self.assertIsNone(child.poll()) + self.assertIs(runtime.process, child) + self.assertIs(runtime.state, RuntimeState.Stopping) + child.poll.return_value = 0 + runtime.dispose() + self.assertIsNone(runtime.process) + self.assertIs(runtime.state, RuntimeState.Disposed) + + def testFailedThreadJoinRetainsReaderAndWatcherUntilRetry(self): + """A reaped child does not prove its pipe readers or watcher have exited.""" + runtime = ExternalCoreProcess( + self.configuration(['-c', 'pass'], str(Path.cwd())) + ) + child = mock.Mock() + child.poll.return_value = 0 + reader, watcher = mock.Mock(), mock.Mock() + reader.is_alive.return_value = watcher.is_alive.return_value = True + runtime._process = child + runtime._readerThreads = [reader] + runtime._watcherThread = watcher + with self.assertRaisesRegex(RuntimeError, 'reader or watcher'): + runtime.dispose() + self.assertIs(runtime.process, child) + self.assertEqual(runtime._readerThreads, [reader]) + self.assertIs(runtime._watcherThread, watcher) + reader.is_alive.return_value = watcher.is_alive.return_value = False + runtime.dispose() + self.assertEqual(runtime._readerThreads, []) + self.assertIsNone(runtime._watcherThread) + self.assertIsNone(runtime.process) + def testDisposedRuntimeCannotAcquireAnotherProcess(self): """Disposal is terminal even when the stored launch specification is valid.""" runtime = ExternalCoreProcess( @@ -471,6 +518,7 @@ class ExternalCoreProcessTest(unittest.TestCase): dnsResolver.resolve.return_value = (True, []) tunRuntime = mock.Mock(spec=CoreRuntime) + tunRuntime.isRunning.return_value = False manager = NoCoreRuntimeConnectionManager(dnsResolver=dnsResolver) diff --git a/tests/test_runtime_lifecycle.py b/tests/test_runtime_lifecycle.py index 8aead3a..6b50c3c 100644 --- a/tests/test_runtime_lifecycle.py +++ b/tests/test_runtime_lifecycle.py @@ -24,7 +24,12 @@ from Furious.Interface import ( RuntimeStartError, RuntimeState, ) -from Furious.Service.RuntimeLease import RuntimeEventRouter, RuntimeLease +from Furious.Service.RuntimeLease import ( + RuntimeEventRouter, + RuntimeLease, + RuntimeLeaseState, +) +from Furious.Service.ConnectionManager import ConnectionManager, _ConnectionStartAttempt from PySide6 import QtCore @@ -99,6 +104,98 @@ class RuntimeLifecycleTest(TestCase): def setUpClass(cls): cls.app = application() + def testFailedLeaseReleaseRemainsOwnedUntilSuccessfulRetry(self): + """Rollback retains failed resources and blocks replacement admission.""" + manager = ConnectionManager() + attempt = _ConnectionStartAttempt(manager, None) + owner = _AttemptOwner() + runtime = _Runtime() + router = RuntimeEventRouter() + router.attach(runtime, owner) + lease = attempt.ownRuntime(runtime, router) + with mock.patch.object( + runtime, 'dispose', side_effect=RuntimeError('still owned') + ): + attempt.rollback() + self.assertEqual(manager._pendingReleases, [lease]) + self.assertIs(lease.state, RuntimeLeaseState.Releasing) + runtime.publishCode(0) + processQtEvents() + self.assertEqual(owner.events, []) + for start in (manager.start, manager.startAsync): + with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'): + start(None, '') + with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'): + manager.stopAll() + self.assertEqual(manager._pendingReleases, [lease]) + resolver = mock.Mock() + manager._dnsResolver = resolver + with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'): + manager.cleanup() + resolver.dispose.assert_called_once_with() + self.assertIsNone(manager._dnsResolver) + self.assertEqual(manager._pendingReleases, [lease]) + manager.stopAll() + self.assertEqual(manager._pendingReleases, []) + self.assertIs(lease.state, RuntimeLeaseState.Released) + manager.cleanup() + owner.deleteLater() + processQtEvents() + + def testFailedCommittedReleaseStillCleansIndependentRuntimes(self): + """One refusal must not prevent reverse release of other committed leases.""" + manager = ConnectionManager() + first, second = _Runtime(), _Runtime() + leases = [ + RuntimeLease(runtime, RuntimeEventRouter()) for runtime in (first, second) + ] + manager._leases.extend(leases) + with mock.patch.object(second, 'dispose', side_effect=RuntimeError('retry')): + with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'): + manager.stopAll() + self.assertIs(first.state, RuntimeState.Disposed) + self.assertEqual(manager._pendingReleases, [leases[1]]) + manager.stopAll() + manager.cleanup() + processQtEvents() + + def testMultiprocessingFailedEscalationRetainsLiveChild(self): + """Do not manufacture an exit or lose the handle of a surviving child.""" + runtime = _ProcessRuntime() + child = mock.Mock() + child.is_alive.return_value = True + child.exitcode = None + runtime._process = child + runtime.setState(RuntimeState.Alive) + with self.assertRaisesRegex(RuntimeError, 'did not stop'): + runtime.dispose() + self.assertTrue(child.is_alive()) + self.assertIs(runtime.process, child) + self.assertIsNone(runtime.lastExit) + child.close.assert_not_called() + child.is_alive.return_value = False + child.exitcode = 0 + runtime.dispose() + self.assertIsNone(runtime.process) + self.assertIs(runtime.state, RuntimeState.Disposed) + + def testMultiprocessingCloseFailureRetainsHandleForRetry(self): + """Physical exit and handle release need independent evidence.""" + runtime = _ProcessRuntime() + child = mock.Mock() + child.is_alive.return_value = False + child.exitcode = 0 + child.close.side_effect = OSError('handle still owned') + runtime._process = child + with self.assertRaisesRegex(RuntimeError, 'could not be closed'): + runtime.dispose() + self.assertIs(runtime.process, child) + self.assertIsNot(runtime.state, RuntimeState.Disposed) + child.close.side_effect = None + runtime.dispose() + self.assertIsNone(runtime.process) + self.assertIs(runtime.state, RuntimeState.Disposed) + def testWatcherThreadExitIsConsumedOnRouterQtThread(self): """Queue a worker-thread publication onto the Qt owner's thread.""" owner = _AttemptOwner()