fix(cache): pass watch keys to Transaction so optimistic lock actually works (#701)

* feat(redis): add Redis Cluster client path to plugin-daemon

Wire the missing third branch so operators can point the daemon at a
Redis Cluster without falling back to dialing an empty REDIS_HOST:PORT
and panicking at startup.

* introduce REDIS_USE_CLUSTERS / REDIS_CLUSTERS /
  REDIS_CLUSTERS_PASSWORD envs, aligned with dify api naming so Helm
  can set a single env group
* branch in PluginManager.Launch for Cluster ahead of the existing
  Sentinel / standalone paths; falls back to REDIS_PASSWORD if the
  cluster-specific password is not set
* implement cache.InitRedisClusterClient via redis.NewClusterClient;
  downstream cache / lock / pub-sub helpers keep working because
  `client` is declared as redis.UniversalClient

Redis Cluster disables SELECT DB, so the cluster branch does not
plumb RedisDB. This is intentional and documented in the release
note accompanying the overall Redis Cluster support work.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* feat(redis): validate Redis cluster addresses and improve transaction handling

* feat(redis): implement CSV parsing utility and refactor Redis transaction handling

* feat(redis): scope plugin-daemon to standalone + sentinel only

Narrow the plugin-daemon Redis client to standalone and sentinel modes;
remove the cluster code paths that were briefly exercised on this branch.

Changes:
- Remove REDIS_USE_CLUSTERS / REDIS_CLUSTERS / REDIS_CLUSTERS_PASSWORD
  env fields from app.Config.
- Remove the cluster branch (plus the two cluster-specific fail-fast
  guards on REDIS_USE_CLUSTERS + REDIS_USE_SENTINEL and REDIS_DB != 0)
  from PluginManager.Launch. The init block is now just sentinel or
  standalone.
- Delete cache.InitRedisClusterClient. The main redis client + all
  helpers (Transaction, pub/sub, lock, etc.) already route through
  redis.UniversalClient so downstream code needs no adjustment.

Intentionally kept:
- Transaction(fn, watchKeys...) signature: the variadic signature is
  backwards-compatible with every existing Transaction(fn) call and the
  one new-style caller (debugging_service) benefits from real WATCH
  semantics even on standalone — this is correctness-independent of
  cluster support.
- parser.SplitAndTrimCSV: the sentinel branch migrated to it in 617e7a5
  and still uses it to parse REDIS_SENTINELS, so the helper has a
  standalone/sentinel consumer and stays.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: lin-snow

internal/core/plugin_manager/manager.go

@@ -2,7 +2,6 @@ package plugin_manager
 
 import (
 	"fmt"
-	"strings"
 
 	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/langgenius/dify-cloud-kit/oss"
@@ -18,6 +17,7 @@ import (
 	"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/log"
+	"github.com/langgenius/dify-plugin-daemon/pkg/utils/parser"
 )
 
 type PluginManager struct {
@@ -128,7 +128,7 @@ func (p *PluginManager) Launch(configuration *app.Config) {
 	// init redis client
 	if configuration.RedisUseSentinel {
 		// use Redis Sentinel
-		sentinels := strings.Split(configuration.RedisSentinels, ",")
+		sentinels := parser.SplitAndTrimCSV(configuration.RedisSentinels)
 		if err := cache.InitRedisSentinelClient(
 			sentinels,
 			configuration.RedisSentinelServiceName,

internal/service/debugging_service/connection_key.go

@@ -45,10 +45,11 @@ func GetConnectionKey(info ConnectionInfo) (string, error) {
 	)
 
 	if err == cache.ErrNotFound {
+		id2key := strings.Join([]string{CONNECTION_KEY_MANAGER_ID2KEY_PREFIX, info.TenantId}, ":")
 		err := cache.Transaction(func(p redis.Pipeliner) error {
 			k := uuid.New().String()
 			_, err = cache.SetNX(
-				strings.Join([]string{CONNECTION_KEY_MANAGER_ID2KEY_PREFIX, info.TenantId}, ":"),
+				id2key,
 				Key{Key: k},
 				CONNECTION_KEY_EXPIRE_TIME,
 				p,
@@ -70,7 +71,7 @@ func GetConnectionKey(info ConnectionInfo) (string, error) {
 			key = &Key{Key: k}
 
 			return nil
-		})
+		}, id2key)
 
 		if err != nil {
 			return "", err

pkg/utils/cache/redis.go

@@ -582,11 +582,16 @@ func Expire(key string, time time.Duration, context ...redis.Cmdable) (bool, err
 	return getCmdable(context...).Expire(ctx, serialKey(key), time).Result()
 }
 
-func Transaction(fn func(redis.Pipeliner) error) error {
+func Transaction(fn func(redis.Pipeliner) error, watchKeys ...string) error {
 	if client == nil {
 		return ErrDBNotInit
 	}
 
+	serialized := make([]string, len(watchKeys))
+	for i, k := range watchKeys {
+		serialized[i] = serialKey(k)
+	}
+
 	return client.Watch(ctx, func(tx *redis.Tx) error {
 		_, err := tx.TxPipelined(ctx, func(p redis.Pipeliner) error {
 			return fn(p)
@@ -595,7 +600,7 @@ func Transaction(fn func(redis.Pipeliner) error) error {
 			return nil
 		}
 		return err
-	})
+	}, serialized...)
 }
 
 func Publish(channel string, message any, context ...redis.Cmdable) error {

pkg/utils/parser/csv.go

@@ -0,0 +1,17 @@
+package parser
+
+import "strings"
+
+func SplitAndTrimCSV(s string) []string {
+	if s == "" {
+		return nil
+	}
+	parts := strings.Split(s, ",")
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if trimmed := strings.TrimSpace(p); trimmed != "" {
+			out = append(out, trimmed)
+		}
+	}
+	return out
+}

pkg/utils/parser/csv_test.go

@@ -0,0 +1,37 @@
+package parser
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestSplitAndTrimCSV(t *testing.T) {
+	tests := []struct {
+		name  string
+		input string
+		want  []string
+	}{
+		{"empty string", "", nil},
+		{"only separators", ",,,", []string{}},
+		{"only whitespace between separators", " , , ,", []string{}},
+		{"single value", "a", []string{"a"}},
+		{"trims surrounding whitespace", " a , b ", []string{"a", "b"}},
+		{"drops empty segments", "a, ,b", []string{"a", "b"}},
+		{"preserves inner colons", "host1:6379, host2:6379", []string{"host1:6379", "host2:6379"}},
+		{"trailing comma", "a,b,", []string{"a", "b"}},
+		{"leading comma", ",a,b", []string{"a", "b"}},
+		{"tabs and mixed whitespace", "\ta\t,\nb\n", []string{"a", "b"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := SplitAndTrimCSV(tt.input)
+			if len(got) == 0 && len(tt.want) == 0 {
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("SplitAndTrimCSV(%q) = %#v, want %#v", tt.input, got, tt.want)
+			}
+		})
+	}
+}