fix: make shared plugin runtime installs idempotent

docs/claude/cache.md

@@ -70,6 +70,61 @@ if acquired {
 }
 ```
 
+## Owner-Aware Distributed Locks (OwnedLock)
+
+The `AcquireOwnedLock` function provides token-based lock ownership with advanced features:
+
+```go
+// Acquire lock with timeout
+lock, err := cache.AcquireOwnedLock(key, expire, tryLockTimeout)
+if err == cache.ErrLockTimeout {
+    // Lock acquisition timed out
+    return err
+}
+if err != nil {
+    return err
+}
+defer lock.Unlock()
+
+// Critical section
+```
+
+The OwnedLock type supports:
+
+**Renewal**: Extend lock expiration while holding it
+```go
+if err := lock.Renew(time.Minute); err == cache.ErrLockNotOwned {
+    // Lock was lost
+}
+```
+
+**Keep-Alive**: Automatic periodic renewal
+```go
+ctx, cancel := context.WithCancel(context.Background())
+defer cancel()
+
+errCh := lock.KeepAlive(ctx, time.Minute)
+// Lock is renewed every expire/3 interval
+// Cancel context to stop keep-alive
+if err := <-errCh; err != nil {
+    // Keep-alive failed (lock lost or error)
+}
+```
+
+**Safe Unlock**: Only the lock owner can unlock
+```go
+if err := lock.Unlock(); err == cache.ErrLockNotOwned {
+    // Lock expired or was never owned
+}
+```
+
+The OwnedLock mechanism uses unique tokens to ensure only the lock owner can renew or release the lock, preventing accidental releases by other processes.
+
+**Error Constants**:
+- `ErrLockTimeout`: Lock acquisition timed out
+- `ErrLockNotOwned`: Returned when attempting to renew or unlock a lock that is not owned
+- `ErrInvalidLockExpiration`: Returned when lock expiration duration is not positive
+
 ## Auto-type Operations
 
 Helper functions in `redis_auto_type.go`:

internal/core/control_panel/launcher_local.go

@@ -76,26 +76,48 @@ func (c *ControlPanel) LaunchLocalPlugin(
 		expire := 15 * time.Minute
 		tryTimeout := 2 * time.Minute
 		log.Info("acquiring distributed init lock", "plugin", pluginUniqueIdentifier.String(), "expire", expire.String())
-		if err := cache.Lock(lockKey, expire, tryTimeout); err != nil {
-			// failed to acquire the lock within timeout
-			err = errors.Join(err, fmt.Errorf("failed to acquire distributed env-init lock"))
-			c.WalkNotifiers(func(notifier ControlPanelNotifier) {
-				notifier.OnLocalRuntimeStartFailed(pluginUniqueIdentifier, err)
-			})
-			// release semaphore and local lock
-			releaseLockAndSemaphore()
-			return nil, nil, err
+		lock, err := cache.AcquireOwnedLock(lockKey, expire, tryTimeout)
+		if err != nil {
+			log.Warn(
+				"failed to acquire distributed init lock; continuing with idempotent initialization",
+				"plugin", pluginUniqueIdentifier.String(),
+				"error", err.Error(),
+			)
 		}
-		defer func() {
-			if unlockErr := cache.Unlock(lockKey); unlockErr != nil {
-				log.Warn("failed to release distributed init lock", "plugin", pluginUniqueIdentifier.String(), "error", unlockErr.Error())
+
+		var stopRenew context.CancelFunc
+		var renewResult <-chan error
+		if lock != nil {
+			var renewCtx context.Context
+			renewCtx, stopRenew = context.WithCancel(context.Background())
+			renewResult = lock.KeepAlive(renewCtx, expire)
+		}
+
+		initErr := runtime.InitEnvironment(decoder)
+		if lock != nil {
+			stopRenew()
+			if renewErr := <-renewResult; renewErr != nil {
+				log.Warn(
+					"lost distributed init lock; runtime initialization remains authoritative",
+					"plugin", pluginUniqueIdentifier.String(),
+					"error", renewErr.Error(),
+				)
+			}
+			if unlockErr := lock.Unlock(); unlockErr != nil {
+				if !errors.Is(unlockErr, cache.ErrLockNotOwned) {
+					log.Warn(
+						"failed to release distributed init lock",
+						"plugin", pluginUniqueIdentifier.String(),
+						"error", unlockErr.Error(),
+					)
+				}
 			} else {
 				log.Info("released distributed init lock", "plugin", pluginUniqueIdentifier.String())
 			}
-		}()
+		}
 
-		if err := runtime.InitEnvironment(decoder); err != nil {
-			err = errors.Join(err, fmt.Errorf("failed to init environment"))
+		if initErr != nil {
+			err = errors.Join(initErr, fmt.Errorf("failed to init environment"))
 			// notify new runtime launch failed
 			c.WalkNotifiers(func(notifier ControlPanelNotifier) {
 				notifier.OnLocalRuntimeStartFailed(pluginUniqueIdentifier, err)

internal/core/plugin_manager/installer.go

@@ -18,6 +18,7 @@ import (
 	"github.com/langgenius/dify-plugin-daemon/pkg/utils/log"
 	"github.com/langgenius/dify-plugin-daemon/pkg/utils/routine"
 	"github.com/langgenius/dify-plugin-daemon/pkg/utils/stream"
+	"gorm.io/gorm/clause"
 )
 
 var (
@@ -27,6 +28,13 @@ var (
 func (p *PluginManager) Install(
 	ctx context.Context,
 	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
+) (*stream.Stream[installation_entities.PluginInstallResponse], error) {
+	return p.EnsureRuntime(ctx, pluginUniqueIdentifier)
+}
+
+func (p *PluginManager) EnsureRuntime(
+	ctx context.Context,
+	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
 ) (*stream.Stream[installation_entities.PluginInstallResponse], error) {
 	if p.config.Platform == app.PLATFORM_LOCAL {
 		return p.installLocal(ctx, pluginUniqueIdentifier)
@@ -170,6 +178,33 @@ func (p *PluginManager) updateServerlessRuntimeModel(
 	return db.Update(&serverlessModel)
 }
 
+func (p *PluginManager) persistServerlessRuntime(
+	pluginUniqueIdentifier plugin_entities.PluginUniqueIdentifier,
+	functionURL string,
+	functionName string,
+) error {
+	serverlessModel := &models.ServerlessRuntime{
+		Checksum:               pluginUniqueIdentifier.Checksum(),
+		Type:                   models.SERVERLESS_RUNTIME_TYPE_SERVERLESS,
+		FunctionURL:            functionURL,
+		FunctionName:           functionName,
+		PluginUniqueIdentifier: pluginUniqueIdentifier.String(),
+	}
+	if err := db.DifyPluginDB.Clauses(clause.OnConflict{
+		Columns: []clause.Column{{Name: "plugin_unique_identifier"}},
+		DoUpdates: clause.AssignmentColumns([]string{
+			"checksum",
+			"type",
+			"function_url",
+			"function_name",
+		}),
+	}).Create(serverlessModel).Error; err != nil {
+		return err
+	}
+
+	return p.ClearServerlessRuntimeCache(pluginUniqueIdentifier)
+}
+
 // whenever a plugin was installed successfully, a record will be inserted into `models.ServerlessRuntime`
 func (p *PluginManager) installServerless(
 	ctx context.Context,
@@ -209,37 +244,11 @@ func (p *PluginManager) installServerless(
 					return
 				}
 
-				// check if the plugin is already installed
-				// NOTE: models.ServerlessRuntime is a tenant-isolated model
-				// it hands only engine-level persist data like which serverless runtime is installed
-				// that's why we placed it here, not in service layer.
-				//
-				// service layer takes care of tenant-level persist data like "which tenant installed which plugin"
-				_, err := db.GetOne[models.ServerlessRuntime](
-					db.Equal("plugin_unique_identifier", pluginUniqueIdentifier.String()),
-					db.Equal("type", string(models.SERVERLESS_RUNTIME_TYPE_SERVERLESS)),
-				)
-				if err == db.ErrDatabaseNotFound {
-					// create a new serverless runtime
-					serverlessModel := &models.ServerlessRuntime{
-						Checksum:               pluginUniqueIdentifier.Checksum(),
-						Type:                   models.SERVERLESS_RUNTIME_TYPE_SERVERLESS,
-						FunctionURL:            functionUrl,
-						FunctionName:           functionName,
-						PluginUniqueIdentifier: pluginUniqueIdentifier.String(),
-					}
-					err = db.Create(serverlessModel)
-					if err != nil {
-						responseStream.Write(installation_entities.PluginInstallResponse{
-							Event: installation_entities.PluginInstallEventError,
-							Data:  "failed to create serverless runtime",
-						})
-						return
-					}
-				} else if err != nil {
+				if err := p.persistServerlessRuntime(pluginUniqueIdentifier, functionUrl, functionName); err != nil {
+					log.Error("failed to persist serverless runtime", "error", err)
 					responseStream.Write(installation_entities.PluginInstallResponse{
 						Event: installation_entities.PluginInstallEventError,
-						Data:  "failed to check if the plugin is already installed",
+						Data:  "failed to persist serverless runtime",
 					})
 					return
 				}
@@ -282,21 +291,21 @@ func (p *PluginManager) installLocal(
 		routinepkg.RoutineLabelKeyModule: "plugin_manager",
 		routinepkg.RoutineLabelKeyMethod: "installLocal",
 	}, func() {
-		// firstly, install the plugin, then launch it, delete it if process fails
+		// First publish the plugin package, then launch it. A failed attempt must not
+		// delete the shared package because another caller may already depend on it.
 		var success bool = false
+		var stopRuntime bool = false
 		var runtime *local_runtime.LocalPluginRuntime
 		var ch <-chan error
 
 		defer responseStream.Close()
+		defer p.controlPanel.EnableLocalPluginAutoLaunch(pluginUniqueIdentifier)
 		defer func() {
-			if !success {
-				p.controlPanel.RemoveLocalPlugin(pluginUniqueIdentifier)
-
-				// release the lock, avoid a potential race condition
-				// which causes plugins never to be scheduled automatically
-				p.controlPanel.EnableLocalPluginAutoLaunch(pluginUniqueIdentifier)
-
-				// forcefully stop runtime, prevent continuous scheduling
+			if !success && stopRuntime {
+				// Stop only the runtime constructed by this attempt. The canonical
+				// package remains available for concurrent callers and retries. A
+				// timeout does not stop the runtime because it may become ready after
+				// the caller has left and be observed by another ensure attempt.
 				if runtime != nil {
 					runtime.Stop(false)
 				}
@@ -340,8 +349,10 @@ func (p *PluginManager) installLocal(
 		}
 
 		ticker := time.NewTicker(5 * time.Second)
+		defer ticker.Stop()
 		timeout := time.Duration(p.config.PythonEnvInitTimeout) * time.Second
 		timer := time.NewTimer(timeout)
+		defer timer.Stop()
 
 		for {
 			select {
@@ -363,6 +374,7 @@ func (p *PluginManager) installLocal(
 				})
 			case err := <-ch:
 				if err != nil {
+					stopRuntime = true
 					responseStream.Write(installation_entities.PluginInstallResponse{
 						Event: installation_entities.PluginInstallEventError,
 						Data:  fmt.Sprintf("failed to launch plugin: %s", err.Error()),

internal/core/plugin_manager/local_launch_test.go

@@ -1 +1,87 @@
 package plugin_manager
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	miniredis "github.com/alicebob/miniredis/v2"
+	cloudoss "github.com/langgenius/dify-cloud-kit/oss"
+	"github.com/langgenius/dify-cloud-kit/oss/factory"
+	"github.com/langgenius/dify-plugin-daemon/internal/types/app"
+	"github.com/langgenius/dify-plugin-daemon/pkg/entities/installation_entities"
+	"github.com/langgenius/dify-plugin-daemon/pkg/plugin_packager/decoder"
+	"github.com/langgenius/dify-plugin-daemon/pkg/utils/cache"
+	"github.com/langgenius/dify-plugin-daemon/pkg/utils/routine"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEnsureLocalRuntimeFailureKeepsInstalledPackage(t *testing.T) {
+	eventData, installed := runFailingLocalRuntimeInstall(t, false)
+
+	require.NotEmpty(t, eventData)
+	require.True(t, installed)
+}
+
+func TestEnsureLocalRuntimeContinuesWhenRedisLockIsUnavailable(t *testing.T) {
+	eventData, installed := runFailingLocalRuntimeInstall(t, true)
+
+	require.Contains(t, eventData, "missing-uv")
+	require.NotContains(t, eventData, "failed to acquire distributed env-init lock")
+	require.True(t, installed)
+}
+
+func runFailingLocalRuntimeInstall(t *testing.T, closeRedisBeforeInstall bool) (string, bool) {
+	t.Helper()
+	routine.InitPool(4)
+
+	redisServer := miniredis.RunT(t)
+	require.NoError(t, cache.InitRedisClient(redisServer.Addr(), cache.RedisCredentials{}, false, 0, nil))
+	t.Cleanup(func() {
+		_ = cache.Close()
+	})
+
+	storageDir := t.TempDir()
+	storage, err := factory.Load("local", cloudoss.OSSArgs{
+		Local: &cloudoss.Local{Path: storageDir},
+	})
+	require.NoError(t, err)
+
+	manager := InitGlobalManager(storage, &app.Config{
+		Platform:                       app.PLATFORM_LOCAL,
+		PluginMediaCachePath:           "assets",
+		PluginMediaCacheSize:           4,
+		PluginAssetCacheSize:           4,
+		PluginInstalledPath:            "installed",
+		PluginPackageCachePath:         "packages",
+		PluginLocalLaunchingConcurrent: 1,
+		PluginWorkingPath:              filepath.Join(storageDir, "working"),
+		PythonEnvInitTimeout:           1,
+		UvPath:                         filepath.Join(storageDir, "missing-uv"),
+	})
+
+	packageBytes, err := os.ReadFile("testdata/openai.difypkg")
+	require.NoError(t, err)
+	packageDecoder, err := decoder.NewZipPluginDecoder(packageBytes)
+	require.NoError(t, err)
+	identifier, err := packageDecoder.UniqueIdentity()
+	require.NoError(t, err)
+	require.NoError(t, manager.packageBucket.Save(identifier.String(), packageBytes))
+	if closeRedisBeforeInstall {
+		require.NoError(t, cache.Close())
+	}
+
+	response, err := manager.EnsureRuntime(context.Background(), identifier)
+	require.NoError(t, err)
+	eventData := ""
+	require.NoError(t, response.Process(func(event installation_entities.PluginInstallResponse) {
+		if event.Event == installation_entities.PluginInstallEventError {
+			eventData = event.Data
+		}
+	}))
+
+	exists, err := manager.installedBucket.Exists(identifier)
+	require.NoError(t, err)
+	return eventData, exists
+}