feat(redis): add plugin sync timeout and enhance redis options handling

- Introduced dedicated plugin sync operation timeout (`homePluginSyncOperationTimeout`) for improved task handling. - Added `cloneRedisOptions` function to safely duplicate Redis options and prevent mutation. - Modified plugin sync to use a dedicated Redis client with independent configurations. - Enhanced `GetPluginSync` logic to handle cancellation, timeouts, and TLS handshake gracefully. - Expanded unit tests to validate new Redis client behavior, plugin sync timeouts, and cancellation handling.
2026-11-08 01:16:47 +08:00 · 2026-07-19 01:41:18 +08:00
parent 93d74a890a
commit b651a1a8fd
2 changed files with 306 additions and 8 deletions
@@ -19,6 +19,7 @@ import (
 	"time"

 	"github.com/redis/go-redis/v9"
+	"github.com/redis/go-redis/v9/maintnotifications"
 	"github.com/router-for-me/CLIProxyAPI/v7/internal/config"
 	"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginstore"
 	log "github.com/sirupsen/logrus"
@@ -37,6 +38,7 @@ const (
 	homeReconnectInterval          = time.Second
 	homeReconnectFailoverThreshold = 3
 	homeRedisOperationTimeout      = 3 * time.Second
+	homePluginSyncOperationTimeout = 2 * time.Minute
 	homeSubscriptionReceiveTimeout = 3 * time.Second
 	redisChannelCluster            = "cluster"
 )
@@ -90,8 +92,9 @@ type Client struct {
 	seedHost string
 	seedPort int

-	cmd *redis.Client
-	sub *redis.Client
+	cmd        *redis.Client
+	cmdOptions *redis.Options
+	sub        *redis.Client

 	heartbeatOK       atomic.Bool
 	clusterNodes      []clusterNode
@@ -143,6 +146,7 @@ func (c *Client) closeClientsLocked() {
 		_ = c.sub.Close()
 	}
 	c.cmd = nil
+	c.cmdOptions = nil
 	c.sub = nil
 }

@@ -186,6 +190,7 @@ func (c *Client) ensureClients() error {
 		if errOptions != nil {
 			return errOptions
 		}
+		c.cmdOptions = cloneRedisOptions(options)
 		c.cmd = redis.NewClient(options)
 	}
 	if c.sub == nil {
@@ -215,6 +220,21 @@ func (c *Client) redisOptionsLocked(addr string) (*redis.Options, error) {
 	}, nil
 }

+func cloneRedisOptions(options *redis.Options) *redis.Options {
+	if options == nil {
+		return nil
+	}
+	cloned := *options
+	if options.TLSConfig != nil {
+		cloned.TLSConfig = options.TLSConfig.Clone()
+	}
+	if options.MaintNotificationsConfig != nil {
+		maintNotifications := *options.MaintNotificationsConfig
+		cloned.MaintNotificationsConfig = &maintNotifications
+	}
+	return &cloned
+}
+
 func (c *Client) homeTLSConfigLocked(addr string) (*tls.Config, error) {
 	serverName := strings.TrimSpace(c.homeCfg.TLS.ServerName)
 	if serverName == "" {
@@ -302,6 +322,19 @@ func (c *Client) commandClient() (*redis.Client, error) {
 	return cmd, nil
 }

+func (c *Client) pluginSyncCommandOptions() (*redis.Options, error) {
+	if errEnsure := c.ensureClients(); errEnsure != nil {
+		return nil, errEnsure
+	}
+	c.mu.Lock()
+	options := cloneRedisOptions(c.cmdOptions)
+	c.mu.Unlock()
+	if options == nil {
+		return nil, ErrNotConnected
+	}
+	return options, nil
+}
+
 func (c *Client) subscriptionClient() (*redis.Client, error) {
 	if errEnsure := c.ensureClients(); errEnsure != nil {
 		return nil, errEnsure
@@ -970,16 +1003,16 @@ func (c *Client) GetPluginTasks(ctx context.Context) ([]PluginTask, error) {
 }

 func (c *Client) GetPluginSync(ctx context.Context, request pluginstore.PluginSyncRequest) (pluginstore.PluginSyncResponse, error) {
-	cmd, errClient := c.commandClient()
-	if errClient != nil {
-		return pluginstore.PluginSyncResponse{}, errClient
+	options, errOptions := c.pluginSyncCommandOptions()
+	if errOptions != nil {
+		return pluginstore.PluginSyncResponse{}, errOptions
 	}
 	payload, errMarshal := json.Marshal(request)
 	if errMarshal != nil {
 		return pluginstore.PluginSyncResponse{}, fmt.Errorf("marshal plugin sync request: %w", errMarshal)
 	}
 	requestCmd := redis.NewStringCmd(ctx, "get", redisKeyPluginSync, string(payload))
-	if errProcess := cmd.Process(ctx, requestCmd); errProcess != nil {
+	if errProcess := processPluginSyncCommand(ctx, options, requestCmd); errProcess != nil {
 		if message, ok := pluginSyncUnsupportedMessage(errProcess.Error()); ok {
 			return pluginstore.PluginSyncResponse{}, fmt.Errorf("%w: %s", ErrPluginSyncUnsupported, message)
 		}
@@ -1013,6 +1046,99 @@ func (c *Client) GetPluginSync(ctx context.Context, request pluginstore.PluginSy
 	return response, nil
 }

+func processPluginSyncCommand(ctx context.Context, options *redis.Options, command redis.Cmder) error {
+	if options == nil {
+		return ErrNotConnected
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	pluginSyncClient := newPluginSyncCommandClient(ctx, options)
+	if pluginSyncClient == nil {
+		return ErrNotConnected
+	}
+	errProcess := pluginSyncClient.Process(ctx, command)
+	errClose := pluginSyncClient.Close()
+	if errContext := ctx.Err(); errContext != nil {
+		return errContext
+	}
+	if errProcess != nil {
+		return errProcess
+	}
+	if errClose != nil {
+		return fmt.Errorf("close plugin sync command client: %w", errClose)
+	}
+	return nil
+}
+
+func newPluginSyncCommandClient(ctx context.Context, template *redis.Options) *redis.Client {
+	options := cloneRedisOptions(template)
+	if options == nil {
+		return nil
+	}
+	options.MaintNotificationsConfig = &maintnotifications.Config{Mode: maintnotifications.ModeDisabled}
+	baseDialer := options.Dialer
+	if baseDialer == nil {
+		baseDialer = pluginSyncDialer(options)
+	}
+	options.Dialer = func(dialCtx context.Context, network string, address string) (net.Conn, error) {
+		conn, errDial := baseDialer(dialCtx, network, address)
+		if errDial != nil {
+			return nil, errDial
+		}
+		return newPluginSyncCancelableConn(ctx, conn), nil
+	}
+	options.ReadTimeout = homePluginSyncOperationTimeout
+	options.MaxRetries = -1
+	return redis.NewClient(options)
+}
+
+func pluginSyncDialer(options *redis.Options) func(context.Context, string, string) (net.Conn, error) {
+	return func(ctx context.Context, network string, address string) (net.Conn, error) {
+		dialer := &net.Dialer{Timeout: options.DialTimeout, KeepAlive: 5 * time.Minute}
+		conn, errDial := dialer.DialContext(ctx, network, address)
+		if errDial != nil {
+			return nil, errDial
+		}
+		if options.TLSConfig == nil {
+			return conn, nil
+		}
+		tlsConn := tls.Client(conn, options.TLSConfig)
+		if errHandshake := tlsConn.HandshakeContext(ctx); errHandshake != nil {
+			return nil, errors.Join(errHandshake, conn.Close())
+		}
+		return tlsConn, nil
+	}
+}
+
+type pluginSyncCancelableConn struct {
+	net.Conn
+	done chan struct{}
+	once sync.Once
+}
+
+func newPluginSyncCancelableConn(ctx context.Context, conn net.Conn) net.Conn {
+	wrapped := &pluginSyncCancelableConn{Conn: conn, done: make(chan struct{})}
+	go func() {
+		select {
+		case <-ctx.Done():
+			if errDeadline := conn.SetDeadline(time.Now()); errDeadline != nil {
+				_ = conn.Close()
+			}
+		case <-wrapped.done:
+		}
+	}()
+	return wrapped
+}
+
+func (c *pluginSyncCancelableConn) Close() error {
+	if c == nil || c.Conn == nil {
+		return net.ErrClosed
+	}
+	c.once.Do(func() { close(c.done) })
+	return c.Conn.Close()
+}
+
 func pluginSyncUnsupportedResponse(raw []byte) (string, bool) {
 	var response struct {
 		Error struct {
@@ -336,6 +336,26 @@ func TestGetPluginTasksUsesPluginTasksKey(t *testing.T) {
 	}
 }

+func TestPluginSyncCommandClientUsesDedicatedTimeout(t *testing.T) {
+	template := &redis.Options{
+		Addr:         "127.0.0.1:1",
+		ReadTimeout:  homeRedisOperationTimeout,
+		WriteTimeout: homeRedisOperationTimeout,
+		MaxRetries:   -1,
+	}
+	pluginSync := newPluginSyncCommandClient(context.Background(), template)
+	if pluginSync == nil {
+		t.Fatal("newPluginSyncCommandClient() = nil")
+	}
+	t.Cleanup(func() { _ = pluginSync.Close() })
+	if pluginSync.Options().ReadTimeout != homePluginSyncOperationTimeout || pluginSync.Options().WriteTimeout != homeRedisOperationTimeout {
+		t.Fatalf("plugin sync timeouts = %s/%s, want %s/%s", pluginSync.Options().ReadTimeout, pluginSync.Options().WriteTimeout, homePluginSyncOperationTimeout, homeRedisOperationTimeout)
+	}
+	if template.ReadTimeout != homeRedisOperationTimeout || template.WriteTimeout != homeRedisOperationTimeout || template.MaxRetries != -1 {
+		t.Fatalf("template options were mutated: read=%s write=%s retries=%d", template.ReadTimeout, template.WriteTimeout, template.MaxRetries)
+	}
+}
+
 func TestGetPluginSyncUsesDedicatedCommandAndDecodesResponse(t *testing.T) {
 	response := pluginstore.PluginSyncResponse{
 		SchemaVersion: pluginstore.PluginSyncSchemaVersion,
@@ -395,6 +415,139 @@ func TestGetPluginSyncUsesDedicatedCommandAndDecodesResponse(t *testing.T) {
 	}
 }

+func TestGetPluginSyncExceedsBaseTimeoutAndKeepsBaseClientUsable(t *testing.T) {
+	response := pluginstore.PluginSyncResponse{
+		SchemaVersion: pluginstore.PluginSyncSchemaVersion,
+		ExpiresAt:     time.Now().UTC().Add(time.Minute),
+		Items:         []pluginstore.PluginSyncItem{},
+	}
+	payload, errMarshal := json.Marshal(response)
+	if errMarshal != nil {
+		t.Fatalf("Marshal() error = %v", errMarshal)
+	}
+	client, _ := newRedisCommandTestClient(t, func(args []string) string {
+		if len(args) < 2 || !strings.EqualFold(args[0], "GET") {
+			return "-ERR unexpected command\r\n"
+		}
+		switch args[1] {
+		case redisKeyPluginSync:
+			time.Sleep(3 * homeRedisTestOperationTimeout)
+			return fmt.Sprintf("$%d\r\n%s\r\n", len(payload), payload)
+		case redisKeyPluginTasks:
+			return "$2\r\n[]\r\n"
+		default:
+			return "-ERR unexpected key\r\n"
+		}
+	})
+	startedAt := time.Now()
+	got, errSync := client.GetPluginSync(context.Background(), pluginstore.PluginSyncRequest{
+		SchemaVersion: pluginstore.PluginSyncSchemaVersion, GOOS: "linux", GOARCH: "amd64",
+	})
+	if errSync != nil {
+		t.Fatalf("GetPluginSync() error = %v", errSync)
+	}
+	got.Clear()
+	if elapsed := time.Since(startedAt); elapsed < 2*homeRedisTestOperationTimeout {
+		t.Fatalf("GetPluginSync() elapsed = %s, want response beyond base timeout", elapsed)
+	}
+	if _, errTasks := client.GetPluginTasks(context.Background()); errTasks != nil {
+		t.Fatalf("GetPluginTasks() after plugin sync error = %v", errTasks)
+	}
+}
+
+func TestGetPluginSyncCancellationInterruptsRead(t *testing.T) {
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var startOnce sync.Once
+	client, commands := newRedisCommandTestClient(t, func(args []string) string {
+		if len(args) >= 2 && args[1] == redisKeyPluginSync {
+			startOnce.Do(func() { close(started) })
+			<-release
+		}
+		return "-ERR cancelled\r\n"
+	})
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-started
+		cancel()
+	}()
+	startedAt := time.Now()
+	_, errSync := client.GetPluginSync(ctx, pluginstore.PluginSyncRequest{
+		SchemaVersion: pluginstore.PluginSyncSchemaVersion, GOOS: "linux", GOARCH: "amd64",
+	})
+	close(release)
+	if !errors.Is(errSync, context.Canceled) {
+		t.Fatalf("GetPluginSync() error = %v, want context.Canceled", errSync)
+	}
+	if elapsed := time.Since(startedAt); elapsed > time.Second {
+		t.Fatalf("GetPluginSync() cancellation took %s", elapsed)
+	}
+	if count := commands.CountKey(redisKeyPluginSync); count != 1 {
+		t.Fatalf("plugin sync command count = %d, want 1", count)
+	}
+}
+
+func TestProcessPluginSyncCommandCancellationInterruptsTLSHandshake(t *testing.T) {
+	listener, errListen := net.Listen("tcp", "127.0.0.1:0")
+	if errListen != nil {
+		t.Fatalf("listen: %v", errListen)
+	}
+	defer func() { _ = listener.Close() }()
+	accepted := make(chan struct{})
+	release := make(chan struct{})
+	serverDone := make(chan error, 1)
+	go func() {
+		conn, errAccept := listener.Accept()
+		if errAccept != nil {
+			serverDone <- errAccept
+			return
+		}
+		close(accepted)
+		<-release
+		serverDone <- conn.Close()
+	}()
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-accepted
+		cancel()
+	}()
+	options := &redis.Options{
+		Addr:                  listener.Addr().String(),
+		TLSConfig:             &tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: true}, //nolint:gosec -- the test peer intentionally never completes TLS.
+		DialTimeout:           time.Second,
+		ReadTimeout:           homeRedisTestOperationTimeout,
+		WriteTimeout:          homeRedisTestOperationTimeout,
+		MaxRetries:            -1,
+		ContextTimeoutEnabled: true,
+	}
+	command := redis.NewStringCmd(ctx, "get", redisKeyPluginSync, `{}`)
+	startedAt := time.Now()
+	errProcess := processPluginSyncCommand(ctx, options, command)
+	close(release)
+	if errServer := <-serverDone; errServer != nil {
+		t.Fatalf("server close error = %v", errServer)
+	}
+	if !errors.Is(errProcess, context.Canceled) {
+		t.Fatalf("processPluginSyncCommand() error = %v, want context.Canceled", errProcess)
+	}
+	if elapsed := time.Since(startedAt); elapsed > time.Second {
+		t.Fatalf("TLS handshake cancellation took %s", elapsed)
+	}
+}
+
+func TestGetPluginTasksRetainsBaseTimeout(t *testing.T) {
+	client, _ := newRedisCommandTestClient(t, func(args []string) string {
+		if len(args) >= 2 && args[1] == redisKeyPluginTasks {
+			time.Sleep(3 * homeRedisTestOperationTimeout)
+			return "$2\r\n[]\r\n"
+		}
+		return "-ERR unexpected command\r\n"
+	})
+	if _, errTasks := client.GetPluginTasks(context.Background()); errTasks == nil {
+		t.Fatal("GetPluginTasks() error = nil, want base read timeout")
+	}
+}
+
 func TestGetPluginSyncRecognizesUnsupportedHomeProtocol(t *testing.T) {
 	tests := []struct {
 		name     string
@@ -513,6 +666,20 @@ func (l *redisCommandLog) Last() []string {
 	return append([]string(nil), l.commands[len(l.commands)-1]...)
 }

+func (l *redisCommandLog) CountKey(key string) int {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	count := 0
+	for _, command := range l.commands {
+		if len(command) >= 2 && command[1] == key {
+			count++
+		}
+	}
+	return count
+}
+
+const homeRedisTestOperationTimeout = 50 * time.Millisecond
+
 func newRedisCommandTestClient(t *testing.T, handler func([]string) string) (*Client, *redisCommandLog) {
 	t.Helper()

@@ -551,13 +718,18 @@ func newRedisCommandTestClient(t *testing.T, handler func([]string) string) (*Cl
 		Port:                    port,
 		DisableClusterDiscovery: true,
 	})
-	client.cmd = redis.NewClient(&redis.Options{
+	options := &redis.Options{
 		Addr:                  listener.Addr().String(),
 		Protocol:              2,
 		DisableIdentity:       true,
+		DialTimeout:           homeRedisTestOperationTimeout,
+		ReadTimeout:           homeRedisTestOperationTimeout,
+		WriteTimeout:          homeRedisTestOperationTimeout,
 		MaxRetries:            -1,
 		ContextTimeoutEnabled: true,
-	})
+	}
+	client.cmdOptions = cloneRedisOptions(options)
+	client.cmd = redis.NewClient(options)
 	t.Cleanup(func() {
 		client.Close()
 	})