fix: synchronize remote debugging plugins across cluster nodes

Remote debugging plugins were not being synchronized across cluster nodes,
causing "no plugin available nodes found" errors when trying to invoke
plugins from different nodes.

1. **Remote debugging plugins not registered to cluster** - The
   `ClusterTunnel` notifier was not being added to ControlPanel
2. **Plugin ID inconsistency** - Remote plugins used different plugin_id
   formats during installation vs. querying
3. **Non-idempotent registration** - `RegisterPlugin` failed on reconnection
   with "plugin has been registered" error

- **internal/types/models/curd/atomic.go**:
  - Unify plugin_id calculation for remote plugins (author/name without version)
  - Remove plugin_id from plugin query conditions
  - Clear old cache when plugin_id is updated

- **internal/cluster/plugin.go**:
  - Make `RegisterPlugin` idempotent by updating existing plugin instead
    of returning error

- **internal/core/control_panel/daemon.go**:
  - Add cluster field to ControlPanel
  - Add SetCluster() method for lazy cluster initialization

- **internal/core/control_panel/server_debugger.go**:
  - Register remote debugging plugins to cluster on connection
  - Unregister from cluster on disconnection

- **internal/core/plugin_manager/manager.go**:
  - Add SetCluster() method to set cluster after initialization

- **internal/server/server.go**:
  - Call SetCluster() instead of AddClusterTunnel()

Only remote debugging plugins are synchronized across cluster nodes.
Local plugins run only on the node where they are installed and are
not registered to the cluster.

- Error handling improvements using `errors.Is()` instead of `==`
- Handle 404 for missing plugin assets gracefully
- Handle already-installed debugging plugins gracefully

- Remote debugging plugin can be invoked from any node in the cluster
- Plugin reconnection works without errors
- Cache invalidation works correctly when plugin_id changes

Author: fatelei

internal/cluster/plugin.go

@@ -37,7 +37,17 @@ func (c *Cluster) RegisterPlugin(lifetime plugin_entities.PluginLifetime) error
 	}
 
 	if c.plugins.Exists(identity.String()) {
-		return errors.New("plugin has been registered")
+		// idempotent: plugin already registered, just update its state
+		if existing, ok := c.plugins.Load(identity.String()); ok {
+			// update the lifetime reference
+			existing.lifetime = lifetime
+			// update plugin state immediately
+			err = c.doPluginStateUpdate(existing)
+			if err != nil {
+				return errors.Join(err, errors.New("failed to update plugin state"))
+			}
+		}
+		return nil
 	}
 
 	l := &pluginLifeTime{

internal/cluster/plugin_test.go

@@ -140,6 +140,107 @@ func TestPluginScheduleLifetime(t *testing.T) {
 	}
 }
 
+func TestPluginRegisterIdempotent(t *testing.T) {
+	plugin := getRandomPluginRuntime()
+	cluster, err := createSimulationCluster(1)
+	if err != nil {
+		t.Errorf("create simulation cluster failed: %v", err)
+		return
+	}
+
+	launchSimulationCluster(cluster)
+	defer closeSimulationCluster(cluster, t)
+
+	// wait for cluster to be ready
+	time.Sleep(time.Second * 1)
+
+	// first registration should succeed
+	err = cluster[0].RegisterPlugin(&plugin)
+	if err != nil {
+		t.Errorf("first register plugin failed: %v", err)
+		return
+	}
+
+	identity, err := plugin.Identity()
+	if err != nil {
+		t.Errorf("get plugin identity failed: %v", err)
+		return
+	}
+
+	hashedIdentity := plugin_entities.HashedIdentity(identity.String())
+
+	// wait for plugin to be scheduled
+	time.Sleep(time.Second * 1)
+
+	// verify plugin is registered
+	nodes, err := cluster[0].FetchPluginAvailableNodesByHashedId(hashedIdentity)
+	if err != nil {
+		t.Errorf("fetch plugin available nodes failed: %v", err)
+		return
+	}
+
+	if len(nodes) != 1 {
+		t.Errorf("plugin not scheduled after first registration")
+		return
+	}
+
+	// second registration with same identity should be idempotent (no error)
+	err = cluster[0].RegisterPlugin(&plugin)
+	if err != nil {
+		t.Errorf("second register plugin failed (should be idempotent): %v", err)
+		return
+	}
+
+	// verify plugin is still registered after second registration
+	nodes, err = cluster[0].FetchPluginAvailableNodesByHashedId(hashedIdentity)
+	if err != nil {
+		t.Errorf("fetch plugin available nodes failed after second registration: %v", err)
+		return
+	}
+
+	if len(nodes) != 1 {
+		t.Errorf("plugin not available after second registration")
+		return
+	}
+
+	// unregister the plugin
+	if err := cluster[0].UnregisterPlugin(&plugin); err != nil {
+		t.Errorf("unregister plugin failed: %v", err)
+		return
+	}
+
+	// verify plugin is unregistered
+	nodes, err = cluster[0].FetchPluginAvailableNodesByHashedId(hashedIdentity)
+	if err != nil {
+		t.Errorf("fetch plugin available nodes failed after unregister: %v", err)
+		return
+	}
+
+	if len(nodes) != 0 {
+		t.Errorf("plugin still available after unregister")
+		return
+	}
+
+	// registration after unregister should succeed again
+	err = cluster[0].RegisterPlugin(&plugin)
+	if err != nil {
+		t.Errorf("register after unregister failed: %v", err)
+		return
+	}
+
+	// verify plugin is registered again
+	nodes, err = cluster[0].FetchPluginAvailableNodesByHashedId(hashedIdentity)
+	if err != nil {
+		t.Errorf("fetch plugin available nodes failed after re-registration: %v", err)
+		return
+	}
+
+	if len(nodes) != 1 {
+		t.Errorf("plugin not available after re-registration")
+		return
+	}
+}
+
 // TODO: I need to implement this test, now it's randomly working
 // func TestPluginScheduleWhenMasterClusterShutdown(t *testing.T) {
 // 	plugins := []fakePlugin{

internal/core/control_panel/daemon.go

@@ -4,6 +4,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/langgenius/dify-plugin-daemon/internal/cluster"
 	"github.com/langgenius/dify-plugin-daemon/internal/core/debugging_runtime"
 	"github.com/langgenius/dify-plugin-daemon/internal/core/local_runtime"
 	"github.com/langgenius/dify-plugin-daemon/internal/core/plugin_manager/media_transport"
@@ -17,6 +18,9 @@ type ControlPanel struct {
 	// app config
 	config *app.Config
 
+	// cluster for remote debugging plugins
+	cluster *cluster.Cluster
+
 	// debugging server
 	debuggingServer *debugging_runtime.RemotePluginServer
 
@@ -82,12 +86,14 @@ func NewControlPanel(
 	mediaBucket *media_transport.MediaBucket,
 	packageBucket *media_transport.PackageBucket,
 	installedBucket *media_transport.InstalledBucket,
+	cluster *cluster.Cluster,
 ) *ControlPanel {
 	return &ControlPanel{
 		config:          config,
 		mediaBucket:     mediaBucket,
 		packageBucket:   packageBucket,
 		installedBucket: installedBucket,
+		cluster:         cluster,
 
 		localPluginLaunchingSemaphore: make(chan bool, config.PluginLocalLaunchingConcurrent),
 
@@ -99,3 +105,7 @@ func NewControlPanel(
 		localPluginInstallationLock: lock.NewGranularityLock(),
 	}
 }
+
+func (c *ControlPanel) SetCluster(cluster *cluster.Cluster) {
+	c.cluster = cluster
+}

internal/core/control_panel/server_debugger.go

@@ -33,6 +33,12 @@ func (c *ControlPanel) onDebuggingRuntimeConnected(
 	// store plugin runtime
 	c.debuggingPluginRuntime.Store(pluginIdentifier, rpr)
 
+	if c.cluster != nil {
+		if err = c.cluster.RegisterPlugin(rpr); err != nil {
+			log.Error("failed to register remote debugging plugin to cluster", "error", err)
+		}
+	}
+
 	// notify notifiers a new debugging runtime is connected
 	c.WalkNotifiers(func(notifier ControlPanelNotifier) {
 		notifier.OnDebuggingRuntimeConnected(rpr)
@@ -51,6 +57,12 @@ func (c *ControlPanel) onDebuggingRuntimeDisconnected(
 		return
 	}
 
+	if c.cluster != nil {
+		if err = c.cluster.UnregisterPlugin(rpr); err != nil {
+			log.Error("failed to unregister remote debugging plugin from cluster", "error", err)
+		}
+	}
+
 	// delete plugin runtime
 	c.debuggingPluginRuntime.Delete(pluginIdentifier)
 

internal/core/plugin_manager/cluster_tunnel.go

@@ -67,3 +67,25 @@ func (t *ClusterTunnel) OnLocalRuntimeStopped(
 ) {
 	// NOP
 }
+
+func (t *ClusterTunnel) OnLocalRuntimeScaleUp(
+	runtime *local_runtime.LocalPluginRuntime,
+	instanceNums int32,
+) {
+	// NOP
+}
+
+func (t *ClusterTunnel) OnLocalRuntimeScaleDown(
+	runtime *local_runtime.LocalPluginRuntime,
+	instanceNums int32,
+) {
+	// NOP
+}
+
+func (t *ClusterTunnel) OnLocalRuntimeInstanceLog(
+	runtime *local_runtime.LocalPluginRuntime,
+	instance *local_runtime.PluginInstance,
+	event plugin_entities.PluginLogEvent,
+) {
+	// NOP
+}

internal/core/plugin_manager/manager.go

@@ -6,6 +6,7 @@ import (
 
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/langgenius/dify-cloud-kit/oss"
+	"github.com/langgenius/dify-plugin-daemon/internal/cluster"
 	controlpanel "github.com/langgenius/dify-plugin-daemon/internal/core/control_panel"
 	"github.com/langgenius/dify-plugin-daemon/internal/core/dify_invocation"
 	"github.com/langgenius/dify-plugin-daemon/internal/core/dify_invocation/calldify"
@@ -80,6 +81,7 @@ func InitGlobalManager(oss oss.OSS, config *app.Config) *PluginManager {
 			mediaBucket,
 			packageBucket,
 			installedBucket,
+			nil, // cluster will be set later via SetCluster
 		),
 		config: config,
 	}
@@ -91,6 +93,10 @@ func InitGlobalManager(oss oss.OSS, config *app.Config) *PluginManager {
 	return manager
 }
 
+func (p *PluginManager) SetCluster(cluster *cluster.Cluster) {
+	p.controlPanel.SetCluster(cluster)
+}
+
 func Manager() *PluginManager {
 	return manager
 }

internal/server/controllers/plugins.go

@@ -18,6 +18,9 @@ func GetAsset(c *gin.Context) {
 	asset, err := pluginManager.GetAsset(c.Param("id"))
 
 	if err != nil {
+		if strings.Contains(err.Error(), "no such file or directory") {
+			return
+		}
 		c.JSON(http.StatusInternalServerError, exception.InternalServerError(err).ToResponse())
 		return
 	}

internal/server/middleware.go

@@ -57,7 +57,7 @@ func (app *App) FetchPluginInstallation() gin.HandlerFunc {
 			},
 		)
 
-		if err == db.ErrDatabaseNotFound {
+		if errors.Is(err, db.ErrDatabaseNotFound) {
 			ctx.AbortWithStatusJSON(404, exception.ErrPluginNotFound().ToResponse())
 			return
 		}
@@ -119,6 +119,7 @@ func (app *App) redirectPluginInvokeByPluginIdentifier(
 	// try find the correct node
 	nodes, err := app.cluster.FetchPluginAvailableNodesById(plugin_unique_identifier.String())
 	if err != nil {
+		log.Error("Failed to fetch plugin nodes by id", "error", err)
 		ctx.AbortWithStatusJSON(
 			500,
 			exception.InternalServerError(
@@ -127,6 +128,7 @@ func (app *App) redirectPluginInvokeByPluginIdentifier(
 		)
 		return
 	} else if len(nodes) == 0 {
+		log.Error("no plugin available nodes found")
 		ctx.AbortWithStatusJSON(
 			404,
 			exception.InternalServerError(

internal/server/server.go

@@ -106,6 +106,9 @@ func (app *App) Run(config *app.Config) {
 	// create cluster
 	app.cluster = cluster.NewCluster(config)
 
+	// set cluster to control panel for remote debugging plugin synchronization
+	app.pluginManager.SetCluster(app.cluster)
+
 	// init manager
 	app.pluginManager.Launch(config)
 

internal/service/install_plugin.go

@@ -316,10 +316,10 @@ func UninstallPlugin(
 		db.Equal("tenant_id", tenant_id),
 		db.Equal("id", plugin_installation_id),
 	)
-	if err == db.ErrDatabaseNotFound {
-		return exception.ErrPluginNotFound().ToResponse()
-	}
 	if err != nil {
+		if errors.Is(err, db.ErrDatabaseNotFound) {
+			return entities.NewSuccessResponse(true)
+		}
 		return exception.InternalServerError(err).ToResponse()
 	}
 
@@ -349,7 +349,7 @@ func UninstallPlugin(
 	pluginInstallationCacheKey := helper.PluginInstallationCacheKey(pluginUniqueIdentifier.PluginID(), tenant_id)
 	_, _ = cache.AutoDelete[models.PluginInstallation](pluginInstallationCacheKey)
 
-	if deleteResponse.IsPluginDeleted && deleteResponse.Plugin != nil && deleteResponse.Plugin.InstallType == plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL {
+	if deleteResponse != nil && deleteResponse.IsPluginDeleted && deleteResponse.Plugin != nil && deleteResponse.Plugin.InstallType == plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL {
 		manager := plugin_manager.Manager()
 		if manager == nil {
 			return exception.InternalServerError(errors.New("plugin manager is not initialized")).ToResponse()
@@ -360,7 +360,7 @@ func UninstallPlugin(
 		}
 
 		shutdownCh, err := manager.ShutdownLocalPluginGracefully(pluginUniqueIdentifier)
-		if err == controlpanel.ErrLocalPluginRuntimeNotFound {
+		if errors.Is(err, controlpanel.ErrLocalPluginRuntimeNotFound) {
 			return entities.NewSuccessResponse(true)
 		} else if err != nil {
 			return exception.InternalServerError(err).ToResponse()