From e7318e3b1f3eabe5e8b5dc68f3470acd777e6a2d Mon Sep 17 00:00:00 2001 From: Kostya Ripak Date: Tue, 30 Jun 2026 16:32:35 +0000 Subject: [PATCH] [plugin management] IJPL-248432 Fix flaky tests for PluginUpdateService - Use MutableSharedFlow for plugin updates PluginDto is mutated in-place when merging updates, so MutableStateFlow conflates all updates into one. Fixing PluginDto is costly, so for now just make sure that every update is consumed. - Guard against exceptions in PluginUpdatesService If external code throws an Exception, it won't cancel the coroutine - Dispatch callbacks in a dedicated coroutine Hopping to UI dispatcher may park plugin update collector on EDT thread and lead to unwanted delays. Merge-request: IJ-MR-210733 Merged-by: Konstantin Ripak GitOrigin-RevId: 958d2c19913d16d4b1485d3cd7dd2cbbbd271f86 --- .../newui/DefaultPluginUpdatesProvider.kt | 21 ++++++--- .../ide/plugins/newui/PluginUpdatesService.kt | 43 ++++++++++++------- 2 files changed, 42 insertions(+), 22 deletions(-) diff --git a/platform/platform-impl/src/com/intellij/ide/plugins/newui/DefaultPluginUpdatesProvider.kt b/platform/platform-impl/src/com/intellij/ide/plugins/newui/DefaultPluginUpdatesProvider.kt index 7fe045fb917d..8aa9203d75ea 100644 --- a/platform/platform-impl/src/com/intellij/ide/plugins/newui/DefaultPluginUpdatesProvider.kt +++ b/platform/platform-impl/src/com/intellij/ide/plugins/newui/DefaultPluginUpdatesProvider.kt @@ -7,6 +7,8 @@ import com.intellij.ide.plugins.PluginEnableStateChangedListener import com.intellij.ide.plugins.PluginStateListener import com.intellij.ide.plugins.PluginStateManager import com.intellij.ide.plugins.api.PluginDto +import com.intellij.openapi.diagnostic.getOrHandleException +import com.intellij.openapi.diagnostic.logger import com.intellij.openapi.extensions.PluginId import com.intellij.openapi.updateSettings.impl.PluginUpdateHandler import kotlinx.coroutines.CoroutineScope @@ -23,6 +25,7 @@ import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import org.jetbrains.annotations.ApiStatus +import kotlin.coroutines.cancellation.CancellationException import kotlin.time.Duration.Companion.milliseconds @OptIn(FlowPreview::class) @@ -33,6 +36,10 @@ class DefaultPluginUpdatesProvider(private val coroutineScope: CoroutineScope) : private val updateRequestFlow = MutableSharedFlow(replay = 1, extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST) private var lastPluginUpdates: PluginUpdatesEvent? = null + companion object { + private val LOG = logger() + } + init { PluginStateManager.addStateListener(object : PluginStateListener { override fun install(descriptor: IdeaPluginDescriptor) { @@ -64,12 +71,14 @@ class DefaultPluginUpdatesProvider(private val coroutineScope: CoroutineScope) : .debounce(300.milliseconds) .collectLatest { updateMutex.withLock { - val model = (PluginUpdateHandler.getInstance().loadAndStorePluginUpdates(null)) - val pluginUpdates = PluginUpdatesEvent(model.pluginUpdates.markLocal(), - model.disabledPluginUpdates.markLocal(), - model.updatesFromCustomRepositories.markLocal()) - lastPluginUpdates = pluginUpdates - emitUpdates(pluginUpdates) + runCatching { + val model = (PluginUpdateHandler.getInstance().loadAndStorePluginUpdates(null)) + val pluginUpdates = PluginUpdatesEvent(model.pluginUpdates.markLocal(), + model.disabledPluginUpdates.markLocal(), + model.updatesFromCustomRepositories.markLocal()) + lastPluginUpdates = pluginUpdates + emitUpdates(pluginUpdates) + }.getOrHandleException { e -> LOG.warn("Failed to load plugin updates:", e) } } } } diff --git a/platform/platform-impl/src/com/intellij/ide/plugins/newui/PluginUpdatesService.kt b/platform/platform-impl/src/com/intellij/ide/plugins/newui/PluginUpdatesService.kt index 81df1e5d56f7..c89aff2bb4f4 100644 --- a/platform/platform-impl/src/com/intellij/ide/plugins/newui/PluginUpdatesService.kt +++ b/platform/platform-impl/src/com/intellij/ide/plugins/newui/PluginUpdatesService.kt @@ -7,10 +7,13 @@ import com.intellij.openapi.application.UI import com.intellij.openapi.application.asContextElement import com.intellij.openapi.components.Service import com.intellij.openapi.components.service +import com.intellij.openapi.diagnostic.getOrHandleException import com.intellij.openapi.diagnostic.logger +import com.intellij.openapi.diagnostic.rethrowControlFlowException import com.intellij.openapi.extensions.PluginId import com.intellij.openapi.progress.runBlockingMaybeCancellable import com.intellij.util.concurrency.annotations.RequiresEdt +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.FlowPreview @@ -23,15 +26,16 @@ import kotlinx.coroutines.flow.catch import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock -import kotlinx.coroutines.withContext import org.jetbrains.annotations.ApiStatus import org.jetbrains.annotations.VisibleForTesting import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CopyOnWriteArrayList import java.util.function.Consumer +import kotlin.concurrent.Volatile import kotlin.concurrent.atomics.ExperimentalAtomicApi import kotlin.time.Duration.Companion.milliseconds @@ -48,7 +52,8 @@ typealias PluginUpdateCallback = Consumer class PluginUpdatesService(val coroutineScope: CoroutineScope) { private val myCallbacks = CopyOnWriteArrayList() - private val pluginUpdateFlow = MutableStateFlow(null) + private val pluginUpdateFlow = MutableSharedFlow(replay = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST) + @Volatile private var lastUpdates: PluginUpdatesEvent? = null private val updateIdsFlow = MutableStateFlow?>(null) private val updateRequestFlow = MutableSharedFlow(replay = 1, extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST) private val providerSnapshots = ConcurrentHashMap() @@ -74,10 +79,11 @@ class PluginUpdatesService(val coroutineScope: CoroutineScope) { init { startUpdateCollection() startUpdateTrigger() + startDispatchingCallbacks() } private suspend fun ensureUpdatesStarted() { - if (pluginUpdateFlow.value == null) { + if (lastUpdates == null) { triggerUpdates() } } @@ -98,22 +104,24 @@ class PluginUpdatesService(val coroutineScope: CoroutineScope) { } private suspend fun onProviderUpdated() { - val merged = updateMutex.withLock { + updateMutex.withLock { val merged = mergeUpdates(providerSnapshots.values) - pluginUpdateFlow.value = merged + lastUpdates = merged + pluginUpdateFlow.emit(merged) updateIdsFlow.value = merged.all.mapTo(HashSet()) { plugin -> plugin.pluginId } - merged - } - withContext(Dispatchers.UI + ModalityState.any().asContextElement()) { - dispatchCallbacks(merged) } } + private fun startDispatchingCallbacks() = coroutineScope.launch(Dispatchers.UI + ModalityState.any().asContextElement()) { + pluginUpdateFlow + .collect { dispatchCallbacks(it) } + } + private fun startUpdateTrigger() = coroutineScope.launch { - updateRequestFlow - .debounce(300.milliseconds) - .collect { - PluginUpdatesProvider.getInstances().forEach { it.update() } + updateRequestFlow.debounce(300.milliseconds).collect { + PluginUpdatesProvider.getInstances().forEach { + runCatching { it.update() }.getOrHandleException { e -> LOG.warn("PluginUpdatesProvider.update() failed:", e) } + } } } @@ -149,7 +157,7 @@ class PluginUpdatesService(val coroutineScope: CoroutineScope) { suspend fun awaitUpdates(): Collection { ensureUpdatesStarted() - return pluginUpdateFlow.filterNotNull().first().all + return pluginUpdateFlow.first().all } @VisibleForTesting @@ -169,11 +177,14 @@ class PluginUpdatesService(val coroutineScope: CoroutineScope) { } private fun getLastUpdates(): PluginUpdatesEvent? { - return pluginUpdateFlow.value + return lastUpdates } private fun dispatchCallbacks(updates: PluginUpdatesEvent) { - myCallbacks.forEach { it.accept(updates) } + myCallbacks.forEach { + runCatching { it.accept(updates) } + .getOrHandleException { e -> LOG.warn("PluginUpdatesEvent callback ${it.javaClass.name} failed:", e) } + } } @VisibleForTesting