diff --git a/Furious/Backends/ExternalCore/Protocols.py b/Furious/Backends/ExternalCore/Protocols.py index 677a3c6..b5e9686 100644 --- a/Furious/Backends/ExternalCore/Protocols.py +++ b/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', diff --git a/Furious/Backends/Hysteria1/Protocols.py b/Furious/Backends/Hysteria1/Protocols.py index 3b2f427..8f675fc 100644 --- a/Furious/Backends/Hysteria1/Protocols.py +++ b/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', diff --git a/Furious/Backends/Hysteria2/Protocols.py b/Furious/Backends/Hysteria2/Protocols.py index 7677a15..c9ab204 100644 --- a/Furious/Backends/Hysteria2/Protocols.py +++ b/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', diff --git a/Furious/Backends/Xray/Protocols.py b/Furious/Backends/Xray/Protocols.py index 3815096..8310570 100644 --- a/Furious/Backends/Xray/Protocols.py +++ b/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, diff --git a/Furious/Extensions/StandardSubscriptions.py b/Furious/Extensions/StandardSubscriptions.py index e7ff0ae..c0eb47d 100644 --- a/Furious/Extensions/StandardSubscriptions.py +++ b/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 diff --git a/Furious/Plugins/AGENTS.md b/Furious/Plugins/AGENTS.md index cab4491..e337fa1 100644 --- a/Furious/Plugins/AGENTS.md +++ b/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 diff --git a/Furious/Plugins/API.py b/Furious/Plugins/API.py index c1064dd..30b6df3 100644 --- a/Furious/Plugins/API.py +++ b/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: diff --git a/Furious/Plugins/Registry.py b/Furious/Plugins/Registry.py index 1a90a5d..b053c5d 100644 --- a/Furious/Plugins/Registry.py +++ b/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) diff --git a/Furious/Repository/Storage.py b/Furious/Repository/Storage.py index 040817c..8fff29a 100644 --- a/Furious/Repository/Storage.py +++ b/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.""" diff --git a/Furious/Repository/Subscriptions.py b/Furious/Repository/Subscriptions.py index 07678ff..cb9ae0f 100644 --- a/Furious/Repository/Subscriptions.py +++ b/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.""" diff --git a/Furious/Service/AGENTS.md b/Furious/Service/AGENTS.md index c88c06a..677ea1d 100644 --- a/Furious/Service/AGENTS.md +++ b/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. diff --git a/Furious/Service/SubscriptionImporter.py b/Furious/Service/SubscriptionImporter.py index e1dbcd3..4cd266f 100644 --- a/Furious/Service/SubscriptionImporter.py +++ b/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), diff --git a/Furious/Service/SubscriptionManager.py b/Furious/Service/SubscriptionManager.py index 1b3aab3..c969d49 100644 --- a/Furious/Service/SubscriptionManager.py +++ b/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.""" diff --git a/Furious/Service/SubscriptionPreparation.py b/Furious/Service/SubscriptionPreparation.py new file mode 100644 index 0000000..ad28f73 --- /dev/null +++ b/Furious/Service/SubscriptionPreparation.py @@ -0,0 +1,109 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# This file is part of Furious. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""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 diff --git a/Furious/Service/SubscriptionSync.py b/Furious/Service/SubscriptionSync.py index 4a49e4c..0170ef0 100644 --- a/Furious/Service/SubscriptionSync.py +++ b/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.""" diff --git a/Furious/Widget/ServerTableView.py b/Furious/Widget/ServerTableView.py index b1cfafd..e12ba1d 100644 --- a/Furious/Widget/ServerTableView.py +++ b/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() diff --git a/Furious/Widget/SubscriptionTableView.py b/Furious/Widget/SubscriptionTableView.py index 5e156fe..715d62a 100644 --- a/Furious/Widget/SubscriptionTableView.py +++ b/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)