[bug] logservice: add timeout for RPC read.

cmd/mo-service/launch.go

@@ -237,7 +237,12 @@ func waitAnyShardReady(client logservice.CNHAKeeperClient) error {
 		if ok, err := func() (bool, error) {
 			details, err := client.GetClusterDetails(ctx)
 			if err != nil {
-				return false, err
+				if errors.Is(err, context.DeadlineExceeded) {
+					logutil.Errorf("wait TN ready timeout: %s", err)
+					return false, err
+				}
+				logutil.Errorf("failed to get cluster details %s", err)
+				return false, nil
 			}
 			for _, store := range details.TNStores {
 				if len(store.Shards) > 0 {

pkg/logservice/client.go

@@ -314,7 +314,15 @@ func connectToLogService(ctx context.Context,
 		addresses[i], addresses[j] = addresses[j], addresses[i]
 	})
 	for _, addr := range addresses {
-		cc, err := getRPCClient(ctx, addr, c.respPool, c.cfg.MaxMessageSize, cfg.EnableCompress, cfg.Tag)
+		cc, err := getRPCClient(
+			ctx,
+			addr,
+			c.respPool,
+			c.cfg.MaxMessageSize,
+			cfg.EnableCompress,
+			0,
+			cfg.Tag,
+		)
 		if err != nil {
 			e = err
 			continue
@@ -511,6 +519,7 @@ func getRPCClient(
 	pool *sync.Pool,
 	maxMessageSize int,
 	enableCompress bool,
+	readTimeout time.Duration,
 	tag ...string) (morpc.RPCClient, error) {
 	mf := func() morpc.Message {
 		return pool.Get().(*RPCResponse)
@@ -521,6 +530,7 @@ func getRPCClient(
 		morpc.WithBackendConnectTimeout(time.Second),
 		morpc.WithBackendHasPayloadResponse(),
 		morpc.WithBackendLogger(logutil.GetGlobalLogger().Named("hakeeper-client-backend")),
+		morpc.WithBackendReadTimeout(readTimeout),
 	}
 	backendOpts = append(backendOpts, GetBackendOptions(ctx)...)
 

pkg/logservice/hakeeper_client.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"math/rand"
 	"sync"
+	"time"
 
 	"go.uber.org/zap"
 
@@ -30,6 +31,10 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/util/trace"
 )
 
+const (
+	defaultBackendReadTimeout = time.Second * 8
+)
+
 type basicHAKeeperClient interface {
 	// Close closes the hakeeper client.
 	Close() error
@@ -571,7 +576,15 @@ func connectToHAKeeper(ctx context.Context,
 		addresses[i], addresses[j] = addresses[j], addresses[i]
 	})
 	for _, addr := range addresses {
-		cc, err := getRPCClient(ctx, addr, c.respPool, defaultMaxMessageSize, cfg.EnableCompress, "connectToHAKeeper")
+		cc, err := getRPCClient(
+			ctx,
+			addr,
+			c.respPool,
+			defaultMaxMessageSize,
+			cfg.EnableCompress,
+			defaultBackendReadTimeout,
+			"connectToHAKeeper",
+		)
 		if err != nil {
 			e = err
 			continue

pkg/logservice/hakeeper_client_test.go

@@ -351,7 +351,14 @@ func testNotHAKeeperErrorIsHandled(t *testing.T, fn func(*testing.T, *managedHAK
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	cc, err := getRPCClient(ctx, cfg1.LogServiceServiceAddr(), c.respPool, defaultMaxMessageSize, false)
+	cc, err := getRPCClient(
+		ctx,
+		cfg1.LogServiceServiceAddr(),
+		c.respPool,
+		defaultMaxMessageSize,
+		false,
+		0,
+	)
 	require.NoError(t, err)
 	c.addr = cfg1.LogServiceServiceAddr()
 	c.client = cc

pkg/logservice/shardinfo.go

@@ -45,7 +45,15 @@ func GetShardInfo(address string, shardID uint64) (ShardInfo, bool, error) {
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
-	cc, err := getRPCClient(ctx, address, respPool, defaultMaxMessageSize, false, "GetShardInfo")
+	cc, err := getRPCClient(
+		ctx,
+		address,
+		respPool,
+		defaultMaxMessageSize,
+		false,
+		0,
+		"GetShardInfo",
+	)
 	if err != nil {
 		return ShardInfo{}, false, err
 	}