From 0ac1864557830e1cf9e6326a09fcaa79175ed37d Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 1 Jul 2026 11:25:48 +0800 Subject: [PATCH 1/9] fix(logservice): fallback from stale replica addresses (cherry picked from commit e21b43d1415ba274d54c061da03db666f2cb43ef) --- pkg/logservice/client.go | 179 ++++++++++++++++++++++++++-------- pkg/logservice/client_test.go | 95 ++++++++++++++++++ 2 files changed, 235 insertions(+), 39 deletions(-) diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go index 28ea88bfbb41a..edc117b6b83d0 100644 --- a/pkg/logservice/client.go +++ b/pkg/logservice/client.go @@ -37,6 +37,11 @@ const ( defaultWriteSocketSize = 64 * 1024 ) +var ( + logServiceConnectAttemptTimeout = 5 * time.Second + logServiceConnectFallbackDelay = 300 * time.Millisecond +) + // IsTempError returns a boolean value indicating whether the specified error // is a temp error that worth to be retried, e.g. timeouts, temp network // issues. Non-temp error caused by program logics rather than some external @@ -490,7 +495,19 @@ func connectToLogService( if len(targets) == 0 { return nil, nil } + addresses := append([]string{}, targets...) + rand.Shuffle(len(addresses), func(i, j int) { + addresses[i], addresses[j] = addresses[j], addresses[i] + }) + return connectToLogServiceAddresses(ctx, sid, addresses, cfg) +} +func connectToLogServiceAddress( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, +) (*client, error) { pool := &sync.Pool{} pool.New = func() interface{} { return &RPCRequest{pool: pool} @@ -504,51 +521,135 @@ func connectToLogService( pool: pool, respPool: respPool, } - var e error - addresses := append([]string{}, targets...) - rand.Shuffle(len(cfg.ServiceAddresses), func(i, j int) { - addresses[i], addresses[j] = addresses[j], addresses[i] - }) - for _, addr := range addresses { - cc, err := getRPCClient( - ctx, - sid, - addr, - c.respPool, - c.cfg.MaxMessageSize, - cfg.EnableCompress, - 0, - cfg.Tag, - ) - if err != nil { - e = err - continue + cc, err := getRPCClient( + ctx, + sid, + addr, + c.respPool, + c.cfg.MaxMessageSize, + cfg.EnableCompress, + time.Second*10, + cfg.Tag, + ) + if err != nil { + return nil, err + } + c.addr = addr + c.client = cc + if cfg.ReadOnly { + if err := c.connectReadOnly(ctx); err != nil { + if closeErr := c.close(); closeErr != nil { + logutil.Error("failed to close the client", zap.Error(closeErr)) + } + return nil, err + } + return c, nil + } + // TODO: add a test to check whether it works when there is no truncated + // LSN known to the logservice. + if err := c.connectReadWrite(ctx); err != nil { + if closeErr := c.close(); closeErr != nil { + logutil.Error("failed to close the client", zap.Error(closeErr)) } - c.addr = addr - c.client = cc - if cfg.ReadOnly { - if err := c.connectReadOnly(ctx); err == nil { - return c, nil - } else { - if err := c.close(); err != nil { - logutil.Error("failed to close the client", zap.Error(err)) + return nil, err + } + return c, nil +} + +func connectToLogServiceAddresses( + ctx context.Context, + sid string, + addresses []string, + cfg ClientConfig, +) (*client, error) { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + type result struct { + client *client + err error + } + results := make(chan result, len(addresses)) + var wg sync.WaitGroup + launch := func(addr string) { + wg.Add(1) + go func() { + defer wg.Done() + attemptCtx, attemptCancel := context.WithTimeout(ctx, logServiceConnectAttemptTimeout) + c, err := connectToLogServiceAddress(attemptCtx, sid, addr, cfg) + attemptCancel() + if err == nil { + select { + case results <- result{client: c}: + case <-ctx.Done(): + if closeErr := c.close(); closeErr != nil { + logutil.Error("failed to close the client", zap.Error(closeErr)) + } } - e = err + return } - } else { - // TODO: add a test to check whether it works when there is no truncated - // LSN known to the logservice. - if err := c.connectReadWrite(ctx); err == nil { - return c, nil - } else { - if err := c.close(); err != nil { - logutil.Error("failed to close the client", zap.Error(err)) - } - e = err + select { + case results <- result{err: err}: + case <-ctx.Done(): } + }() + } + + var lastErr error + launched, active := 0, 0 + for active > 0 || launched < len(addresses) { + if active == 0 { + launch(addresses[launched]) + launched++ + active++ + continue + } + + var fallback <-chan time.Time + var timer *time.Timer + if launched < len(addresses) { + timer = time.NewTimer(logServiceConnectFallbackDelay) + fallback = timer.C + } + + select { + case r := <-results: + stopConnectFallbackTimer(timer) + active-- + if r.client != nil { + cancel() + wg.Wait() + return r.client, nil + } + lastErr = r.err + case <-fallback: + launch(addresses[launched]) + launched++ + active++ + case <-ctx.Done(): + stopConnectFallbackTimer(timer) + cancel() + wg.Wait() + if lastErr != nil { + return nil, lastErr + } + return nil, ctx.Err() + } + } + wg.Wait() + return nil, lastErr +} + +func stopConnectFallbackTimer(timer *time.Timer) { + if timer == nil { + return + } + if !timer.Stop() { + select { + case <-timer.C: + default: } } - return nil, e } func (c *client) close() error { diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index 1eea2f9663a1c..92c7a086081a3 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -19,6 +19,8 @@ import ( "crypto/rand" "fmt" "math" + "net" + "sync" "testing" "time" @@ -147,6 +149,99 @@ func TestClientCanBeConnectedByReverseProxy(t *testing.T) { }() } +func TestClientCreationFallsBackFromUnreachableReplicaAddress(t *testing.T) { + defer leaktest.AfterTest(t)() + var cfg Config + genCfg := func() Config { + cfg = getServiceTestConfig() + return cfg + } + defer vfs.ReportLeakedFD(cfg.FS, t) + service, err := NewServiceWithRetry( + genCfg, + newFS(), + nil, + WithBackendFilter(func(msg morpc.Message, backendAddr string) bool { + return true + }), + ) + require.NoError(t, err) + defer func() { + assert.NoError(t, service.Close()) + }() + + init := make(map[uint64]string) + init[2] = service.ID() + require.NoError(t, service.store.startReplica(1, 2, init, false)) + + hangingAddress, cleanup := startHangingTCPServer(t) + defer cleanup() + + scfg := ClientConfig{ + LogShardID: 1, + TNReplicaID: 2, + MaxMessageSize: defaultMaxMessageSize, + } + + origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectFallbackDelay = 50 * time.Millisecond + defer func() { logServiceConnectFallbackDelay = origFallbackDelay }() + + timeout := 5 * time.Second + ctx, cancel := context.WithTimeout(context.Background(), timeout) + defer cancel() + start := time.Now() + c, err := connectToLogServiceAddresses(ctx, "", []string{ + hangingAddress, + cfg.LogServiceServiceAddr(), + }, scfg) + elapsed := time.Since(start) + + require.NoError(t, err) + require.NotNil(t, c) + defer func() { + assert.NoError(t, c.close()) + }() + assert.Less(t, elapsed, time.Second) +} + +func startHangingTCPServer(t *testing.T) (string, func()) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + done := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for { + conn, err := ln.Accept() + if err != nil { + return + } + wg.Add(1) + go func() { + defer wg.Done() + defer func() { + assert.NoError(t, conn.Close()) + }() + <-done + }() + } + }() + + var once sync.Once + cleanup := func() { + once.Do(func() { + assert.NoError(t, ln.Close()) + close(done) + wg.Wait() + }) + } + return ln.Addr().String(), cleanup +} + func TestClientGetTSOTimestamp(t *testing.T) { fn := func(t *testing.T, s *Service, cfg ClientConfig, c Client) { ctx, cancel := context.WithTimeout(context.Background(), time.Second) From e3b9a4a5d73365a996157e1854815123278ed647 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Wed, 1 Jul 2026 20:22:37 +0800 Subject: [PATCH 2/9] fix(logservice): address review findings for stale replica fallback - Restore getRPCClient read timeout to 0 (per-attempt ctx already bounds the connection handshake; read timeout can reset idle/in-flight clients) - Preserve leader-first ordering for discovery/reverse-proxy by extracting shardInfoReplicaAddresses and bypassing the shuffle in connectToLogService - Add closeUnusedConnectResultClients to drain and close all non-winning successful clients after winner selection - Add regression tests: leader-first ordering, all-addresses-fail, parent context cancellation, multiple-success loser cleanup, and fallback timer Co-Authored-By: Claude --- pkg/logservice/client.go | 47 +++++++-- pkg/logservice/client_test.go | 189 ++++++++++++++++++++++++++++++++++ 2 files changed, 225 insertions(+), 11 deletions(-) diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go index edc117b6b83d0..d46e2026575c4 100644 --- a/pkg/logservice/client.go +++ b/pkg/logservice/client.go @@ -42,6 +42,8 @@ var ( logServiceConnectFallbackDelay = 300 * time.Millisecond ) +var connectToLogServiceAddressFn = connectToLogServiceAddress + // IsTempError returns a boolean value indicating whether the specified error // is a temp error that worth to be retried, e.g. timeouts, temp network // issues. Non-temp error caused by program logics rather than some external @@ -473,7 +475,12 @@ func connectToLogServiceByReverseProxy( if !ok { return nil, moerr.NewLogServiceNotReady(ctx) } - addresses := make([]string, 0) + addresses := shardInfoReplicaAddresses(si) + return connectToLogServiceAddresses(ctx, sid, addresses, cfg) +} + +func shardInfoReplicaAddresses(si ShardInfo) []string { + addresses := make([]string, 0, len(si.Replicas)) leaderAddress, ok := si.Replicas[si.ReplicaID] if ok { addresses = append(addresses, leaderAddress) @@ -483,7 +490,7 @@ func connectToLogServiceByReverseProxy( addresses = append(addresses, address) } } - return connectToLogService(ctx, sid, addresses, cfg) + return addresses } func connectToLogService( @@ -528,7 +535,7 @@ func connectToLogServiceAddress( c.respPool, c.cfg.MaxMessageSize, cfg.EnableCompress, - time.Second*10, + 0, cfg.Tag, ) if err != nil { @@ -556,6 +563,11 @@ func connectToLogServiceAddress( return c, nil } +type connectResult struct { + client *client + err error +} + func connectToLogServiceAddresses( ctx context.Context, sid string, @@ -565,22 +577,18 @@ func connectToLogServiceAddresses( ctx, cancel := context.WithCancel(ctx) defer cancel() - type result struct { - client *client - err error - } - results := make(chan result, len(addresses)) + results := make(chan connectResult, len(addresses)) var wg sync.WaitGroup launch := func(addr string) { wg.Add(1) go func() { defer wg.Done() attemptCtx, attemptCancel := context.WithTimeout(ctx, logServiceConnectAttemptTimeout) - c, err := connectToLogServiceAddress(attemptCtx, sid, addr, cfg) + c, err := connectToLogServiceAddressFn(attemptCtx, sid, addr, cfg) attemptCancel() if err == nil { select { - case results <- result{client: c}: + case results <- connectResult{client: c}: case <-ctx.Done(): if closeErr := c.close(); closeErr != nil { logutil.Error("failed to close the client", zap.Error(closeErr)) @@ -589,7 +597,7 @@ func connectToLogServiceAddresses( return } select { - case results <- result{err: err}: + case results <- connectResult{err: err}: case <-ctx.Done(): } }() @@ -619,6 +627,7 @@ func connectToLogServiceAddresses( if r.client != nil { cancel() wg.Wait() + closeUnusedConnectResultClients(results) return r.client, nil } lastErr = r.err @@ -630,6 +639,7 @@ func connectToLogServiceAddresses( stopConnectFallbackTimer(timer) cancel() wg.Wait() + closeUnusedConnectResultClients(results) if lastErr != nil { return nil, lastErr } @@ -640,6 +650,21 @@ func connectToLogServiceAddresses( return nil, lastErr } +func closeUnusedConnectResultClients(results <-chan connectResult) { + for { + select { + case r := <-results: + if r.client != nil { + if closeErr := r.client.close(); closeErr != nil { + logutil.Error("failed to close the client", zap.Error(closeErr)) + } + } + default: + return + } + } +} + func stopConnectFallbackTimer(timer *time.Timer) { if timer == nil { return diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index 92c7a086081a3..a55a553a62868 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -205,6 +205,195 @@ func TestClientCreationFallsBackFromUnreachableReplicaAddress(t *testing.T) { assert.Less(t, elapsed, time.Second) } +func TestShardInfoReplicaAddressesKeepsLeaderFirst(t *testing.T) { + addresses := shardInfoReplicaAddresses(ShardInfo{ + ReplicaID: 2, + Replicas: map[uint64]string{ + 1: "replica-1", + 2: "leader", + 3: "replica-3", + }, + }) + require.Len(t, addresses, 3) + assert.Equal(t, "leader", addresses[0]) + assert.ElementsMatch(t, []string{"leader", "replica-1", "replica-3"}, addresses) +} + +func TestConnectToLogServiceWithNoTargets(t *testing.T) { + c, err := connectToLogService(context.Background(), "", nil, ClientConfig{}) + require.NoError(t, err) + require.Nil(t, c) +} + +func TestConnectToLogServiceAddressesReturnsLastError(t *testing.T) { + origAttemptTimeout := logServiceConnectAttemptTimeout + origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectAttemptTimeout = 50 * time.Millisecond + logServiceConnectFallbackDelay = time.Millisecond + defer func() { + logServiceConnectAttemptTimeout = origAttemptTimeout + logServiceConnectFallbackDelay = origFallbackDelay + }() + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + c, err := connectToLogServiceAddresses(ctx, "", []string{ + "127.0.0.1:1", + "127.0.0.1:2", + }, ClientConfig{MaxMessageSize: defaultMaxMessageSize}) + require.Error(t, err) + require.Nil(t, c) +} + +func TestConnectToLogServiceAddressesReturnsContextError(t *testing.T) { + hangingAddress, cleanup := startHangingTCPServer(t) + defer cleanup() + + origAttemptTimeout := logServiceConnectAttemptTimeout + origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectAttemptTimeout = time.Second + logServiceConnectFallbackDelay = time.Second + defer func() { + logServiceConnectAttemptTimeout = origAttemptTimeout + logServiceConnectFallbackDelay = origFallbackDelay + }() + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + c, err := connectToLogServiceAddresses( + ctx, + "", + []string{hangingAddress}, + ClientConfig{MaxMessageSize: defaultMaxMessageSize}, + ) + require.Error(t, err) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Nil(t, c) +} + +func TestConnectToLogServiceAddressesClosesSuccessfulLosers(t *testing.T) { + origConnect := connectToLogServiceAddressFn + origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectFallbackDelay = time.Millisecond + defer func() { + connectToLogServiceAddressFn = origConnect + logServiceConnectFallbackDelay = origFallbackDelay + }() + + called := make(chan string, 2) + release := make(chan struct{}) + closeCount := 0 + var closeMu sync.Mutex + connectToLogServiceAddressFn = func( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, + ) (*client, error) { + called <- addr + select { + case <-release: + case <-ctx.Done(): + return nil, ctx.Err() + } + return &client{ + client: &closeTrackingRPCClient{ + closeMu: &closeMu, + closeCount: &closeCount, + }, + }, nil + } + + done := make(chan *client, 1) + errs := make(chan error, 1) + go func() { + c, err := connectToLogServiceAddresses( + context.Background(), + "", + []string{"first", "second"}, + ClientConfig{}, + ) + if err != nil { + errs <- err + return + } + done <- c + }() + + require.Equal(t, "first", readCalledAddress(t, called)) + require.Equal(t, "second", readCalledAddress(t, called)) + close(release) + + var c *client + select { + case c = <-done: + case err := <-errs: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("timed out waiting for connection result") + } + require.NotNil(t, c) + + closeMu.Lock() + assert.Equal(t, 1, closeCount) + closeMu.Unlock() + + require.NoError(t, c.close()) + closeMu.Lock() + assert.Equal(t, 2, closeCount) + closeMu.Unlock() +} + +func TestStopConnectFallbackTimer(t *testing.T) { + stopConnectFallbackTimer(nil) + + timer := time.NewTimer(time.Hour) + stopConnectFallbackTimer(timer) + + expired := time.NewTimer(time.Nanosecond) + time.Sleep(time.Millisecond) + stopConnectFallbackTimer(expired) +} + +func readCalledAddress(t *testing.T, called <-chan string) string { + t.Helper() + select { + case addr := <-called: + return addr + case <-time.After(time.Second): + t.Fatal("timed out waiting for connection attempt") + } + return "" +} + +type closeTrackingRPCClient struct { + closeMu *sync.Mutex + closeCount *int +} + +func (c *closeTrackingRPCClient) Send(context.Context, string, morpc.Message) (*morpc.Future, error) { + return nil, errors.New("unexpected send") +} + +func (c *closeTrackingRPCClient) NewStream(context.Context, string, bool) (morpc.Stream, error) { + return nil, errors.New("unexpected stream") +} + +func (c *closeTrackingRPCClient) Ping(context.Context, string) error { + return errors.New("unexpected ping") +} + +func (c *closeTrackingRPCClient) Close() error { + c.closeMu.Lock() + defer c.closeMu.Unlock() + *c.closeCount++ + return nil +} + +func (c *closeTrackingRPCClient) CloseBackend() error { + return nil +} + func startHangingTCPServer(t *testing.T) (string, func()) { t.Helper() ln, err := net.Listen("tcp", "127.0.0.1:0") From 7aad95e7984c5c98e814185a42b4a7396bb0a393 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Thu, 2 Jul 2026 10:09:16 +0800 Subject: [PATCH 3/9] fix(logservice): close losing fallback clients --- pkg/logservice/client_test.go | 42 +++++++++++++++++++++++++++++++++-- 1 file changed, 40 insertions(+), 2 deletions(-) diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index a55a553a62868..f70ac3b546be3 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -375,7 +375,7 @@ func (c *closeTrackingRPCClient) Send(context.Context, string, morpc.Message) (* return nil, errors.New("unexpected send") } -func (c *closeTrackingRPCClient) NewStream(context.Context, string, bool) (morpc.Stream, error) { +func (c *closeTrackingRPCClient) NewStream(string, bool) (morpc.Stream, error) { return nil, errors.New("unexpected stream") } @@ -386,7 +386,7 @@ func (c *closeTrackingRPCClient) Ping(context.Context, string) error { func (c *closeTrackingRPCClient) Close() error { c.closeMu.Lock() defer c.closeMu.Unlock() - *c.closeCount++ + (*c.closeCount)++ return nil } @@ -431,6 +431,44 @@ func startHangingTCPServer(t *testing.T) (string, func()) { return ln.Addr().String(), cleanup } +type closeTrackingRPCClient struct { + mu sync.Mutex + closed int +} + +func (c *closeTrackingRPCClient) Send( + context.Context, + string, + morpc.Message, +) (*morpc.Future, error) { + return nil, nil +} + +func (c *closeTrackingRPCClient) NewStream(string, bool) (morpc.Stream, error) { + return nil, nil +} + +func (c *closeTrackingRPCClient) Ping(context.Context, string) error { + return nil +} + +func (c *closeTrackingRPCClient) Close() error { + c.mu.Lock() + defer c.mu.Unlock() + c.closed++ + return nil +} + +func (c *closeTrackingRPCClient) CloseBackend() error { + return nil +} + +func (c *closeTrackingRPCClient) closeCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.closed +} + func TestClientGetTSOTimestamp(t *testing.T) { fn := func(t *testing.T, s *Service, cfg ClientConfig, c Client) { ctx, cancel := context.WithTimeout(context.Background(), time.Second) From f0b6d82ad14cf6c83a0ee756c57065a8fa12c7d2 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Thu, 2 Jul 2026 10:31:51 +0800 Subject: [PATCH 4/9] fix(logservice): remove duplicate close tracking test client --- pkg/logservice/client_test.go | 38 ----------------------------------- 1 file changed, 38 deletions(-) diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index f70ac3b546be3..be13c7bc4a38d 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -431,44 +431,6 @@ func startHangingTCPServer(t *testing.T) (string, func()) { return ln.Addr().String(), cleanup } -type closeTrackingRPCClient struct { - mu sync.Mutex - closed int -} - -func (c *closeTrackingRPCClient) Send( - context.Context, - string, - morpc.Message, -) (*morpc.Future, error) { - return nil, nil -} - -func (c *closeTrackingRPCClient) NewStream(string, bool) (morpc.Stream, error) { - return nil, nil -} - -func (c *closeTrackingRPCClient) Ping(context.Context, string) error { - return nil -} - -func (c *closeTrackingRPCClient) Close() error { - c.mu.Lock() - defer c.mu.Unlock() - c.closed++ - return nil -} - -func (c *closeTrackingRPCClient) CloseBackend() error { - return nil -} - -func (c *closeTrackingRPCClient) closeCount() int { - c.mu.Lock() - defer c.mu.Unlock() - return c.closed -} - func TestClientGetTSOTimestamp(t *testing.T) { fn := func(t *testing.T, s *Service, cfg ClientConfig, c Client) { ctx, cancel := context.WithTimeout(context.Background(), time.Second) From 3f3c7a283e667a75e333466c1360b24c612894c3 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Fri, 3 Jul 2026 16:05:40 +0800 Subject: [PATCH 5/9] fix(logservice): preserve terminal context errors --- pkg/logservice/client.go | 16 ++++++-- pkg/logservice/client_test.go | 72 +++++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 3 deletions(-) diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go index d46e2026575c4..8ef7b91bc333f 100644 --- a/pkg/logservice/client.go +++ b/pkg/logservice/client.go @@ -577,6 +577,10 @@ func connectToLogServiceAddresses( ctx, cancel := context.WithCancel(ctx) defer cancel() + if err := ctx.Err(); err != nil { + return nil, err + } + results := make(chan connectResult, len(addresses)) var wg sync.WaitGroup launch := func(addr string) { @@ -607,6 +611,9 @@ func connectToLogServiceAddresses( launched, active := 0, 0 for active > 0 || launched < len(addresses) { if active == 0 { + if err := ctx.Err(); err != nil { + return nil, err + } launch(addresses[launched]) launched++ active++ @@ -632,6 +639,12 @@ func connectToLogServiceAddresses( } lastErr = r.err case <-fallback: + if err := ctx.Err(); err != nil { + cancel() + wg.Wait() + closeUnusedConnectResultClients(results) + return nil, err + } launch(addresses[launched]) launched++ active++ @@ -640,9 +653,6 @@ func connectToLogServiceAddresses( cancel() wg.Wait() closeUnusedConnectResultClients(results) - if lastErr != nil { - return nil, lastErr - } return nil, ctx.Err() } } diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index be13c7bc4a38d..8bf98a6faf6e3 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -271,6 +271,78 @@ func TestConnectToLogServiceAddressesReturnsContextError(t *testing.T) { require.Nil(t, c) } +func TestConnectToLogServiceAddressesReturnsContextErrorAfterPreviousFailure(t *testing.T) { + origConnect := connectToLogServiceAddressFn + origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectFallbackDelay = time.Millisecond + defer func() { + connectToLogServiceAddressFn = origConnect + logServiceConnectFallbackDelay = origFallbackDelay + }() + + firstErr := errors.New("first failed") + connectToLogServiceAddressFn = func( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, + ) (*client, error) { + if addr == "first" { + return nil, firstErr + } + <-ctx.Done() + time.Sleep(10 * time.Millisecond) + return nil, ctx.Err() + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + c, err := connectToLogServiceAddresses( + ctx, + "", + []string{"first", "second"}, + ClientConfig{}, + ) + require.Error(t, err) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Nil(t, c) +} + +func TestConnectToLogServiceAddressesWithCanceledContextDoesNotDial(t *testing.T) { + origConnect := connectToLogServiceAddressFn + defer func() { + connectToLogServiceAddressFn = origConnect + }() + + called := make(chan struct{}, 1) + connectToLogServiceAddressFn = func( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, + ) (*client, error) { + called <- struct{}{} + return nil, errors.New("unexpected dial") + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + c, err := connectToLogServiceAddresses( + ctx, + "", + []string{"first"}, + ClientConfig{}, + ) + require.Error(t, err) + require.ErrorIs(t, err, context.Canceled) + require.Nil(t, c) + select { + case <-called: + t.Fatal("unexpected connection attempt") + default: + } +} + func TestConnectToLogServiceAddressesClosesSuccessfulLosers(t *testing.T) { origConnect := connectToLogServiceAddressFn origFallbackDelay := logServiceConnectFallbackDelay From b4b67948314d7cc4258d5d5a7c232621c3e3d059 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Fri, 3 Jul 2026 16:40:19 +0800 Subject: [PATCH 6/9] fix(logservice): include last connect error on timeout --- pkg/logservice/client.go | 9 ++++++++- pkg/logservice/client_test.go | 1 + 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go index 8ef7b91bc333f..df8c2db77b6e6 100644 --- a/pkg/logservice/client.go +++ b/pkg/logservice/client.go @@ -653,13 +653,20 @@ func connectToLogServiceAddresses( cancel() wg.Wait() closeUnusedConnectResultClients(results) - return nil, ctx.Err() + return nil, connectContextError(ctx.Err(), lastErr) } } wg.Wait() return nil, lastErr } +func connectContextError(ctxErr error, lastErr error) error { + if ctxErr == nil || lastErr == nil { + return ctxErr + } + return errors.Wrapf(ctxErr, "last logservice connect error: %v", lastErr) +} + func closeUnusedConnectResultClients(results <-chan connectResult) { for { select { diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index 8bf98a6faf6e3..b92aa8e16be29 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -305,6 +305,7 @@ func TestConnectToLogServiceAddressesReturnsContextErrorAfterPreviousFailure(t * ) require.Error(t, err) require.ErrorIs(t, err, context.DeadlineExceeded) + require.ErrorContains(t, err, firstErr.Error()) require.Nil(t, c) } From 40cb45a4468b41ce111ac4e2ae688d41df8e6312 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Fri, 3 Jul 2026 17:41:35 +0800 Subject: [PATCH 7/9] fix logservice connect attempt timeout cause --- pkg/logservice/client.go | 8 ++++++-- pkg/logservice/client_test.go | 37 +++++++++++++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go index df8c2db77b6e6..102c1e04aa996 100644 --- a/pkg/logservice/client.go +++ b/pkg/logservice/client.go @@ -587,7 +587,11 @@ func connectToLogServiceAddresses( wg.Add(1) go func() { defer wg.Done() - attemptCtx, attemptCancel := context.WithTimeout(ctx, logServiceConnectAttemptTimeout) + attemptCtx, attemptCancel := context.WithTimeoutCause( + ctx, + logServiceConnectAttemptTimeout, + moerr.CauseNewLogServiceClient, + ) c, err := connectToLogServiceAddressFn(attemptCtx, sid, addr, cfg) attemptCancel() if err == nil { @@ -601,7 +605,7 @@ func connectToLogServiceAddresses( return } select { - case results <- connectResult{err: err}: + case results <- connectResult{err: moerr.AttachCause(attemptCtx, err)}: case <-ctx.Done(): } }() diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index b92aa8e16be29..9a6f6178b58c9 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -309,6 +309,43 @@ func TestConnectToLogServiceAddressesReturnsContextErrorAfterPreviousFailure(t * require.Nil(t, c) } +func TestConnectToLogServiceAddressesReturnsAttemptTimeoutCause(t *testing.T) { + origConnect := connectToLogServiceAddressFn + origAttemptTimeout := logServiceConnectAttemptTimeout + origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectAttemptTimeout = 20 * time.Millisecond + logServiceConnectFallbackDelay = time.Millisecond + defer func() { + connectToLogServiceAddressFn = origConnect + logServiceConnectAttemptTimeout = origAttemptTimeout + logServiceConnectFallbackDelay = origFallbackDelay + }() + + connectToLogServiceAddressFn = func( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, + ) (*client, error) { + <-ctx.Done() + return nil, ctx.Err() + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + c, err := connectToLogServiceAddresses( + ctx, + "", + []string{"first", "second"}, + ClientConfig{}, + ) + require.Error(t, err) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.ErrorIs(t, err, moerr.CauseNewLogServiceClient) + require.NoError(t, ctx.Err()) + require.Nil(t, c) +} + func TestConnectToLogServiceAddressesWithCanceledContextDoesNotDial(t *testing.T) { origConnect := connectToLogServiceAddressFn defer func() { From 26d56ade7a46af45cd208d632c3b6863c4be9dad Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Mon, 6 Jul 2026 12:03:17 +0800 Subject: [PATCH 8/9] fix(logservice): preserve caller connect timeout --- pkg/logservice/client.go | 13 +++---------- pkg/logservice/client_test.go | 24 +++++++++++------------- 2 files changed, 14 insertions(+), 23 deletions(-) diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go index 102c1e04aa996..e8de5152b5372 100644 --- a/pkg/logservice/client.go +++ b/pkg/logservice/client.go @@ -38,8 +38,7 @@ const ( ) var ( - logServiceConnectAttemptTimeout = 5 * time.Second - logServiceConnectFallbackDelay = 300 * time.Millisecond + logServiceConnectFallbackDelay = 300 * time.Millisecond ) var connectToLogServiceAddressFn = connectToLogServiceAddress @@ -587,13 +586,7 @@ func connectToLogServiceAddresses( wg.Add(1) go func() { defer wg.Done() - attemptCtx, attemptCancel := context.WithTimeoutCause( - ctx, - logServiceConnectAttemptTimeout, - moerr.CauseNewLogServiceClient, - ) - c, err := connectToLogServiceAddressFn(attemptCtx, sid, addr, cfg) - attemptCancel() + c, err := connectToLogServiceAddressFn(ctx, sid, addr, cfg) if err == nil { select { case results <- connectResult{client: c}: @@ -605,7 +598,7 @@ func connectToLogServiceAddresses( return } select { - case results <- connectResult{err: moerr.AttachCause(attemptCtx, err)}: + case results <- connectResult{err: moerr.AttachCause(ctx, err)}: case <-ctx.Done(): } }() diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index 9a6f6178b58c9..730d78905c997 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -226,12 +226,9 @@ func TestConnectToLogServiceWithNoTargets(t *testing.T) { } func TestConnectToLogServiceAddressesReturnsLastError(t *testing.T) { - origAttemptTimeout := logServiceConnectAttemptTimeout origFallbackDelay := logServiceConnectFallbackDelay - logServiceConnectAttemptTimeout = 50 * time.Millisecond logServiceConnectFallbackDelay = time.Millisecond defer func() { - logServiceConnectAttemptTimeout = origAttemptTimeout logServiceConnectFallbackDelay = origFallbackDelay }() @@ -249,12 +246,9 @@ func TestConnectToLogServiceAddressesReturnsContextError(t *testing.T) { hangingAddress, cleanup := startHangingTCPServer(t) defer cleanup() - origAttemptTimeout := logServiceConnectAttemptTimeout origFallbackDelay := logServiceConnectFallbackDelay - logServiceConnectAttemptTimeout = time.Second logServiceConnectFallbackDelay = time.Second defer func() { - logServiceConnectAttemptTimeout = origAttemptTimeout logServiceConnectFallbackDelay = origFallbackDelay }() @@ -309,30 +303,33 @@ func TestConnectToLogServiceAddressesReturnsContextErrorAfterPreviousFailure(t * require.Nil(t, c) } -func TestConnectToLogServiceAddressesReturnsAttemptTimeoutCause(t *testing.T) { +func TestConnectToLogServiceAddressesUsesCallerContextForAttempts(t *testing.T) { origConnect := connectToLogServiceAddressFn - origAttemptTimeout := logServiceConnectAttemptTimeout origFallbackDelay := logServiceConnectFallbackDelay - logServiceConnectAttemptTimeout = 20 * time.Millisecond logServiceConnectFallbackDelay = time.Millisecond defer func() { connectToLogServiceAddressFn = origConnect - logServiceConnectAttemptTimeout = origAttemptTimeout logServiceConnectFallbackDelay = origFallbackDelay }() + deadlines := make(chan time.Time, 2) connectToLogServiceAddressFn = func( ctx context.Context, sid string, addr string, cfg ClientConfig, ) (*client, error) { + deadline, ok := ctx.Deadline() + require.True(t, ok) + deadlines <- deadline <-ctx.Done() return nil, ctx.Err() } - ctx, cancel := context.WithTimeout(context.Background(), time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) defer cancel() + expectedDeadline, ok := ctx.Deadline() + require.True(t, ok) c, err := connectToLogServiceAddresses( ctx, "", @@ -341,9 +338,10 @@ func TestConnectToLogServiceAddressesReturnsAttemptTimeoutCause(t *testing.T) { ) require.Error(t, err) require.ErrorIs(t, err, context.DeadlineExceeded) - require.ErrorIs(t, err, moerr.CauseNewLogServiceClient) - require.NoError(t, ctx.Err()) require.Nil(t, c) + for i := 0; i < 2; i++ { + require.WithinDuration(t, expectedDeadline, <-deadlines, time.Millisecond) + } } func TestConnectToLogServiceAddressesWithCanceledContextDoesNotDial(t *testing.T) { From 5fe0ab4546b56d8e298bceaefc0c83795c5597e4 Mon Sep 17 00:00:00 2001 From: jiangxinmeng Date: Mon, 6 Jul 2026 17:29:42 +0800 Subject: [PATCH 9/9] fix(logservice): avoid blocking on stale connect attempts --- pkg/logservice/client.go | 51 ++++++++--- pkg/logservice/client_test.go | 155 +++++++++++++++++++++++++++++++++- 2 files changed, 191 insertions(+), 15 deletions(-) diff --git a/pkg/logservice/client.go b/pkg/logservice/client.go index e8de5152b5372..62ae0e49a7ee8 100644 --- a/pkg/logservice/client.go +++ b/pkg/logservice/client.go @@ -38,7 +38,8 @@ const ( ) var ( - logServiceConnectFallbackDelay = 300 * time.Millisecond + logServiceConnectAttemptTimeout = 5 * time.Second + logServiceConnectFallbackDelay = 300 * time.Millisecond ) var connectToLogServiceAddressFn = connectToLogServiceAddress @@ -586,8 +587,17 @@ func connectToLogServiceAddresses( wg.Add(1) go func() { defer wg.Done() - c, err := connectToLogServiceAddressFn(ctx, sid, addr, cfg) + attemptCtx, attemptCancel := connectAttemptContext(ctx) + defer attemptCancel() + + c, err := connectToLogServiceAddressFn(attemptCtx, sid, addr, cfg) if err == nil { + if err := ctx.Err(); err != nil { + if closeErr := c.close(); closeErr != nil { + logutil.Error("failed to close the client", zap.Error(closeErr)) + } + return + } select { case results <- connectResult{client: c}: case <-ctx.Done(): @@ -598,7 +608,7 @@ func connectToLogServiceAddresses( return } select { - case results <- connectResult{err: moerr.AttachCause(ctx, err)}: + case results <- connectResult{err: moerr.AttachCause(attemptCtx, err)}: case <-ctx.Done(): } }() @@ -629,17 +639,13 @@ func connectToLogServiceAddresses( stopConnectFallbackTimer(timer) active-- if r.client != nil { - cancel() - wg.Wait() - closeUnusedConnectResultClients(results) + cleanupConnectAttempts(cancel, &wg, results) return r.client, nil } lastErr = r.err case <-fallback: if err := ctx.Err(); err != nil { - cancel() - wg.Wait() - closeUnusedConnectResultClients(results) + cleanupConnectAttempts(cancel, &wg, results) return nil, err } launch(addresses[launched]) @@ -647,9 +653,7 @@ func connectToLogServiceAddresses( active++ case <-ctx.Done(): stopConnectFallbackTimer(timer) - cancel() - wg.Wait() - closeUnusedConnectResultClients(results) + cleanupConnectAttempts(cancel, &wg, results) return nil, connectContextError(ctx.Err(), lastErr) } } @@ -657,6 +661,29 @@ func connectToLogServiceAddresses( return nil, lastErr } +func connectAttemptContext(ctx context.Context) (context.Context, context.CancelFunc) { + if _, ok := ctx.Deadline(); ok { + return context.WithCancel(ctx) + } + return context.WithTimeoutCause( + ctx, + logServiceConnectAttemptTimeout, + moerr.CauseNewLogServiceClient, + ) +} + +func cleanupConnectAttempts( + cancel context.CancelFunc, + wg *sync.WaitGroup, + results <-chan connectResult, +) { + cancel() + go func() { + wg.Wait() + closeUnusedConnectResultClients(results) + }() +} + func connectContextError(ctxErr error, lastErr error) error { if ctxErr == nil || lastErr == nil { return ctxErr diff --git a/pkg/logservice/client_test.go b/pkg/logservice/client_test.go index 730d78905c997..e5601a1b7d349 100644 --- a/pkg/logservice/client_test.go +++ b/pkg/logservice/client_test.go @@ -305,10 +305,13 @@ func TestConnectToLogServiceAddressesReturnsContextErrorAfterPreviousFailure(t * func TestConnectToLogServiceAddressesUsesCallerContextForAttempts(t *testing.T) { origConnect := connectToLogServiceAddressFn + origAttemptTimeout := logServiceConnectAttemptTimeout origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectAttemptTimeout = time.Second logServiceConnectFallbackDelay = time.Millisecond defer func() { connectToLogServiceAddressFn = origConnect + logServiceConnectAttemptTimeout = origAttemptTimeout logServiceConnectFallbackDelay = origFallbackDelay }() @@ -344,6 +347,46 @@ func TestConnectToLogServiceAddressesUsesCallerContextForAttempts(t *testing.T) } } +func TestConnectToLogServiceAddressesAddsDeadlineForBackgroundContext(t *testing.T) { + origConnect := connectToLogServiceAddressFn + origAttemptTimeout := logServiceConnectAttemptTimeout + logServiceConnectAttemptTimeout = 20 * time.Millisecond + defer func() { + connectToLogServiceAddressFn = origConnect + logServiceConnectAttemptTimeout = origAttemptTimeout + }() + + deadlineSeen := make(chan struct{}, 1) + connectToLogServiceAddressFn = func( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, + ) (*client, error) { + _, ok := ctx.Deadline() + require.True(t, ok) + deadlineSeen <- struct{}{} + <-ctx.Done() + return nil, ctx.Err() + } + + c, err := connectToLogServiceAddresses( + context.Background(), + "", + []string{"first"}, + ClientConfig{}, + ) + require.Error(t, err) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.ErrorIs(t, err, moerr.CauseNewLogServiceClient) + require.Nil(t, c) + select { + case <-deadlineSeen: + default: + t.Fatal("attempt context deadline was not observed") + } +} + func TestConnectToLogServiceAddressesWithCanceledContextDoesNotDial(t *testing.T) { origConnect := connectToLogServiceAddressFn defer func() { @@ -379,6 +422,110 @@ func TestConnectToLogServiceAddressesWithCanceledContextDoesNotDial(t *testing.T } } +func TestConnectToLogServiceAddressesReturnsWinnerWithoutWaitingForLoser(t *testing.T) { + origConnect := connectToLogServiceAddressFn + origFallbackDelay := logServiceConnectFallbackDelay + logServiceConnectFallbackDelay = time.Millisecond + defer func() { + connectToLogServiceAddressFn = origConnect + logServiceConnectFallbackDelay = origFallbackDelay + }() + + firstStarted := make(chan struct{}) + releaseFirst := make(chan struct{}) + connectToLogServiceAddressFn = func( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, + ) (*client, error) { + if addr == "first" { + close(firstStarted) + <-releaseFirst + return nil, errors.New("first failed") + } + return &client{ + client: &closeTrackingRPCClient{ + closeMu: &sync.Mutex{}, + closeCount: new(int), + }, + }, nil + } + + done := make(chan *client, 1) + errs := make(chan error, 1) + go func() { + c, err := connectToLogServiceAddresses( + context.Background(), + "", + []string{"first", "second"}, + ClientConfig{}, + ) + if err != nil { + errs <- err + return + } + done <- c + }() + + <-firstStarted + var c *client + select { + case c = <-done: + case err := <-errs: + require.NoError(t, err) + case <-time.After(100 * time.Millisecond): + t.Fatal("timed out waiting for connection result") + } + require.NotNil(t, c) + require.NoError(t, c.close()) + close(releaseFirst) +} + +func TestConnectToLogServiceAddressesReturnsContextErrorWithoutWaitingForLoser(t *testing.T) { + origConnect := connectToLogServiceAddressFn + defer func() { + connectToLogServiceAddressFn = origConnect + }() + + started := make(chan struct{}) + release := make(chan struct{}) + connectToLogServiceAddressFn = func( + ctx context.Context, + sid string, + addr string, + cfg ClientConfig, + ) (*client, error) { + close(started) + <-release + return nil, errors.New("first failed") + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + done := make(chan error, 1) + go func() { + c, err := connectToLogServiceAddresses( + ctx, + "", + []string{"first"}, + ClientConfig{}, + ) + require.Nil(t, c) + done <- err + }() + + <-started + select { + case err := <-done: + require.Error(t, err) + require.ErrorIs(t, err, context.DeadlineExceeded) + case <-time.After(100 * time.Millisecond): + t.Fatal("timed out waiting for context error") + } + close(release) +} + func TestConnectToLogServiceAddressesClosesSuccessfulLosers(t *testing.T) { origConnect := connectToLogServiceAddressFn origFallbackDelay := logServiceConnectFallbackDelay @@ -442,9 +589,11 @@ func TestConnectToLogServiceAddressesClosesSuccessfulLosers(t *testing.T) { } require.NotNil(t, c) - closeMu.Lock() - assert.Equal(t, 1, closeCount) - closeMu.Unlock() + require.Eventually(t, func() bool { + closeMu.Lock() + defer closeMu.Unlock() + return closeCount == 1 + }, time.Second, time.Millisecond) require.NoError(t, c.close()) closeMu.Lock()