From e3fc1469bf82b5ddb1e5996388bf702f4c5b0924 Mon Sep 17 00:00:00 2001 From: Loren Eteval Date: Sat, 19 Sep 2026 21:05:54 +0800 Subject: [PATCH] Retain failed download runtime cleanup for retry Signed-off-by: Loren Eteval --- Furious/Service/AGENTS.md | 3 + Furious/Service/ProfileTesting.py | 66 +++++++-- tests/README.md | 2 +- tests/test_profile_test_jobs.py | 215 ++++++++++++++++++++++++++++++ 4 files changed, 275 insertions(+), 11 deletions(-) diff --git a/Furious/Service/AGENTS.md b/Furious/Service/AGENTS.md index 65cea72..858e2a5 100644 --- a/Furious/Service/AGENTS.md +++ b/Furious/Service/AGENTS.md @@ -99,6 +99,9 @@ for execution, and Qt for lifetime primitives. This scope owns multi-stage workf - Download jobs own a temporary proxy-only runtime, readiness timer, port, network reply, and cancellation path. Serial and concurrent admission share scheduler semantics; startup never blocks admission on a grace wait. Reentrant cancellation defers terminal deletion until the active start frame unwinds. + Failed runtime release transfers its lease from the terminal worker to the scheduler before worker deletion. + Keep the port and concurrency slot reserved until a later drain/cancel/shutdown retries cleanup successfully; + final shutdown reports remaining leases and preserves retry ownership. Completion is not proof of resource release. ## Verification diff --git a/Furious/Service/ProfileTesting.py b/Furious/Service/ProfileTesting.py index 34e3873..0c0c4d5 100644 --- a/Furious/Service/ProfileTesting.py +++ b/Furious/Service/ProfileTesting.py @@ -762,7 +762,7 @@ class _DownloadSpeedWorker(HttpGetManager): self.progressed.emit(self, self.result) def completionCallback(self, **_kwargs): - """Dispose runtime callbacks before publishing terminal completion.""" + """Attempt runtime cleanup before publishing terminal completion.""" self.coreStartupTimer.stop() self.timeoutTimer.stop() @@ -778,13 +778,19 @@ class _DownloadSpeedWorker(HttpGetManager): return lease is not None and lease.runtime.isRunning() def _releaseRuntime(self): - """Release this worker's exact runtime lease once.""" + """Forget the exact lease only after all of its resources are released.""" + lease = self._runtimeLease + + if lease is not None and lease.release(): + self._runtimeLease = None + + def takeRuntimeLease(self): + """Transfer unfinished cleanup to the scheduler before worker deletion.""" lease = self._runtimeLease self._runtimeLease = None - if lease is not None: - lease.release() + return lease def runCompletionCallback(self, **kwargs): """Defer terminal publication until synchronous startup has unwound.""" @@ -1103,11 +1109,16 @@ class _DownloadSpeedScheduler(QtCore.QObject): self.queue = collections.deque() self.activeJobs = {} self.activePorts = set() + self._pendingReleases = {} + self._shuttingDown = False self.nextPort = portRange.start self.drainScheduled = False def enqueue(self, profiles, options: DownloadSpeedTestOptions): """Capture each profile with the same explicit operation options.""" + if self._shuttingDown: + return + self.queue.extend( _DownloadSpeedTestJob(ProfileTestTarget.capture(profile), options) for profile in profiles @@ -1117,6 +1128,8 @@ class _DownloadSpeedScheduler(QtCore.QObject): def cancelAll(self): """Cancel every pending and active job through one terminal path.""" + self._retryPendingReleases() + for job in self.queue: job.state = ProfileTestJobState.Cancelled @@ -1127,9 +1140,26 @@ class _DownloadSpeedScheduler(QtCore.QObject): worker.cancel() + def _retryPendingReleases(self): + """Keep failed runtimes and their ports owned until cleanup succeeds.""" + for port, lease in tuple(self._pendingReleases.items()): + if lease.release(): + del self._pendingReleases[port] + + self.activePorts.discard(port) + + def shutdown(self): + """Close admission and report resources still retained for another retry.""" + self._shuttingDown = True + + self.cancelAll() + + if self._pendingReleases: + raise RuntimeError('Download runtime cleanup is incomplete') + def scheduleDrain(self): """Schedule valid pending jobs without recursive startup.""" - if self.drainScheduled: + if self.drainScheduled or self._shuttingDown: return self.drainScheduled = True @@ -1140,12 +1170,20 @@ class _DownloadSpeedScheduler(QtCore.QObject): """Start valid jobs while concurrency and local ports are available.""" self.drainScheduled = False + if self._shuttingDown: + return + if _appIsExiting(): self.cancelAll() return - while self.queue and len(self.activeJobs) < self.maxConcurrency: + self._retryPendingReleases() + + while ( + self.queue + and len(self.activeJobs) + len(self._pendingReleases) < self.maxConcurrency + ): job = self.queue.popleft() if self._resolveTarget(job.target) is None: @@ -1224,13 +1262,19 @@ class _DownloadSpeedScheduler(QtCore.QObject): @QtCore.Slot(object, object) def handleWorkerFinished(self, worker, result): - """Release one terminal worker after disposing its runtime callbacks.""" + """Transfer outstanding cleanup before deleting one terminal worker.""" active = self.activeJobs.pop(id(worker), None) if active is None: return _, job, port = active + lease = worker.takeRuntimeLease() + + # Completion may notify reentrant consumers. Establish the durable + # cleanup owner before publishing a result or deleting the worker. + if lease is not None: + self._pendingReleases[port] = lease if job.state is not ProfileTestJobState.Cancelled: if self._publishResult(job.target, result): @@ -1238,7 +1282,9 @@ class _DownloadSpeedScheduler(QtCore.QObject): else: job.state = ProfileTestJobState.Cancelled - self.activePorts.discard(port) + if lease is None: + self.activePorts.discard(port) + worker.deleteLater() self.scheduleDrain() @@ -1477,6 +1523,6 @@ class ProfileTestManager(QtCore.QObject): self._latencyScheduler.shutdown() finally: try: - self._serialDownloadScheduler.cancelAll() + self._serialDownloadScheduler.shutdown() finally: - self._concurrentDownloadScheduler.cancelAll() + self._concurrentDownloadScheduler.shutdown() diff --git a/tests/README.md b/tests/README.md index 755f8db..8e714f5 100644 --- a/tests/README.md +++ b/tests/README.md @@ -97,7 +97,7 @@ worker. Choose tests by the changed contract rather than by filename alone. | [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. | | [test_subscription_manager.py](test_subscription_manager.py) | Stable request generations, provider metadata, batch/partial failures, timer policy, targeted updates, real Stop Updates input, worker responsiveness, cancellation and synchronous shutdown ownership. | | [test_subscription_scalability.py](test_subscription_scalability.py) | Deterministic 1/3/8-group preparation and commit with 1,500 profiles per group and bounded workers; uses the offline benchmark helper. | -| [test_profile_test_jobs.py](test_profile_test_jobs.py) | Stable profile/fingerprint jobs, stale results, endpoint deduplication, bounded fan-out, adaptive Tcping, download scheduling, reusable Stop All, real pool/thread teardown and retry, narrow cell repaint. | +| [test_profile_test_jobs.py](test_profile_test_jobs.py) | Stable profile/fingerprint jobs, stale results, endpoint deduplication, bounded fan-out, adaptive Tcping, download scheduling and failed runtime/port cleanup ownership, reusable Stop All, real pool/thread teardown and retry, narrow cell repaint. | | [test_service_runtime.py](test_service_runtime.py) | Update-response validation, HTTP timeouts/context release/reentrant destruction, duplicate completion, plugin-page registration, bounded connectivity requests, blocked statistics-worker callback lifetime. | | [test_xray_asset_download.py](test_xray_asset_download.py) | Checksum validation, atomic asset replacement, real pool delivery, early owner destruction, callback release, plugin shutdown. | | [test_endpoint_info.py](test_endpoint_info.py) | Opt-in proxy-only discovery using fake HTTP responses, fallback/cache/session invalidation, privacy/presentation controls, local map styling and persistent-scene contracts. | diff --git a/tests/test_profile_test_jobs.py b/tests/test_profile_test_jobs.py index 3aa7ee9..f1e895f 100644 --- a/tests/test_profile_test_jobs.py +++ b/tests/test_profile_test_jobs.py @@ -56,6 +56,7 @@ from tests.support import ( import threading import unittest +from contextlib import contextmanager from types import SimpleNamespace from unittest import mock @@ -86,6 +87,10 @@ class _ControlledDownloadWorker(QtCore.QObject): def start(self): """Leave completion under explicit test control.""" + def takeRuntimeLease(self): + """This fixture owns no runtime resources to transfer.""" + return None + def publish(self, speed): """Publish one non-terminal result without mutating the snapshot.""" if self.terminal: @@ -214,6 +219,44 @@ class _ImmediateRuntime(CoreRuntime): self.setState(RuntimeState.Exited) +class _CleanupRefusingRuntime(_ImmediateRuntime): + """Keep execution/resource evidence independent of the owner's references.""" + + def __init__(self, failure, **kwargs): + super().__init__(**kwargs) + + self.failure = failure + self.executionAlive = False + self.resourceOwned = False + + def start(self): + super().start() + + self.executionAlive = True + self.resourceOwned = True + + def isRunning(self): + return self.executionAlive + + def stop(self): + if self.failure == 'stop': + raise RuntimeError('execution still running') + + self.executionAlive = False + + super().stop() + + def dispose(self): + self.stop() + + if self.failure == 'dispose': + raise RuntimeError('resource still owned') + + self.resourceOwned = False + + super().dispose() + + class _CancelDuringStartDownloadWorker(_DownloadSpeedWorker): """Re-enter subscription invalidation while a worker is starting.""" @@ -311,6 +354,178 @@ class ProfileTestServiceTest(unittest.TestCase): return manager + @contextmanager + def _runtimeDownloads(self, profiles, failure='stop'): + """Use real workers/leases and Qt delivery without processes or network.""" + manager = self._manager(profiles, controlledDownloads=False) + workers, runtimes = [], [] + + def workerFactory(*args, **kwargs): + worker = _DownloadSpeedWorker(*args, **kwargs) + worker.CoreStartupGraceMilliseconds = 60_000 + workers.append(worker) + + return worker + + def createRuntime(*_args, **kwargs): + runtime = _CleanupRefusingRuntime( + failure, exitCallback=kwargs['exitCallback'] + ) + runtimes.append(runtime) + + return PreparedRuntime(runtime) + + for scheduler in ( + manager._serialDownloadScheduler, + manager._concurrentDownloadScheduler, + ): + scheduler.workerFactory = workerFactory + + registry = mock.Mock() + registry.prepareDownloadTest.side_effect = lambda profile, _port: profile + registry.createCoreRuntime.side_effect = createRuntime + + with mock.patch( + 'Furious.Service.ProfileTesting.getPluginRegistry', return_value=registry + ), mock.patch('Furious.Service.ProfileTesting.AppLogManager'), mock.patch( + 'sys.excepthook' + ) as qtErrors: + try: + yield manager, workers, runtimes + + qtErrors.assert_not_called() + finally: + for runtime in runtimes: + runtime.failure = None + runtime.dispose() + + manager.shutdown() + + processQtEvents() + + def testDownloadWorkerRetainsFailedLeaseUntilReleaseSucceeds(self): + """A failed stop or dispose must leave the worker's exact lease reachable.""" + for failure in ('stop', 'dispose'): + with self.subTest(failure=failure): + profile = self._profile('profile', 'example.test') + + with self._runtimeDownloads((profile,), failure) as ( + manager, + workers, + runtimes, + ): + manager.testDownloadSpeed((profile,), concurrent=False) + processQtEvents() + + worker, runtime = workers[0], runtimes[0] + lease = worker._runtimeLease + + with self.assertLogs('Furious.Service.RuntimeLease', level='ERROR'): + worker._releaseRuntime() + + self.assertIs(worker._runtimeLease, lease) + self.assertTrue(runtime.resourceOwned) + self.assertEqual(runtime.isRunning(), failure == 'stop') + + runtime.failure = None + worker._releaseRuntime() + + self.assertIsNone(worker._runtimeLease) + self.assertFalse(runtime.resourceOwned) + self.assertFalse(runtime.isRunning()) + + def testDownloadCompletionRetainsFailedRuntimeAndPortUntilRetry(self): + """A terminal worker can die while its scheduler retains unreleased execution.""" + for concurrent in (False, True): + with self.subTest(concurrent=concurrent): + profiles = [self._profile(str(i), f'{i}.example') for i in range(2)] + + with self._runtimeDownloads(profiles) as (manager, workers, runtimes): + scheduler = ( + manager._concurrentDownloadScheduler + if concurrent + else manager._serialDownloadScheduler + ) + results = [] + manager.resultApplied.connect( + lambda _profile, result: results.append(result) + ) + manager.testDownloadSpeed(profiles, concurrent=concurrent) + processQtEvents() + + worker, runtime = workers[0], runtimes[0] + port = worker.port + lease = worker._runtimeLease + destroyed = [] + worker.destroyed.connect(lambda: destroyed.append(True)) + results.clear() + + with self.assertLogs('Furious.Service.RuntimeLease', level='ERROR'): + worker.setResult('1.00 MiB/s', publish=False) + worker.runCompletionCallback() + + processQtEvents() + + self.assertTrue(runtime.isRunning()) + self.assertTrue(runtime.resourceOwned) + self.assertEqual(destroyed, [True]) + self.assertEqual(len(workers), 1) + self.assertIn(port, scheduler.activePorts) + self.assertIs(scheduler._pendingReleases[port], lease) + self.assertEqual( + [result.value for result in results], ['1.00 MiB/s'] + ) + self.assertEqual(len(scheduler.queue), 1) + + runtime.publishExit(RuntimeExit(1, RuntimeExitReason.Unexpected)) + processQtEvents() + self.assertEqual(len(results), 1) + + runtime.failure = None + scheduler.scheduleDrain() + processQtEvents() + + self.assertFalse(runtime.isRunning()) + self.assertFalse(runtime.resourceOwned) + self.assertFalse(scheduler._pendingReleases) + self.assertEqual(len(workers), 2) + self.assertEqual(scheduler.activePorts, {workers[1].port}) + self.assertFalse(isValid(lease.router)) + + def testDownloadShutdownRetainsFailuresAndRetriesBothSchedulers(self): + """Failed final cleanup keeps its owner and still cleans independent jobs.""" + profiles = [self._profile(str(i), f'{i}.example') for i in range(2)] + + with self._runtimeDownloads(profiles) as (manager, workers, runtimes): + manager.testDownloadSpeed(profiles[:1], concurrent=False) + manager.testDownloadSpeed(profiles[1:], concurrent=True) + processQtEvents() + self.assertEqual(len(workers), 2) + runtimes[1].failure = None + + with self.assertLogs('Furious.Service.RuntimeLease', level='ERROR'): + with self.assertRaisesRegex(RuntimeError, 'cleanup is incomplete'): + manager.shutdown() + processQtEvents() + + self.assertTrue(runtimes[0].resourceOwned) + self.assertFalse(runtimes[1].resourceOwned) + self.assertTrue(manager._serialDownloadScheduler._pendingReleases) + self.assertFalse(manager._concurrentDownloadScheduler.activePorts) + self.assertTrue(all(not isValid(worker) for worker in workers)) + + manager.testDownloadSpeed(profiles) + processQtEvents() + self.assertEqual(len(workers), 2) + + runtimes[0].failure = None + manager.shutdown() + processQtEvents() + + self.assertFalse(runtimes[0].resourceOwned) + self.assertFalse(manager._serialDownloadScheduler._pendingReleases) + self.assertFalse(manager._serialDownloadScheduler.activePorts) + def testCancelAllPreservesResultsRejectsLatePingAndAllowsNewTests(self): """Cancel active and queued work across all schedulers without shutting down.""" profiles = [self._profile(str(i), f'{i}.example') for i in range(3)]