From b651a1a8fd8ea1a43d061a08b42cef8fa3160179 Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Sun, 19 Jul 2026 01:41:18 +0800 Subject: [PATCH] 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. --- internal/home/client.go | 138 +++++++++++++++++++++++++-- internal/home/client_test.go | 176 ++++++++++++++++++++++++++++++++++- 2 files changed, 306 insertions(+), 8 deletions(-) diff --git a/internal/home/client.go b/internal/home/client.go index 279f5e83..af001a71 100644 --- a/internal/home/client.go +++ b/internal/home/client.go @@ -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 { diff --git a/internal/home/client_test.go b/internal/home/client_test.go index b9480c08..40110d08 100644 --- a/internal/home/client_test.go +++ b/internal/home/client_test.go @@ -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() })