fix(plugin): reload Zipline service on same-version replace

- Add `generation` counter to serialized state so DataStore emits
  on every addPlugin (same-version replace produces identical JSON)
- Run all Zipline JNI calls (close, service.close) on ziplineDispatcher
  thread — wrong thread causes segfaults
- Stop ALL services in flatMapLatest finally block instead of only
  deselected ones (leaked zombie services)
- Wrap finally block in NonCancellable + try-catch per service
This commit is contained in:
Kingkor Roy Tirtho 2026-07-07 19:06:44 +06:00
parent f05f0727b6
commit f012defd79
3 changed files with 50 additions and 13 deletions

View File

@ -125,6 +125,7 @@ open class ZiplinePluginService(
} }
private val lifecycleMutex = Mutex() private val lifecycleMutex = Mutex()
private var ziplineLoader: ZiplineLoader private var ziplineLoader: ZiplineLoader
private var ziplineInstance: Zipline? = null
private val serviceRegistry = mutableMapOf<KClass<*>, ZiplineService>() private val serviceRegistry = mutableMapOf<KClass<*>, ZiplineService>()
init { init {
@ -271,6 +272,7 @@ open class ZiplinePluginService(
return return
} }
logger.d { "[$applicationName] start(): loading plugin from $manifestUrl" }
withContext(ziplineDispatcher.dispatcher) { withContext(ziplineDispatcher.dispatcher) {
trace("start(): inside zipline dispatcher before loadOnce") trace("start(): inside zipline dispatcher before loadOnce")
val result = ziplineLoader.loadOnce( val result = ziplineLoader.loadOnce(
@ -280,6 +282,8 @@ open class ZiplinePluginService(
) )
when (result) { when (result) {
is LoadResult.Success -> { is LoadResult.Success -> {
logger.d { "[$applicationName] start(): loadOnce succeeded, consuming services" }
ziplineInstance = result.zipline
// Now we consume the initializer // Now we consume the initializer
val initializer = result.zipline.take<Initializer>(Initializer_SERVICE_NAME) val initializer = result.zipline.take<Initializer>(Initializer_SERVICE_NAME)
// Bind host services before initialization, so plugins can use them in their initializer // Bind host services before initialization, so plugins can use them in their initializer
@ -310,10 +314,19 @@ open class ZiplinePluginService(
override suspend fun stop() { override suspend fun stop() {
lifecycleMutex.withLock { lifecycleMutex.withLock {
trace("stop(): entered") trace("stop(): entered")
withContext(ziplineDispatcher.dispatcher) {
ziplineInstance?.close()
ziplineInstance = null
for (service in serviceRegistry.values) { for (service in serviceRegistry.values) {
try {
trace("stop(): closing service ${service::class.simpleName}") trace("stop(): closing service ${service::class.simpleName}")
service.close() service.close()
} catch (_: Exception) {
trace("stop(): error closing service ${service::class.simpleName}")
} }
}
}
serviceRegistry.clear()
scope.cancel() scope.cancel()
loggedInStateFlow.value = false loggedInStateFlow.value = false
ziplineDispatcher.close() ziplineDispatcher.close()

View File

@ -34,6 +34,7 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.IO import kotlinx.coroutines.IO
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async import kotlinx.coroutines.async
import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.awaitCancellation
@ -147,9 +148,13 @@ class PluginManager(
@OptIn(ExperimentalCoroutinesApi::class) @OptIn(ExperimentalCoroutinesApi::class)
val ziplineServices = state val ziplineServices = state
.filterIsInstance<PluginManagerStates.Data>() .filterIsInstance<PluginManagerStates.Data>()
.map { it.selectedPlugins } .map { it.selectedPlugins to it.generation }
// Paired with generation so same-version replaces (structurally equal
// PluginEntry) still trigger a restart. Without it the DataStore skips
// the emission (identical JSON) and the flow stays alive with stale code.
.distinctUntilChanged() .distinctUntilChanged()
.flatMapLatest { plugins -> .flatMapLatest { (plugins, gen) ->
logger.d { "ziplineServices: flow restarting with generation=$gen, selectedPlugins=${plugins.keys}" }
flow { flow {
val ziplineServices: MutableMap<PluginAbility, PluginService?> = val ziplineServices: MutableMap<PluginAbility, PluginService?> =
mutableMapOf() mutableMapOf()
@ -166,6 +171,7 @@ class PluginManager(
ziplineServicesByPluginID[plugin.id] = service ziplineServicesByPluginID[plugin.id] = service
service.start() service.start()
} else { } else {
logger.d { "ziplineServices: creating new ZiplinePluginService for ${plugin.name} (${plugin.id})" }
val service = ZiplinePluginService( val service = ZiplinePluginService(
applicationName = plugin.name, applicationName = plugin.name,
manifestUrl = "http://localhost?path=${(pluginsDir / plugin.id.toPath() / "manifest.zipline.json")}", manifestUrl = "http://localhost?path=${(pluginsDir / plugin.id.toPath() / "manifest.zipline.json")}",
@ -180,13 +186,17 @@ class PluginManager(
// Keep the flow alive until the next plugin is selected. // Keep the flow alive until the next plugin is selected.
awaitCancellation() awaitCancellation()
} finally { } finally {
// Stop services that are no longer selected withContext(NonCancellable) {
val selectedPluginIDs = plugins.values.map { it.id }.toSet() logger.d { "ziplineServices: finally block stopping ${ziplineServicesByPluginID.size} services" }
ziplineServicesByPluginID.forEach { (pluginID, service) -> ziplineServicesByPluginID.forEach { (_, service) ->
if (!selectedPluginIDs.contains(pluginID)) { try {
service.stop() service.stop()
} catch (_: Exception) {
logger.e { "ziplineServices: error stopping service" }
} }
} }
logger.d { "ziplineServices: all services stopped" }
}
} }
} }
} }
@ -495,7 +505,12 @@ class PluginManager(
currentState.selectedPlugins.mapValues { (_, selectedPlugin) -> currentState.selectedPlugins.mapValues { (_, selectedPlugin) ->
if (selectedPlugin.id == plugin.id) plugin else selectedPlugin if (selectedPlugin.id == plugin.id) plugin else selectedPlugin
} }
val newState = PluginManagerStates.Data(updatedPlugins, updatedSelectedPlugins) val newState = PluginManagerStates.Data(
plugins = updatedPlugins,
selectedPlugins = updatedSelectedPlugins,
generation = currentState.generation + 1
)
logger.d { "addPlugin: bumped generation to ${newState.generation} for plugin ${plugin.id}" }
updatePluginsState(newState) updatePluginsState(newState)
} }
} }
@ -510,7 +525,11 @@ class PluginManager(
val updatedPlugins = currentState.plugins.filterNot { it.id == plugin.id } val updatedPlugins = currentState.plugins.filterNot { it.id == plugin.id }
val updatedSelectedPlugins = val updatedSelectedPlugins =
currentState.selectedPlugins.filterValues { it.id != plugin.id } currentState.selectedPlugins.filterValues { it.id != plugin.id }
val newState = PluginManagerStates.Data(updatedPlugins, updatedSelectedPlugins) val newState = PluginManagerStates.Data(
plugins = updatedPlugins,
selectedPlugins = updatedSelectedPlugins,
generation = currentState.generation
)
withContext(Dispatchers.IO) { withContext(Dispatchers.IO) {
val pluginDir = pluginsDir / plugin.id.toPath() val pluginDir = pluginsDir / plugin.id.toPath()

View File

@ -55,7 +55,12 @@ sealed class PluginManagerStates {
@Serializable @Serializable
data class Data( data class Data(
val plugins: List<PluginEntry>, val plugins: List<PluginEntry>,
val selectedPlugins: Map<PluginAbility, PluginEntry> = emptyMap() val selectedPlugins: Map<PluginAbility, PluginEntry> = emptyMap(),
// Bumped on every addPlugin so the DataStore always detects a structural
// change — without it, same-version replaces produce identical JSON and
// the DataStore's internal distinctUntilChanged blocks the state emission,
// leaving the ziplineServices flow stuck with the old code.
val generation: Long = 0L
) : PluginManagerStates() ) : PluginManagerStates()
data object Loading : PluginManagerStates() data object Loading : PluginManagerStates()