@@ -4,10 +4,13 @@ import (
44 "errors"
55 "fmt"
66 "sync"
7+ "time"
78
89 "github.com/langgenius/dify-plugin-daemon/internal/core/local_runtime"
910 "github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
1011 routinepkg "github.com/langgenius/dify-plugin-daemon/pkg/routine"
12+ "github.com/langgenius/dify-plugin-daemon/pkg/utils/cache"
13+ "github.com/langgenius/dify-plugin-daemon/pkg/utils/log"
1114 "github.com/langgenius/dify-plugin-daemon/pkg/utils/routine"
1215)
1316
@@ -62,15 +65,41 @@ func (c *ControlPanel) LaunchLocalPlugin(
6265 // init environment
6366 // whatever it's a user request to launch a plugin or a new plugin was found
6467 // by watch dog, initialize environment is a must
65- if err := runtime .InitEnvironment (decoder ); err != nil {
66- err = errors .Join (err , fmt .Errorf ("failed to init environment" ))
67- // notify new runtime launch failed
68- c .WalkNotifiers (func (notifier ControlPanelNotifier ) {
69- notifier .OnLocalRuntimeStartFailed (pluginUniqueIdentifier , err )
70- })
71- // release semaphore
72- releaseLockAndSemaphore ()
73- return nil , nil , err
68+ // To avoid cross-pod races on Python venv creation, guard InitEnvironment with a Redis-based distributed lock.
69+ {
70+ lockKey := fmt .Sprintf ("env_init_lock:%s" , pluginUniqueIdentifier .String ())
71+ // expire: generous upper bound for env initialization; tryLockTimeout: wait up to the same duration
72+ expire := 15 * time .Minute
73+ tryTimeout := 2 * time .Minute
74+ log .Info ("acquiring distributed init lock" , "plugin" , pluginUniqueIdentifier .String (), "expire" , expire .String ())
75+ if err := cache .Lock (lockKey , expire , tryTimeout ); err != nil {
76+ // failed to acquire the lock within timeout
77+ err = errors .Join (err , fmt .Errorf ("failed to acquire distributed env-init lock" ))
78+ c .WalkNotifiers (func (notifier ControlPanelNotifier ) {
79+ notifier .OnLocalRuntimeStartFailed (pluginUniqueIdentifier , err )
80+ })
81+ // release semaphore and local lock
82+ releaseLockAndSemaphore ()
83+ return nil , nil , err
84+ }
85+ defer func () {
86+ if unlockErr := cache .Unlock (lockKey ); unlockErr != nil {
87+ log .Warn ("failed to release distributed init lock" , "plugin" , pluginUniqueIdentifier .String (), "error" , unlockErr .Error ())
88+ } else {
89+ log .Info ("released distributed init lock" , "plugin" , pluginUniqueIdentifier .String ())
90+ }
91+ }()
92+
93+ if err := runtime .InitEnvironment (decoder ); err != nil {
94+ err = errors .Join (err , fmt .Errorf ("failed to init environment" ))
95+ // notify new runtime launch failed
96+ c .WalkNotifiers (func (notifier ControlPanelNotifier ) {
97+ notifier .OnLocalRuntimeStartFailed (pluginUniqueIdentifier , err )
98+ })
99+ // release semaphore
100+ releaseLockAndSemaphore ()
101+ return nil , nil , err
102+ }
74103 }
75104
76105 once := sync.Once {}
0 commit comments