Scale subscription update processing

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

Furious/Backends/ExternalCore/Protocols.py

@@ -47,6 +47,7 @@ _TRANSLATABLE = (
 class ExternalCoreProtocolHandler(ProtocolHandler):
     """Own local External Core mappings while declining URI import/export."""
 
+    workerSafe = True
     descriptor = ProtocolDescriptor(
         id=EXTERNAL_CORE_TYPE,
         displayName='External Core',

Furious/Backends/Hysteria1/Protocols.py

@@ -51,6 +51,7 @@ _TRANSLATABLE = (
 class Hysteria1ProtocolHandler(ProtocolHandler):
     """Own Hysteria 1 URI, mapping, validation, and export behavior."""
 
+    workerSafe = True
     descriptor = ProtocolDescriptor(
         id='hysteria1',
         displayName='Hysteria1',

Furious/Backends/Hysteria2/Protocols.py

@@ -51,6 +51,7 @@ _TRANSLATABLE = (
 class Hysteria2ProtocolHandler(ProtocolHandler):
     """Own Hysteria 2 URI, mapping, validation, and export behavior."""
 
+    workerSafe = True
     descriptor = ProtocolDescriptor(
         id='hysteria2',
         displayName='Hysteria2',

Furious/Backends/Xray/Protocols.py

@@ -48,6 +48,8 @@ __all__ = ['XRAY_PROTOCOL_HANDLERS']
 class XrayProtocolHandler(ProtocolHandler):
     """Adapt one Xray outbound protocol to the host protocol contract."""
 
+    workerSafe = True
+
     def __init__(
         self,
         descriptor,

Furious/Extensions/StandardSubscriptions.py

@@ -50,6 +50,7 @@ def _shareLinks(text: str):
 class PlainShareLinkDecoder(SubscriptionDecoder):
     """Decode newline-delimited plain-text share links."""
 
+    workerSafe = True
     decoderId = 'plain-share-links'
     displayName = 'Share Links (plain text)'
     priority = 100
@@ -76,6 +77,7 @@ class PlainShareLinkDecoder(SubscriptionDecoder):
 class Base64ShareLinkDecoder(SubscriptionDecoder):
     """Decode a Base64 envelope containing plain share links."""
 
+    workerSafe = True
     decoderId = 'base64-share-links'
     displayName = 'Share Links (Base64)'
     priority = 90

Furious/Plugins/AGENTS.md

@@ -25,6 +25,9 @@
   entry-point plugins; do not give bundled code hidden repository/UI side channels.
 - Evolve contracts additively when practical. Before a breaking change, inspect external discovery, compatibility
   exports, every bundled implementation, tests, and compiled inclusion; do not infer compatibility from built-ins alone.
+- Capability instances default to GUI-thread-only for background subscription preparation. A decoder or protocol
+  handler opts into worker execution only after its parsing, validation, caches, globals, and Qt usage are audited as
+  safe for concurrent copied inputs; keep unclassified third-party capability execution on the GUI thread.
 
 ## Verification
 

Furious/Plugins/API.py

@@ -91,6 +91,9 @@ class PluginCapability:
     """Define one independently queryable plugin capability."""
 
     capabilityKind = CapabilityKind.Utility
+    # Capability instances may be process-lifetime and externally supplied.
+    # Worker execution therefore requires an explicit opt-in after auditing.
+    workerSafe = False
 
     @property
     def capabilityId(self) -> str:

Furious/Plugins/Registry.py

@@ -576,6 +576,37 @@ class PluginRegistry:
             )
         )
 
+    def subscriptionDecoderWorkerSafe(self, decoderId=None) -> bool:
+        """Return whether the selected subscription decoder path is worker-safe."""
+        if decoderId:
+            entry = self._decoders.get(_normalizeIdentifier(decoderId))
+            decoders = (entry[1],) if entry is not None else tuple()
+        else:
+            decoders = self.subscriptionDecoders()
+
+        return bool(decoders) and all(
+            bool(getattr(decoder, 'workerSafe', False)) for decoder in decoders
+        )
+
+    def subscriptionItemWorkerSafe(self, item: SubscriptionItem) -> bool:
+        """Classify only the protocol capability needed by one decoded item."""
+        if item.uri is not None:
+            entry = self._schemes.get(_schemeFromURI(item.uri))
+
+            return entry is None or bool(getattr(entry[1], 'workerSafe', False))
+
+        protocol = item.configuration.get('type', '')
+        entry = self._protocols.get(_normalizeIdentifier(protocol))
+
+        if entry is not None:
+            return bool(getattr(entry[1], 'workerSafe', False))
+
+        # Undiscriminated mappings are probed against every handler.
+        return all(
+            bool(getattr(handler, 'workerSafe', False))
+            for _plugin, handler in self._protocolEntries
+        )
+
     def trafficStatsProviders(self):
         """Return registered runtime traffic-statistics providers."""
         return self.capabilities(CapabilityKind.TrafficStats)

Furious/Repository/Storage.py

@@ -123,6 +123,16 @@ class Storage:
         """Persist one subscription group through the shared repository."""
         Storage._UserSubsStorage().upsertGroup(group)
 
+    @staticmethod
+    def upsertSubscriptionGroups(groups):
+        """Mutate a validated subscription-group batch in one commit."""
+        Storage._UserSubsStorage().upsertGroups(groups)
+
+    @staticmethod
+    def persistSubscriptionGroups():
+        """Serialize the current subscription repository once."""
+        Storage._UserSubsStorage().sync()
+
     @staticmethod
     def removeSubscriptionGroup(unique: str) -> SubscriptionGroup | None:
         """Remove one subscription group through the shared repository."""

Furious/Repository/Subscriptions.py

@@ -221,10 +221,22 @@ class UserSubs(Mixins.CleanupOnExit, StorageBackend):
 
     def upsertGroup(self, group: SubscriptionGroup):
         """Insert or replace one group without changing its identity."""
-        if not group.id:
-            raise ValueError('subscription group ID must not be empty')
+        self.upsertGroups((group,))
 
-        self._data[group.id] = group.toMapping()
+    def upsertGroups(self, groups):
+        """Atomically insert or replace a batch of validated groups."""
+        staged = {}
+
+        for group in groups:
+            if not isinstance(group, SubscriptionGroup):
+                raise TypeError('subscription group must be a SubscriptionGroup')
+
+            if not group.id:
+                raise ValueError('subscription group ID must not be empty')
+
+            staged[group.id] = group.toMapping()
+
+        self._data.update(staged)
 
     def removeGroup(self, unique: str) -> SubscriptionGroup | None:
         """Remove and return one group definition."""

Furious/Service/AGENTS.md

@@ -22,6 +22,9 @@
   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;
   synchronization prepares one group reconciliation; the manager owns request/schedule generations and commits it.
+  Large payload import and reconciliation preparation run in the manager's bounded pool over copied payload/profile
+  data. Workers never read live repositories or Qt models; the GUI thread verifies the full source signature and group
+  revision, commits while preserving live profile identity/local metadata, then publishes coalesced status/structure.
   Post-commit reconnect/test invalidation failure is reported without undoing the committed profiles.
 - Log transport, traffic collection, and metric history remain bounded and independent of page visibility. Rendering may
   be lazy; collection/draining ownership is not.

Furious/Service/SubscriptionImporter.py

@@ -30,9 +30,14 @@ __all__ = [
     'SubscriptionImportResult',
     'SubscriptionImportService',
     'SubscriptionSource',
+    'SubscriptionWorkerUnsafe',
 ]
 
 
+class SubscriptionWorkerUnsafe(RuntimeError):
+    """Signal that one plugin capability must remain on the GUI thread."""
+
+
 @dataclass(frozen=True)
 class SubscriptionSource:
     """Describe where a subscription payload came from."""
@@ -62,8 +67,25 @@ class SubscriptionImportService:
         """Use the supplied capability registry or the application registry."""
         self.registry = registry or getPluginRegistry()
 
-    def importPayload(self, data: bytes, source: SubscriptionSource):
+    def importPayload(
+        self,
+        data: bytes,
+        source: SubscriptionSource,
+        *,
+        requireWorkerSafe: bool = False,
+        isCancelled=None,
+    ):
         """Decode *data* and construct supported profiles for *source*."""
+        if requireWorkerSafe and not self.registry.subscriptionDecoderWorkerSafe(
+            source.decoderId
+        ):
+            raise SubscriptionWorkerUnsafe(
+                'subscription capabilities did not opt in to worker execution'
+            )
+
+        if callable(isCancelled) and isCancelled():
+            return None
+
         result = self.registry.decodeSubscription(data, source.decoderId)
 
         if result is None:
@@ -74,6 +96,14 @@ class SubscriptionImportService:
         identityOccurrences = {}
 
         for item in result.items:
+            if callable(isCancelled) and isCancelled():
+                return None
+
+            if requireWorkerSafe and not self.registry.subscriptionItemWorkerSafe(item):
+                raise SubscriptionWorkerUnsafe(
+                    'subscription protocol capability did not opt in to worker execution'
+                )
+
             value = item.configuration if item.configuration is not None else item.uri
             metadata = {
                 **dict(item.metadata),

Furious/Service/SubscriptionManager.py

@@ -31,6 +31,11 @@ from Furious.Repository import Storage
 from Furious.Service.SubscriptionImporter import (
     SubscriptionImportService,
     SubscriptionSource,
+    SubscriptionWorkerUnsafe,
+)
+from Furious.Service.SubscriptionPreparation import (
+    SubscriptionPreparationJob,
+    SubscriptionPreparationRelay,
 )
 from Furious.Service.SubscriptionSync import SubscriptionSynchronizer
 
@@ -40,6 +45,8 @@ from PySide6.QtNetwork import QNetworkRequest
 from dataclasses import dataclass
 
 import re
+import os
+import time
 import logging
 import datetime
 
@@ -99,6 +106,17 @@ class SubscriptionUpdateBatch:
     showMessageBox: bool
 
 
+@dataclass
+class _SubscriptionBatchState:
+    """Track logical completion independently from network reply completion."""
+
+    pending: set
+    showMessageBox: bool
+    successful: list
+    failed: list
+    structural: bool = False
+
+
 class SubscriptionManager(HttpGetManager):
     """Own subscription networking, decoding, reconciliation, and persistence."""
 
@@ -124,6 +142,23 @@ class SubscriptionManager(HttpGetManager):
         self._requestVersions = {}
         self._activeReplies = {}
         self._replySubscriptions = {}
+        self._batches = {}
+        self._preparationJobs = {}
+        self._preparationPayloads = {}
+        self._nextBatchId = 0
+        self._nextPreparationJobId = 0
+        self._shuttingDown = False
+
+        self._preparationRelay = SubscriptionPreparationRelay(self)
+        self._preparationRelay.completed.connect(
+            self._handlePreparationOutcome,
+            QtCore.Qt.ConnectionType.QueuedConnection,
+        )
+
+        self._preparationPool = QtCore.QThreadPool(self)
+        self._preparationPool.setMaxThreadCount(
+            max(1, min((os.cpu_count() or 1) // 2, 4))
+        )
 
         self.refreshAutoUpdates()
 
@@ -157,9 +192,25 @@ class SubscriptionManager(HttpGetManager):
         subscription = Storage.UserSubs().get(unique)
 
         return bool(
-            subscription
+            not self._shuttingDown
+            and subscription
             and self._requestVersions.get(unique) == version
             and subscription.get('webURL') == kwargs.get('webURL')
+            and (
+                'requestSignature' not in kwargs
+                or self._requestSignature(subscription) == kwargs['requestSignature']
+            )
+        )
+
+    @staticmethod
+    def _requestSignature(subscription):
+        """Capture every source option whose edit invalidates prepared data."""
+        return (
+            subscription.get('webURL', ''),
+            subscription.get('enabled', True),
+            subscription.get('userAgent', ''),
+            subscription.get('filter', ''),
+            subscription.get('lastDecoderId', ''),
         )
 
     @QtCore.Slot(object)
@@ -296,7 +347,7 @@ class SubscriptionManager(HttpGetManager):
             self._pruneRequestVersion(unique)
 
     def cancelUpdates(self, unique: str | None = None):
-        """Cancel exact active replies and invalidate their eventual completions."""
+        """Cancel network and preparation work and invalidate eventual completions."""
         if unique is None:
             subscriptions = {
                 *self._requestVersions,
@@ -312,6 +363,332 @@ class SubscriptionManager(HttpGetManager):
             if unique is None or self._replySubscriptions.get(reply) == unique:
                 reply.abort()
 
+        for job in tuple(self._preparationJobs.values()):
+            if unique is None or job.context.get('unique') == unique:
+                job.cancel()
+
+    def shutdown(self):
+        """Boundedly stop every manager-owned request, timer, and preparation job."""
+        if self._shuttingDown:
+            return
+
+        self._shuttingDown = True
+
+        for timer in self._autoUpdateTimers.values():
+            timer.stop()
+
+        self.cancelUpdates()
+        self._preparationPool.clear()
+
+        for job in self._preparationJobs.values():
+            job.cancel()
+
+        # Running third-party Python code may not be interruptible. Keep the
+        # relay/pool alive until those exact jobs finish rather than allowing a
+        # worker to publish through a destroyed QObject during application exit.
+        self._preparationPool.waitForDone()
+
+        self._preparationJobs.clear()
+        self._preparationPayloads.clear()
+        self._batches.clear()
+
+    @staticmethod
+    def _filterImportResult(result, profileFilter: str, remark: str):
+        """Apply a copied regex filter as part of subscription preparation."""
+        profileFilter = str(profileFilter).strip()
+
+        if result is None or not profileFilter:
+            return result
+
+        try:
+            pattern = re.compile(profileFilter, re.IGNORECASE)
+        except re.error as ex:
+            logger.error(
+                f'invalid subscription filter for {remark!r}: {ex}. '
+                f'Importing all profiles'
+            )
+
+            return result
+
+        return type(result)(
+            result.decoderId,
+            tuple(
+                profile
+                for profile in result.profiles
+                if pattern.search(str(getattr(profile, 'itemRemark', '')))
+            ),
+            result.rejectedItems,
+        )
+
+    def _startPreparationJob(self, stage: str, context: dict, work):
+        """Dispatch one copied operation to the bounded preparation pool."""
+        if not self._isCurrentRequest(context):
+            self._finishOperation(context)
+
+            return None
+
+        self._nextPreparationJobId += 1
+
+        jobId = self._nextPreparationJobId
+        job = SubscriptionPreparationJob(
+            jobId,
+            stage,
+            context,
+            work,
+            self._preparationRelay,
+        )
+
+        self._preparationJobs[jobId] = job
+        self._preparationPool.start(job)
+
+        return jobId
+
+    def _startImportPreparation(self, data: bytes, context: dict):
+        """Decode, parse, normalize, and validate a copied payload off-thread."""
+        source = SubscriptionSource(
+            context.get('unique', ''),
+            context.get('webURL', ''),
+            context.get('remark', ''),
+            context.get('decoderId'),
+        )
+        importer = self.importer
+        filterResult = type(self)._filterImportResult
+        profileFilter = str(context.get('filter', ''))
+        remark = str(context.get('remark', ''))
+
+        def work(isCancelled):
+            """Operate only on captured bytes, values, and plugin data capabilities."""
+            result = importer.importPayload(
+                data,
+                source,
+                requireWorkerSafe=True,
+                isCancelled=isCancelled,
+            )
+
+            return filterResult(result, profileFilter, remark)
+
+        jobId = self._startPreparationJob('import', context, work)
+
+        if jobId is not None:
+            self._preparationPayloads[jobId] = data
+
+    def _runGuiThreadImport(self, data: bytes, context: dict):
+        """Isolate non-opted-in third-party parsing on its required GUI thread."""
+        source = SubscriptionSource(
+            context.get('unique', ''),
+            context.get('webURL', ''),
+            context.get('remark', ''),
+            context.get('decoderId'),
+        )
+
+        try:
+            result = self.importer.importPayload(data, source)
+            result = self._filterImportResult(
+                result,
+                context.get('filter', ''),
+                context.get('remark', ''),
+            )
+        except Exception as ex:
+            # Any non-exit exceptions
+
+            logger.exception(
+                f'failed to prepare subscription {context.get("unique", "")!r}'
+            )
+
+            self._failOperation(context, str(ex) or type(ex).__name__)
+
+            return
+
+        self._handleImportedResult(context, result, 0.0)
+
+    def _handleImportedResult(self, context: dict, result, duration: float):
+        """Capture current group data and dispatch copied reconciliation work."""
+        if not self._isCurrentRequest(context):
+            self._finishOperation(context)
+
+            return
+
+        if result is None or not result.profiles:
+            self._failOperation(context, 'UnsupportedSubscriptionFormat')
+
+            return
+
+        context = {**context, 'decoderId': result.decoderId}
+
+        logger.info(
+            f'prepared subscription ({context.get("remark", "")}, '
+            f'{context.get("unique", "")!r}) with {len(result.profiles)} profiles '
+            f'from {result.decoderId!r}; rejected {result.rejectedItems}; '
+            f'decode/parse {duration:.3f}s'
+        )
+
+        try:
+            snapshot = self.synchronizer.snapshot(
+                Storage.UserServers(),
+                context.get('unique', ''),
+            )
+        except Exception as ex:
+            # Any non-exit exceptions
+
+            self._failOperation(context, str(ex) or type(ex).__name__)
+
+            return
+
+        synchronizer = self.synchronizer
+        incoming = result.profiles
+
+        def work(isCancelled):
+            """Compare copied existing and incoming profiles without live state."""
+            if isCancelled():
+                return None
+
+            return synchronizer.prepare(snapshot, incoming)
+
+        self._startPreparationJob('reconcile', context, work)
+
+    @QtCore.Slot(object)
+    def _handlePreparationOutcome(self, outcome):
+        """Validate one queued worker result and commit only on this Qt thread."""
+        jobId = getattr(outcome, 'jobId', -1)
+
+        self._preparationJobs.pop(jobId, None)
+
+        fallbackPayload = self._preparationPayloads.pop(jobId, None)
+        context = getattr(outcome, 'context', {})
+
+        if self._shuttingDown:
+            return
+
+        if getattr(outcome, 'cancelled', False) or not self._isCurrentRequest(context):
+            logger.debug(
+                f'discard stale/cancelled subscription preparation for '
+                f'{context.get("unique", "")!r}'
+            )
+
+            self._finishOperation(context)
+
+            return
+
+        if getattr(outcome, 'errorType', ''):
+            if (
+                outcome.errorType == SubscriptionWorkerUnsafe.__name__
+                and fallbackPayload is not None
+            ):
+                self._runGuiThreadImport(fallbackPayload, context)
+            else:
+                logger.error(
+                    f'subscription {outcome.stage} preparation failed for '
+                    f'{context.get("unique", "")!r} ({outcome.errorType})'
+                )
+
+                self._failOperation(context, outcome.error)
+
+            return
+
+        if outcome.stage == 'import':
+            self._handleImportedResult(context, outcome.value, outcome.duration)
+
+            return
+
+        if outcome.stage != 'reconcile' or outcome.value is None:
+            self._finishOperation(context)
+
+            return
+
+        started = time.perf_counter()
+
+        try:
+            result = self._synchronizePreparedProfiles(context['unique'], outcome.value)
+        except Exception as ex:
+            # Any non-exit exceptions
+
+            error = str(ex) or type(ex).__name__
+
+            if not self._isCurrentRequest(context):
+                self._finishOperation(context)
+            else:
+                logger.exception(
+                    f'failed to commit subscription {context.get("unique", "")!r}'
+                )
+                self._failOperation(context, error)
+
+            return
+
+        committed = {**context, 'syncResult': result}
+
+        self.subscriptionCommitted.emit(context['unique'])
+        self._recordGroupSuccess(committed, result)
+        self.subscriptionStateChanged.emit((context['unique'],))
+        self._finishOperation(committed, successful=committed, structural=True)
+
+        logger.info(
+            f'committed subscription {context["unique"]!r}; reconciliation '
+            f'{outcome.duration:.3f}s, GUI commit {time.perf_counter() - started:.3f}s'
+        )
+
+    def _failOperation(self, context: dict, error: str):
+        """Finalize one current logical operation as an isolated failure."""
+        failed = {**context, 'error': error}
+
+        if self._isCurrentRequest(context):
+            self._recordGroupFailure(failed)
+            self.subscriptionStateChanged.emit((context.get('unique', ''),))
+
+        self._finishOperation(context, failed=failed)
+
+    def _finishOperation(
+        self,
+        context: dict,
+        *,
+        successful=None,
+        failed=None,
+        structural: bool = False,
+    ):
+        """Complete one batch member and publish one coalesced batch outcome."""
+        batchId = context.get('batchId')
+        state = self._batches.get(batchId)
+
+        if state is None or self._shuttingDown:
+            return
+
+        token = (context.get('unique', ''), context.get('requestVersion'))
+
+        if token not in state.pending:
+            return
+
+        state.pending.remove(token)
+
+        if successful is not None:
+            state.successful.append(successful)
+        if failed is not None and self._isCurrentRequest(context):
+            state.failed.append(failed)
+
+        state.structural = state.structural or structural
+
+        if state.pending:
+            return
+
+        self._batches.pop(batchId, None)
+
+        try:
+            Storage.persistSubscriptionGroups()
+        except Exception:
+            # Any non-exit exceptions
+
+            logger.exception('failed to persist completed subscription batch')
+
+        if state.structural:
+            self.subscriptionsChanged.emit()
+
+        if state.successful or state.failed:
+            self.updateCompleted.emit(
+                SubscriptionUpdateBatch(
+                    tuple(state.successful),
+                    tuple(state.failed),
+                    state.showMessageBox,
+                )
+            )
+
     def _synchronizeProfiles(self, unique: str, profiles):
         """Reconcile one group and apply connection effects without UI ownership."""
         servers = Storage.UserServers()
@@ -371,6 +748,59 @@ class SubscriptionManager(HttpGetManager):
 
         return result
 
+    def _synchronizePreparedProfiles(self, unique: str, plan):
+        """Commit one accepted worker plan and apply GUI-owned connection effects."""
+        servers = Storage.UserServers()
+        activatedIndex = Storage.UserActivatedItemIndex()
+
+        activeProfileId = ''
+        activeWasManagedByGroup = False
+
+        if 0 <= activatedIndex < len(servers):
+            active = servers[activatedIndex]
+            activeProfileId = active.metadata.profileId
+            activeWasManagedByGroup = (
+                active.itemSubscription == unique and active.itemSubscriptionManaged
+            )
+
+        controller = AppConnectionController()
+        wasConnected = controller is not None and controller.isConnected()
+        result = self.synchronizer.commit(servers, plan)
+        newActivatedIndex = next(
+            (
+                index
+                for index, profile in enumerate(servers)
+                if profile.metadata.profileId == activeProfileId
+            ),
+            -1,
+        )
+
+        try:
+            AppSettings.set('ActivatedItemIndex', str(newActivatedIndex))
+        except Exception:
+            # Any non-exit exceptions
+
+            logger.exception(
+                'failed to persist the active profile index after '
+                f'synchronizing subscription {unique!r}'
+            )
+
+        try:
+            if wasConnected and activeProfileId:
+                if newActivatedIndex < 0 and activeWasManagedByGroup:
+                    controller.startDisconnection()
+                elif activeProfileId in result.changedProfileIds:
+                    controller.startReconnection()
+        except Exception:
+            # Any non-exit exceptions
+
+            logger.exception(
+                f'failed to apply connection effects after synchronizing '
+                f'subscription {unique!r}'
+            )
+
+        return result
+
     @staticmethod
     def _recordGroupFailure(param):
         """Best-effort persist one current request's terminal failure state."""
@@ -507,6 +937,9 @@ class SubscriptionManager(HttpGetManager):
 
     def completionCallback(self, **kwargs):
         """Complete a batch after its final reply finishes."""
+        if 'batchId' in kwargs:
+            return
+
         depthMap = kwargs.get('depthMap', {})
         depthMap['depth'] -= 1
 
@@ -516,6 +949,20 @@ class SubscriptionManager(HttpGetManager):
     def successCallback(self, networkReply, **kwargs):
         """Decode one successful subscription response."""
         if not self._isCurrentRequest(kwargs):
+            self._finishOperation(kwargs)
+
+            return
+
+        if 'batchId' in kwargs:
+            data = bytes(networkReply.readAll().data())
+            decoderId = kwargs.get('decoderId') or kwargs.get('lastDecoderId')
+            context = {**kwargs, 'decoderId': decoderId}
+
+            if self.importer.registry.subscriptionDecoderWorkerSafe(decoderId):
+                self._startImportPreparation(data, context)
+            else:
+                self._runGuiThreadImport(data, context)
+
             return
 
         unique = kwargs.get('unique', '')
@@ -573,6 +1020,8 @@ class SubscriptionManager(HttpGetManager):
     def failureCallback(self, networkReply, **kwargs):
         """Record one failed subscription response."""
         if not self._isCurrentRequest(kwargs):
+            self._finishOperation(kwargs)
+
             return
 
         unique = kwargs.get('unique', '')
@@ -583,6 +1032,11 @@ class SubscriptionManager(HttpGetManager):
 
         logger.error(f'update subscription ({remark}, {unique!r}) failed: {error}')
 
+        if 'batchId' in kwargs:
+            self._failOperation(kwargs, error)
+
+            return
+
         failureArgs.append({'error': error, **kwargs})
 
     def updateSubsByWebGET(self, **kwargs):
@@ -631,8 +1085,14 @@ class SubscriptionManager(HttpGetManager):
             return
 
         changedSubscriptions = []
+        groups = []
+        operations = []
 
-        for unique, _subscription in batch:
+        self._nextBatchId += 1
+
+        batchId = self._nextBatchId
+
+        for unique, subscription in batch:
             group = Storage.SubscriptionGroup(unique)
 
             if group is None:
@@ -641,26 +1101,57 @@ class SubscriptionManager(HttpGetManager):
             group.lastSyncStatus = 'syncing'
             group.lastSyncError = ''
 
-            Storage.upsertSubscriptionGroup(group)
+            self.cancelUpdates(unique)
+
+            version = self._requestVersions[unique]
+            context = {
+                'unique': unique,
+                'remark': subscription.get('remark', ''),
+                'webURL': subscription.get('webURL', ''),
+                'userAgent': subscription.get('userAgent', ''),
+                'filter': subscription.get('filter', ''),
+                'decoderId': subscription.get('lastDecoderId') or None,
+                'lastDecoderId': subscription.get('lastDecoderId', ''),
+                'batchId': batchId,
+                'requestVersion': version,
+                'requestSignature': self._requestSignature(subscription),
+            }
+
+            groups.append(group)
+            operations.append(context)
             changedSubscriptions.append(unique)
 
+        if not operations:
+            return
+
+        Storage.upsertSubscriptionGroups(groups)
+        Storage.persistSubscriptionGroups()
+
+        self._batches[batchId] = _SubscriptionBatchState(
+            {(context['unique'], context['requestVersion']) for context in operations},
+            bool(kwargs.get('showMessageBox', True)),
+            [],
+            [],
+        )
+
         if changedSubscriptions:
             self.subscriptionStateChanged.emit(tuple(changedSubscriptions))
 
-        depthMap = {'depth': len(batch)}
-        successArgs = []
-        failureArgs = []
+        for context in operations:
+            try:
+                self.updateSubsByWebGET(
+                    **context,
+                    logActionMessage=bool(kwargs.get('logActionMessage', False)),
+                )
+            except Exception as ex:
+                # Any non-exit exceptions
 
-        for unique, subscription in batch:
-            self.updateSubsByWebGET(
-                unique=unique,
-                **subscription,
-                **kwargs,
-                depthMap=depthMap,
-                successArgs=successArgs,
-                failureArgs=failureArgs,
-                requestVersion=self._nextRequestVersion(unique),
-            )
+                logger.exception(
+                    f'failed to start subscription request for '
+                    f'{context.get("unique", "")!r}'
+                )
+
+                self._failOperation(context, str(ex) or type(ex).__name__)
 
     def updateSubsByUnique(self, unique: str, **kwargs):
         """Update one enabled subscription through the canonical batch path."""

Furious/Service/SubscriptionPreparation.py

@@ -0,0 +1,109 @@
+# Copyright (C) 2024–present  Loren Eteval & contributors <loren.eteval@proton.me>
+#
+# 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 <https://www.gnu.org/licenses/>.
+
+"""Run copied subscription preparation work in a bounded Qt thread pool."""
+
+from __future__ import annotations
+
+from PySide6 import QtCore
+
+from dataclasses import dataclass
+
+import threading
+import time
+
+__all__ = [
+    'SubscriptionPreparationJob',
+    'SubscriptionPreparationOutcome',
+    'SubscriptionPreparationRelay',
+]
+
+
+@dataclass(frozen=True)
+class SubscriptionPreparationOutcome:
+    """Return plain worker data to the subscription manager's Qt thread."""
+
+    jobId: int
+    stage: str
+    context: dict
+    value: object = None
+    errorType: str = ''
+    error: str = ''
+    cancelled: bool = False
+    duration: float = 0.0
+    workerThreadId: int = 0
+
+
+class SubscriptionPreparationRelay(QtCore.QObject):
+    """Provide one process-lifetime queued result path for all pool jobs."""
+
+    completed = QtCore.Signal(object)
+
+
+class SubscriptionPreparationJob(QtCore.QRunnable):
+    """Own one cancellable plain-data preparation callable."""
+
+    def __init__(self, jobId: int, stage: str, context: dict, work, relay):
+        """Capture copied operation context and a non-Qt work callable."""
+        super().__init__()
+
+        self.jobId = jobId
+        self.stage = stage
+        self.context = dict(context)
+        self.work = work
+        self.relay = relay
+        self.cancelled = threading.Event()
+
+        self.setAutoDelete(True)
+
+    def cancel(self):
+        """Make queued/running work stop cooperatively and reject its result."""
+        self.cancelled.set()
+
+    def run(self):
+        """Execute without touching QObject, widget, model, or live repository state."""
+        started = time.perf_counter()
+        value = None
+        errorType = ''
+        error = ''
+
+        try:
+            if not self.cancelled.is_set():
+                value = self.work(self.cancelled.is_set)
+        except Exception as ex:
+            # Any non-exit exceptions
+
+            errorType = type(ex).__name__
+            error = str(ex) or errorType
+
+        try:
+            self.relay.completed.emit(
+                SubscriptionPreparationOutcome(
+                    self.jobId,
+                    self.stage,
+                    self.context,
+                    value,
+                    errorType,
+                    error,
+                    self.cancelled.is_set(),
+                    time.perf_counter() - started,
+                    threading.get_ident(),
+                )
+            )
+        except RuntimeError:
+            # The owning manager may already be gone after an abnormal teardown.
+            pass

Furious/Service/SubscriptionSync.py

@@ -25,7 +25,12 @@ import copy
 
 from dataclasses import dataclass
 
-__all__ = ['SubscriptionSyncResult', 'SubscriptionSynchronizer']
+__all__ = [
+    'SubscriptionSyncPlan',
+    'SubscriptionSyncResult',
+    'SubscriptionSyncSnapshot',
+    'SubscriptionSynchronizer',
+]
 
 
 @dataclass(frozen=True)
@@ -41,6 +46,25 @@ class SubscriptionSyncResult:
     changedProfileIds: tuple[str, ...]
 
 
+@dataclass(frozen=True)
+class SubscriptionSyncSnapshot:
+    """Hold copied group profiles and the live source revision they represent."""
+
+    groupId: str
+    profiles: tuple[ServerProfile, ...]
+    sourceRevision: tuple[tuple[str, str, str], ...]
+
+
+@dataclass(frozen=True)
+class SubscriptionSyncPlan:
+    """Describe a worker-prepared replacement for one subscription group."""
+
+    groupId: str
+    sourceRevision: tuple[tuple[str, str, str], ...]
+    profiles: tuple[ServerProfile, ...]
+    result: SubscriptionSyncResult
+
+
 class SubscriptionSynchronizer:
     """Own stable, group-scoped profile reconciliation semantics."""
 
@@ -53,6 +77,112 @@ class SubscriptionSynchronizer:
         'tags',
     )
 
+    @staticmethod
+    def _sourceRevision(profiles, groupId: str):
+        """Fingerprint the relevant live group state without using row identity."""
+        return tuple(
+            (
+                profile.metadata.profileId,
+                profile.metadata.subscriptionProfileKey,
+                profileConnectionFingerprint(profile),
+            )
+            for profile in profiles
+            if profile.itemSubscription == groupId and profile.itemSubscriptionManaged
+        )
+
+    def snapshot(self, profiles, groupId: str) -> SubscriptionSyncSnapshot:
+        """Copy only the live profiles owned by *groupId* for worker preparation."""
+        if not groupId:
+            raise ValueError('subscription group ID must not be empty')
+
+        owned = tuple(
+            copy.deepcopy(profile)
+            for profile in profiles
+            if profile.itemSubscription == groupId and profile.itemSubscriptionManaged
+        )
+
+        return SubscriptionSyncSnapshot(
+            groupId,
+            owned,
+            self._sourceRevision(profiles, groupId),
+        )
+
+    def prepare(
+        self,
+        snapshot: SubscriptionSyncSnapshot,
+        incomingProfiles,
+    ) -> SubscriptionSyncPlan:
+        """Build a reconciliation plan using copied profiles only."""
+        working = list(snapshot.profiles)
+        result = self.reconcile(working, incomingProfiles, snapshot.groupId)
+
+        return SubscriptionSyncPlan(
+            snapshot.groupId,
+            snapshot.sourceRevision,
+            tuple(working),
+            result,
+        )
+
+    def commit(self, profiles, plan: SubscriptionSyncPlan) -> SubscriptionSyncResult:
+        """Atomically apply a current worker plan while preserving live identities."""
+        groupId = plan.groupId
+
+        if self._sourceRevision(profiles, groupId) != plan.sourceRevision:
+            raise RuntimeError('subscription profile source changed during preparation')
+
+        managedIndexes = [
+            index
+            for index, profile in enumerate(profiles)
+            if profile.itemSubscription == groupId and profile.itemSubscriptionManaged
+        ]
+        insertionIndex = min(managedIndexes) if managedIndexes else len(profiles)
+        existingById = {
+            profiles[index].metadata.profileId: profiles[index]
+            for index in managedIndexes
+        }
+        synchronized = []
+
+        for prepared in plan.profiles:
+            existing = existingById.pop(prepared.metadata.profileId, None)
+
+            if existing is None:
+                synchronized.append(prepared)
+
+                continue
+
+            metadata = copy.deepcopy(prepared.metadata)
+
+            for fieldName in self.LocalMetadataFields:
+                setattr(metadata, fieldName, getattr(existing.metadata, fieldName))
+
+            existing.connection = prepared.connection
+            existing.metadata = metadata
+            synchronized.append(existing)
+
+        for removed in existingById.values():
+            removed.deleted = True
+
+        unmanagedOrOther = [
+            profile
+            for profile in profiles
+            if not (
+                profile.itemSubscription == groupId and profile.itemSubscriptionManaged
+            )
+        ]
+        finalProfiles = (
+            unmanagedOrOther[:insertionIndex]
+            + synchronized
+            + unmanagedOrOther[insertionIndex:]
+        )
+
+        profiles[:] = finalProfiles
+
+        for index, profile in enumerate(finalProfiles):
+            profile.index = index
+            profile.deleted = False
+
+        return plan.result
+
     @staticmethod
     def _keyAssignments(profiles: list[ServerProfile], groupId: str):
         """Plan deterministic keys for legacy profiles without mutating them."""

Furious/Widget/ServerTableView.py

@@ -1917,6 +1917,7 @@ class ServerTableView(
 
     def cleanup(self):
         """Release resources owned by the user servers Qt table view."""
+        self.subsManager.shutdown()
         self.profileTestManager.shutdown()
         self._clearSubscriptionActions()
 

Furious/Widget/SubscriptionTableView.py

@@ -728,18 +728,32 @@ class SubscriptionTableView(Mixins.QTranslatable, AppQTableView):
             self.ItemKey.index('profiles'),
         )
 
-        for unique in dict.fromkeys(uniques):
-            row = rows.get(unique)
+        changedRows = sorted(
+            row
+            for unique in dict.fromkeys(uniques)
+            if (row := rows.get(unique)) is not None
+        )
+
+        if not changedRows:
+            return
+
+        firstRow = previousRow = changedRows[0]
+
+        for row in (*changedRows[1:], None):
+            if row is not None and row == previousRow + 1:
+                previousRow = row
 
-            if row is None:
                 continue
 
             self.sourceModel.dataChanged.emit(
-                self.sourceModel.index(row, firstColumn),
-                self.sourceModel.index(row, lastColumn),
+                self.sourceModel.index(firstRow, firstColumn),
+                self.sourceModel.index(previousRow, lastColumn),
                 [],
             )
 
+            if row is not None:
+                firstRow = previousRow = row
+
     def flushItem(self, row, column, item):
         """Refresh item."""
         if row < 0 or row >= self.sourceModel.rowCount():
@@ -806,6 +820,9 @@ class SubscriptionTableView(Mixins.QTranslatable, AppQTableView):
         subsob = {unique: group.toMapping()}
 
         if unique in Storage.UserSubs():
+            if self.subsManager is not None:
+                self.subsManager.cancelUpdates(unique)
+
             row = list(Storage.UserSubs().keys()).index(unique)
 
             Storage.upsertSubscriptionGroup(group)