Refactor plugin start task management to ensure cancellation of existing tasks (#8170)

Author: JohnNiang

application/src/main/java/run/halo/app/core/reconciler/PluginReconciler.java

@@ -337,6 +337,7 @@ private Result enablePlugin(Plugin plugin) {
                 .lastTransitionTime(clock.instant())
                 .build());
             status.setPhase(Plugin.Phase.FAILED);
+            removeStartTaskIfPresent(pluginName);
             return Result.doNotRetry();
         }
         var pluginState = current.getPluginState();
@@ -351,8 +352,9 @@ private Result enablePlugin(Plugin plugin) {
                 .lastTransitionTime(clock.instant())
                 .build());
             status.setPhase(Plugin.Phase.STARTED);
+            removeStartTaskIfPresent(pluginName);
             requestToReloadPluginsOptionallyDependentOn(pluginName);
-            return null;
+            return Result.doNotRetry();
         }
         if (pluginState.isFailed()) {
             var t = current.getFailedException();
@@ -367,22 +369,30 @@ private Result enablePlugin(Plugin plugin) {
                 .lastTransitionTime(clock.instant())
                 .build());
             status.setPhase(Plugin.Phase.FAILED);
+            removeStartTaskIfPresent(pluginName);
             return Result.doNotRetry();
         }
         if (!Plugin.Phase.STARTING.equals(status.getPhase())) {
-            pluginStartTasks.computeIfAbsent(pluginName, name -> scheduler.schedule(() -> {
-                log.info("Starting plugin {} in background thread.", name);
-                try {
-                    var state = pluginManager.startPlugin(name);
-                    log.info("Plugin {} started with state {}.", name, state);
-                } catch (Throwable t) {
-                    var pluginWrapper = pluginManager.getPlugin(name);
-                    if (pluginWrapper != null) {
-                        pluginWrapper.setPluginState(PluginState.FAILED);
-                        pluginWrapper.setFailedException(t);
-                    }
+            pluginStartTasks.compute(pluginName, (name, old) -> {
+                if (old != null && !old.isDisposed()) {
+                    log.info("Cancelling old starting task for plugin {}.", name);
+                    old.dispose();
+                    log.info("Cancelled old starting task for plugin {}.", name);
                 }
-            }));
+                return scheduler.schedule(() -> {
+                    log.info("Starting plugin {} in background thread.", name);
+                    try {
+                        var state = pluginManager.startPlugin(name);
+                        log.info("Plugin {} started with state {}.", name, state);
+                    } catch (Throwable t) {
+                        var pluginWrapper = pluginManager.getPlugin(name);
+                        if (pluginWrapper != null) {
+                            pluginWrapper.setPluginState(PluginState.FAILED);
+                            pluginWrapper.setFailedException(t);
+                        }
+                    }
+                });
+            });
             status.setPhase(Plugin.Phase.STARTING);
             conditions.addAndEvictFIFO(Condition.builder()
                 .type(ConditionType.PROGRESSING)
@@ -441,12 +451,7 @@ private Result disablePlugin(Plugin plugin) {
             }
             try {
                 // First, stop starting task if exists
-                pluginStartTasks.computeIfPresent(pluginName, (name, disposable) -> {
-                    log.info("Cancelling starting task for plugin {}.", name);
-                    disposable.dispose();
-                    log.info("Cancelled starting task for plugin {}.", name);
-                    return null;
-                });
+                removeStartTaskIfPresent(pluginName);
                 pluginManager.disablePlugin(pluginName);
             } catch (Throwable e) {
                 conditions.addAndEvictFIFO(Condition.builder()
@@ -805,6 +810,17 @@ public Controller setupWith(ControllerBuilder builder) {
             .build();
     }
 
+    private void removeStartTaskIfPresent(String pluginName) {
+        pluginStartTasks.computeIfPresent(pluginName, (name, disposable) -> {
+            if (!disposable.isDisposed()) {
+                log.info("Cancelling starting task for plugin {}.", name);
+                disposable.dispose();
+                log.info("Cancelled starting task for plugin {}.", name);
+            }
+            return null;
+        });
+    }
+
     private Result createOrUpdateReverseProxy(Plugin plugin) {
         String pluginName = plugin.getMetadata().getName();
         String reverseProxyName = buildReverseProxyName(pluginName);