Retain failed download runtime cleanup for retry

Signed-off-by: Loren Eteval <loren.eteval@proton.me>

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
 

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()

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. |

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)]