Repository: tikv/client-go
Pull request #932 (2023-08-07 11:45:10)

use context logger as much as possible (#908)
cherry-pick #908


Patch:
diff --git a/internal/client/client_batch.go b/internal/client/client_batch.go
index a3e2a0d..bf25a6d 100644
--- a/internal/client/client_batch.go
+++ b/internal/client/client_batch.go
@@ -778,7 +778,7 @@ func sendBatchRequest(
 	select {
 	case batchConn.batchCommandsCh <- entry:
 	case <-ctx.Done():
-		logutil.BgLogger().Debug("send request is cancelled",
+		logutil.Logger(ctx).Debug("send request is cancelled",
 			zap.String("to", addr), zap.String("cause", ctx.Err().Error()))
 		return nil, errors.WithStack(ctx.Err())
 	case <-timer.C:
@@ -794,7 +794,7 @@ func sendBatchRequest(
 		return tikvrpc.FromBatchCommandsResponse(res)
 	case <-ctx.Done():
 		atomic.StoreInt32(&entry.canceled, 1)
-		logutil.BgLogger().Debug("wait response is cancelled",
+		logutil.Logger(ctx).Debug("wait response is cancelled",
 			zap.String("to", addr), zap.String("cause", ctx.Err().Error()))
 		return nil, errors.WithStack(ctx.Err())
 	case <-timer.C:
diff --git a/internal/locate/region_cache.go b/internal/locate/region_cache.go
index 59c413e..ec96b71 100644
--- a/internal/locate/region_cache.go
+++ b/internal/locate/region_cache.go
@@ -741,7 +741,7 @@ func (c *RegionCache) GetTiKVRPCContext(bo *retry.Backoffer, id RegionVerID, rep
 	storeFailEpoch := atomic.LoadUint32(&store.epoch)
 	if storeFailEpoch != regionStore.storeEpochs[storeIdx] {
 		cachedRegion.invalidate(Other)
-		logutil.BgLogger().Info("invalidate current region, because others failed on same store",
+		logutil.Logger(bo.GetCtx()).Info("invalidate current region, because others failed on same store",
 			zap.Uint64("region", id.GetID()),
 			zap.String("store", store.addr))
 		return nil, nil
@@ -862,7 +862,7 @@ func (c *RegionCache) GetTiFlashRPCContext(bo *retry.Backoffer, id RegionVerID,
 		storeFailEpoch := atomic.LoadUint32(&store.epoch)
 		if storeFailEpoch != regionStore.storeEpochs[storeIdx] {
 			cachedRegion.invalidate(Other)
-			logutil.BgLogger().Info("invalidate current region, because others failed on same store",
+			logutil.Logger(bo.GetCtx()).Info("invalidate current region, because others failed on same store",
 				zap.Uint64("region", id.GetID()),
 				zap.String("store", store.addr))
 			// TiFlash will always try to find out a valid peer, avoiding to retry too many times.
@@ -1878,7 +1878,7 @@ func (c *RegionCache) OnRegionEpochNotMatch(bo *retry.Backoffer, ctx *RPCContext
 			(meta.GetRegionEpoch().GetConfVer() < ctx.Region.confVer ||
 				meta.GetRegionEpoch().GetVersion() < ctx.Region.ver) {
 			err := errors.Errorf("region epoch is ahead of tikv. rpc ctx: %+v, currentRegions: %+v", ctx, currentRegions)
-			logutil.BgLogger().Info("region epoch is ahead of tikv", zap.Error(err))
+			logutil.Logger(bo.GetCtx()).Info("region epoch is ahead of tikv", zap.Error(err))
 			return true, bo.Backoff(retry.BoRegionMiss, err)
 		}
 	}
@@ -2392,6 +2392,24 @@ const (
 	tombstone
 )
 
+// String implements fmt.Stringer interface.
+func (s resolveState) String() string {
+	switch s {
+	case unresolved:
+		return "unresolved"
+	case resolved:
+		return "resolved"
+	case needCheck:
+		return "needCheck"
+	case deleted:
+		return "deleted"
+	case tombstone:
+		return "tombstone"
+	default:
+		return fmt.Sprintf("unknown-%v", uint64(s))
+	}
+}
+
 // IsTiFlash returns true if the storeType is TiFlash
 func (s *Store) IsTiFlash() bool {
 	return s.storeType == tikvrpc.TiFlash
@@ -2531,6 +2549,12 @@ func (s *Store) changeResolveStateTo(from, to resolveState) bool {
 			return false
 		}
 		if atomic.CompareAndSwapUint64(&s.state, uint64(from), uint64(to)) {
+			logutil.BgLogger().Info("change store resolve state",
+				zap.Uint64("store", s.storeID),
+				zap.String("addr", s.addr),
+				zap.String("from", from.String()),
+				zap.String("to", to.String()),
+				zap.String("liveness-state", s.getLivenessState().String()))
 			return true
 		}
 	}
@@ -2631,6 +2655,20 @@ const (
 	unknown
 )
 
+// String implements fmt.Stringer interface.
+func (s livenessState) String() string {
+	switch s {
+	case unreachable:
+		return "unreachable"
+	case reachable:
+		return "reachable"
+	case unknown:
+		return "unknown"
+	default:
+		return fmt.Sprintf("unknown-%v", uint32(s))
+	}
+}
+
 func (s *Store) startHealthCheckLoopIfNeeded(c *RegionCache, liveness livenessState) {
 	// This mechanism doesn't support non-TiKV stores currently.
 	if s.storeType != tikvrpc.TiKV {
diff --git a/internal/locate/region_request.go b/internal/locate/region_request.go
index e05a5a8..2c478a5 100644
--- a/internal/locate/region_request.go
+++ b/internal/locate/region_request.go
@@ -35,6 +35,7 @@
 package locate
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math"
@@ -402,7 +403,7 @@ func (state *tryFollower) next(bo *retry.Backoffer, selector *replicaSelector) (
 		}
 		targetReplica = selector.replicas[idx]
 		// Each follower is only tried once
-		if !targetReplica.isExhausted(1) {
+		if !targetReplica.isExhausted(1) && targetReplica.store.getLivenessState() != unreachable {
 			state.lastIdx = idx
 			selector.targetIdx = idx
 			break
@@ -587,14 +588,15 @@ func (state *accessFollower) next(bo *retry.Backoffer, selector *replicaSelector
 	}
 	// If there is no candidate, fallback to the leader.
 	if selector.targetIdx < 0 {
+		leader := selector.replicas[state.leaderIdx]
+		leaderInvalid := leader.isEpochStale() || (!state.option.leaderOnly && leader.isExhausted(1))
 		if len(state.option.labels) > 0 {
-			logutil.BgLogger().Warn(
-				"unable to find stores with given labels",
-				zap.Any("labels", state.option.labels),
-			)
+			logutil.Logger(bo.GetCtx()).Warn("unable to find stores with given labels",
+				zap.Uint64("region", selector.region.GetID()),
+				zap.Bool("leader-invalid", leaderInvalid),
+				zap.Any("labels", state.option.labels))
 		}
-		leader := selector.replicas[state.leaderIdx]
-		if leader.isEpochStale() || (!state.option.leaderOnly && leader.isExhausted(1)) {
+		if leaderInvalid {
 			metrics.TiKVReplicaSelectorFailureCounter.WithLabelValues("exhausted").Inc()
 			selector.invalidateRegion()
 			return nil, nil
@@ -628,8 +630,8 @@ func (state *accessFollower) onSendFailure(bo *retry.Backoffer, selector *replic
 }
 
 func (state *accessFollower) isCandidate(idx AccessIndex, replica *replica) bool {
-	// the epoch is staled or retry exhausted.
-	if replica.isEpochStale() || replica.isExhausted(1) {
+	// the epoch is staled or retry exhausted, or the store is unreachable.
+	if replica.isEpochStale() || replica.isExhausted(1) || replica.store.getLivenessState() == unreachable {
 		return false
 	}
 	// The request can only be sent to the leader.
@@ -1168,6 +1170,7 @@ func (s *RegionRequestSender) SendReqCtx(
 		}()
 	}
 
+	totalErrors := make(map[string]int)
 	for {
 		if retryTimes > 0 {
 			req.IsRetryRequest = true
@@ -1200,10 +1203,7 @@ func (s *RegionRequestSender) SendReqCtx(
 
 			// TODO: Change the returned error to something like "region missing in cache",
 			// and handle this error like EpochNotMatch, which means to re-split the request and retry.
-			logutil.Logger(bo.GetCtx()).Debug(
-				"throwing pseudo region error due to region not found in cache",
-				zap.Stringer("region", &regionID),
-			)
+			s.logSendReqError(bo, "throwing pseudo region error due to no replica available", regionID, retryTimes, req, totalErrors)
 			resp, err = tikvrpc.GenRegionErrorResp(req, &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}})
 			return resp, nil, retryTimes, err
 		}
@@ -1229,6 +1229,8 @@ func (s *RegionRequestSender) SendReqCtx(
 		var retry bool
 		resp, retry, err = s.sendReqToRegion(bo, rpcCtx, req, timeout)
 		if err != nil {
+			msg := fmt.Sprintf("send request failed, err: %v", err.Error())
+			s.logSendReqError(bo, msg, regionID, retryTimes, req, totalErrors)
 			return nil, nil, retryTimes, err
 		}
 
@@ -1260,14 +1262,19 @@ func (s *RegionRequestSender) SendReqCtx(
 			return nil, nil, retryTimes, err
 		}
 		if regionErr != nil {
+			regionErrLabel := regionErrorToLabel(regionErr)
+			totalErrors[regionErrLabel]++
 			retry, err = s.onRegionError(bo, rpcCtx, req, regionErr)
 			if err != nil {
+				msg := fmt.Sprintf("send request on region error failed, err: %v", err.Error())
+				s.logSendReqError(bo, msg, regionID, retryTimes, req, totalErrors)
 				return nil, nil, retryTimes, err
 			}
 			if retry {
 				retryTimes++
 				continue
 			}
+			s.logSendReqError(bo, "send request meet region error without retry", regionID, retryTimes, req, totalErrors)
 		} else {
 			if s.replicaSelector != nil {
 				s.replicaSelector.onSendSuccess()
@@ -1280,6 +1287,75 @@ func (s *RegionRequestSender) SendReqCtx(
 	}
 }
 
+func (s *RegionRequestSender) logSendReqError(bo *retry.Backoffer, msg string, regionID RegionVerID, retryTimes int, req *tikvrpc.Request, totalErrors map[string]int) {
+	var replicaStatus []string
+	replicaSelectorState := "nil"
+	cacheRegionIsValid := "unknown"
+	if s.replicaSelector != nil {
+		switch s.replicaSelector.state.(type) {
+		case *accessKnownLeader:
+			replicaSelectorState = "accessKnownLeader"
+		case *accessFollower:
+			replicaSelectorState = "accessFollower"
+		case *accessByKnownProxy:
+			replicaSelectorState = "accessByKnownProxy"
+		case *tryFollower:
+			replicaSelectorState = "tryFollower"
+		case *tryNewProxy:
+			replicaSelectorState = "tryNewProxy"
+		case *invalidLeader:
+			replicaSelectorState = "invalidLeader"
+		case *invalidStore:
+			replicaSelectorState = "invalidStore"
+		case *stateBase:
+			replicaSelectorState = "stateBase"
+		case nil:
+			replicaSelectorState = "nil"
+		}
+		if s.replicaSelector.region != nil {
+			if s.replicaSelector.region.isValid() {
+				cacheRegionIsValid = "true"
+			} else {
+				cacheRegionIsValid = "false"
+			}
+		}
+		for _, replica := range s.replicaSelector.replicas {
+			replicaStatus = append(replicaStatus, fmt.Sprintf("peer: %v, store: %v, isEpochStale: %v, attempts: %v, replica-epoch: %v, store-epoch: %v, store-state: %v, store-liveness-state: %v",
+				replica.peer.GetId(),
+				replica.store.storeID,
+				replica.isEpochStale(),
+				replica.attempts,
+				replica.epoch,
+				atomic.LoadUint32(&replica.store.epoch),
+				replica.store.getResolveState(),
+				replica.store.getLivenessState(),
+			))
+		}
+	}
+	var totalErrorStr bytes.Buffer
+	for err, cnt := range totalErrors {
+		if totalErrorStr.Len() > 0 {
+			totalErrorStr.WriteString(", ")
+		}
+		totalErrorStr.WriteString(err)
+		totalErrorStr.WriteString(":")
+		totalErrorStr.WriteString(strconv.Itoa(cnt))
+	}
+	logutil.Logger(bo.GetCtx()).Info(msg,
+		zap.Uint64("req-ts", req.GetStartTS()),
+		zap.String("req-type", req.Type.String()),
+		zap.String("region", regionID.String()),
+		zap.String("region-is-valid", cacheRegionIsValid),
+		zap.Int("retry-times", retryTimes),
+		zap.String("replica-read-type", req.ReplicaReadType.String()),
+		zap.String("replica-selector-state", replicaSelectorState),
+		zap.Bool("stale-read", req.StaleRead),
+		zap.String("replica-status", strings.Join(replicaStatus, "; ")),
+		zap.Int("total-backoff-ms", bo.GetTotalSleep()),
+		zap.Int("total-backoff-times", bo.GetTotalBackoffTimes()),
+		zap.String("total-region-errors", totalErrorStr.String()))
+}
+
 // RPCCancellerCtxKey is context key attach rpc send cancelFunc collector to ctx.
 type RPCCancellerCtxKey struct{}
 
@@ -1548,7 +1624,7 @@ func (s *RegionRequestSender) onSendFail(bo *retry.Backoffer, ctx *RPCContext, e
 			// If we don't cancel, but the error code is Canceled, it must be from grpc remote.
 			// This may happen when tikv is killed and exiting.
 			// Backoff and retry in this case.
-			logutil.BgLogger().Warn("receive a grpc cancel signal from remote", zap.Error(err))
+			logutil.Logger(bo.GetCtx()).Warn("receive a grpc cancel signal from remote", zap.Error(err))
 		}
 	}
 
@@ -1685,7 +1761,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if notLeader := regionErr.GetNotLeader(); notLeader != nil {
 		// Retry if error is `NotLeader`.
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `NotLeader` retry later",
 			zap.String("notLeader", notLeader.String()),
 			zap.String("ctx", ctx.String()),
@@ -1726,7 +1802,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if regionErr.GetRecoveryInProgress() != nil {
 		s.regionCache.InvalidateCachedRegion(ctx.Region)
-		logutil.BgLogger().Debug("tikv reports `RecoveryInProgress`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `RecoveryInProgress`", zap.Stringer("ctx", ctx))
 		err = bo.Backoff(retry.BoRegionRecoveryInProgress, errors.Errorf("region recovery in progress, ctx: %v", ctx))
 		if err != nil {
 			return false, err
@@ -1736,7 +1812,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if regionErr.GetIsWitness() != nil {
 		s.regionCache.InvalidateCachedRegion(ctx.Region)
-		logutil.BgLogger().Debug("tikv reports `IsWitness`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `IsWitness`", zap.Stringer("ctx", ctx))
 		err = bo.Backoff(retry.BoIsWitness, errors.Errorf("is witness, ctx: %v", ctx))
 		if err != nil {
 			return false, err
@@ -1748,7 +1824,7 @@ func (s *RegionRequestSender) onRegionError(
 	// if a request meets the FlashbackInProgress error, it should stop retrying immediately
 	// to avoid unnecessary backoff and potential unexpected data status to the user.
 	if flashbackInProgress := regionErr.GetFlashbackInProgress(); flashbackInProgress != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `FlashbackInProgress`",
 			zap.Stringer("req", req),
 			zap.Stringer("ctx", ctx),
@@ -1777,7 +1853,7 @@ func (s *RegionRequestSender) onRegionError(
 	// prepared for the flashback before, it should stop retrying immediately to avoid
 	// unnecessary backoff.
 	if regionErr.GetFlashbackNotPrepared() != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `FlashbackNotPrepared`",
 			zap.Stringer("req", req),
 			zap.Stringer("ctx", ctx),
@@ -1795,13 +1871,13 @@ func (s *RegionRequestSender) onRegionError(
 	}
 
 	if regionErr.GetKeyNotInRegion() != nil {
-		logutil.BgLogger().Error("tikv reports `KeyNotInRegion`", zap.Stringer("req", req), zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Error("tikv reports `KeyNotInRegion`", zap.Stringer("req", req), zap.Stringer("ctx", ctx))
 		s.regionCache.InvalidateCachedRegion(ctx.Region)
 		return false, nil
 	}
 
 	if epochNotMatch := regionErr.GetEpochNotMatch(); epochNotMatch != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `EpochNotMatch` retry later",
 			zap.Stringer("EpochNotMatch", epochNotMatch),
 			zap.Stringer("ctx", ctx),
@@ -1817,7 +1893,7 @@ func (s *RegionRequestSender) onRegionError(
 		if s.replicaSelector != nil {
 			return s.replicaSelector.onServerIsBusy(bo, ctx, req, serverIsBusy)
 		}
-		logutil.BgLogger().Warn(
+		logutil.Logger(bo.GetCtx()).Warn(
 			"tikv reports `ServerIsBusy` retry later",
 			zap.String("reason", regionErr.GetServerIsBusy().GetReason()),
 			zap.Stringer("ctx", ctx),
@@ -1837,7 +1913,7 @@ func (s *RegionRequestSender) onRegionError(
 	// We can't know whether the request is committed or not, so it's an undetermined error too,
 	// but we don't handle it now.
 	if regionErr.GetStaleCommand() != nil {
-		logutil.BgLogger().Debug("tikv reports `StaleCommand`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `StaleCommand`", zap.Stringer("ctx", ctx))
 		if s.replicaSelector != nil {
 			// Needn't backoff because the new leader should be elected soon
 			// and the replicaSelector will try the next peer.
@@ -1852,7 +1928,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	if storeNotMatch := regionErr.GetStoreNotMatch(); storeNotMatch != nil {
 		// store not match
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `StoreNotMatch` retry later",
 			zap.Stringer("storeNotMatch", storeNotMatch),
 			zap.Stringer("ctx", ctx),
@@ -1866,12 +1942,12 @@ func (s *RegionRequestSender) onRegionError(
 	}
 
 	if regionErr.GetRaftEntryTooLarge() != nil {
-		logutil.BgLogger().Warn("tikv reports `RaftEntryTooLarge`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Warn("tikv reports `RaftEntryTooLarge`", zap.Stringer("ctx", ctx))
 		return false, errors.New(regionErr.String())
 	}
 
 	if regionErr.GetMaxTimestampNotSynced() != nil {
-		logutil.BgLogger().Debug("tikv reports `MaxTimestampNotSynced`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `MaxTimestampNotSynced`", zap.Stringer("ctx", ctx))
 		err = bo.Backoff(retry.BoMaxTsNotSynced, errors.Errorf("max timestamp not synced, ctx: %v", ctx))
 		if err != nil {
 			return false, err
@@ -1881,7 +1957,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	// A read request may be sent to a peer which has not been initialized yet, we should retry in this case.
 	if regionErr.GetRegionNotInitialized() != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `RegionNotInitialized` retry later",
 			zap.Uint64("store-id", ctx.Store.storeID),
 			zap.Uint64("region-id", regionErr.GetRegionNotInitialized().GetRegionId()),
@@ -1896,7 +1972,7 @@ func (s *RegionRequestSender) onRegionError(
 
 	// The read-index can't be handled timely because the region is splitting or merging.
 	if regionErr.GetReadIndexNotReady() != nil {
-		logutil.BgLogger().Debug(
+		logutil.Logger(bo.GetCtx()).Debug(
 			"tikv reports `ReadIndexNotReady` retry later",
 			zap.Uint64("store-id", ctx.Store.storeID),
 			zap.Uint64("region-id", regionErr.GetRegionNotInitialized().GetRegionId()),
@@ -1911,7 +1987,7 @@ func (s *RegionRequestSender) onRegionError(
 	}
 
 	if regionErr.GetProposalInMergingMode() != nil {
-		logutil.BgLogger().Debug("tikv reports `ProposalInMergingMode`", zap.Stringer("ctx", ctx))
+		logutil.Logger(bo.GetCtx()).Debug("tikv reports `ProposalInMergingMode`", zap.Stringer("ctx", ctx))
 		// The region is merging and it can't provide service until merge finished, so backoff.
 		err = bo.Backoff(retry.BoRegionScheduling, errors.Errorf("region is merging, ctx: %v", ctx))
 		if err != nil {
@@ -1924,7 +2000,7 @@ func (s *RegionRequestSender) onRegionError(
 	// This error is specific to stale read and the target replica is randomly selected. If the request is sent
 	// to the leader, the data must be ready, so we don't backoff here.
 	if regionErr.GetDataIsNotReady() != nil {
-		logutil.BgLogger().Warn(
+		logutil.Logger(bo.GetCtx()).Warn(
 			"tikv reports `DataIsNotReady` retry later",
 			zap.Uint64("store-id", ctx.Store.storeID),
 			zap.Uint64("peer-id", regionErr.GetDataIsNotReady().GetPeerId()),
@@ -1942,7 +2018,7 @@ func (s *RegionRequestSender) onRegionError(
 		return true, nil
 	}
 
-	logutil.BgLogger().Debug(
+	logutil.Logger(bo.GetCtx()).Debug(
 		"tikv reports region failed",
 		zap.Stringer("regionErr", regionErr),
 		zap.Stringer("ctx", ctx),
diff --git a/internal/retry/backoff.go b/internal/retry/backoff.go
index 6a27d05..19f6fca 100644
--- a/internal/retry/backoff.go
+++ b/internal/retry/backoff.go
@@ -35,9 +35,11 @@
 package retry
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -131,7 +133,7 @@ func (b *Backoffer) BackoffWithMaxSleepTxnLockFast(maxSleepMs int, err error) er
 // and never sleep more than maxSleepMs for each sleep.
 func (b *Backoffer) BackoffWithCfgAndMaxSleep(cfg *Config, maxSleepMs int, err error) error {
 	if strings.Contains(err.Error(), tikverr.MismatchClusterID) {
-		logutil.BgLogger().Fatal("critical error", zap.Error(err))
+		logutil.Logger(b.ctx).Fatal("critical error", zap.Error(err))
 	}
 	select {
 	case <-b.ctx.Done():
@@ -150,12 +152,24 @@ func (b *Backoffer) BackoffWithCfgAndMaxSleep(cfg *Config, maxSleepMs int, err e
 				errMsg += "\n" + err.Error()
 			}
 		}
+		var backoffDetail bytes.Buffer
+		totalTimes := 0
+		for name, times := range b.backoffTimes {
+			totalTimes += times
+			if backoffDetail.Len() > 0 {
+				backoffDetail.WriteString(", ")
+			}
+			backoffDetail.WriteString(name)
+			backoffDetail.WriteString(":")
+			backoffDetail.WriteString(strconv.Itoa(times))
+		}
+		errMsg += fmt.Sprintf("\ntotal-backoff-times: %v, backoff-detail: %v", totalTimes, backoffDetail.String())
 		returnedErr := err
 		if longestSleepCfg != nil {
 			errMsg += fmt.Sprintf("\nlongest sleep type: %s, time: %dms", longestSleepCfg.String(), longestSleepTime)
 			returnedErr = longestSleepCfg.err
 		}
-		logutil.BgLogger().Warn(errMsg)
+		logutil.Logger(b.ctx).Warn(errMsg)
 		// Use the backoff type that contributes most to the timeout to generate a MySQL error.
 		return errors.WithStack(returnedErr)
 	}
diff --git a/kv/store_vars.go b/kv/store_vars.go
index cce3e14..184975c 100644
--- a/kv/store_vars.go
+++ b/kv/store_vars.go
@@ -35,6 +35,8 @@
 package kv
 
 import (
+	"fmt"
+
 	"go.uber.org/atomic"
 )
 
@@ -72,3 +74,21 @@ const (
 func (r ReplicaReadType) IsFollowerRead() bool {
 	return r != ReplicaReadLeader
 }
+
+// String implements fmt.Stringer interface.
+func (r ReplicaReadType) String() string {
+	switch r {
+	case ReplicaReadLeader:
+		return "leader"
+	case ReplicaReadFollower:
+		return "follower"
+	case ReplicaReadMixed:
+		return "mixed"
+	case ReplicaReadLearner:
+		return "learner"
+	case ReplicaReadPreferLeader:
+		return "prefer-leader"
+	default:
+		return fmt.Sprintf("unknown-%v", byte(r))
+	}
+}


Test patch:
diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go
index 5bad3f3..236e7b5 100644
--- a/internal/locate/region_request3_test.go
+++ b/internal/locate/region_request3_test.go
@@ -278,6 +278,12 @@ func refreshEpochs(regionStore *regionStore) {
 	}
 }
 
+func refreshLivenessStates(regionStore *regionStore) {
+	for _, store := range regionStore.stores {
+		atomic.StoreUint32(&store.livenessState, uint32(reachable))
+	}
+}
+
 func AssertRPCCtxEqual(s *testRegionRequestToThreeStoresSuite, rpcCtx *RPCContext, target *replica, proxy *replica) {
 	s.Equal(rpcCtx.Store, target.store)
 	s.Equal(rpcCtx.Peer, target.peer)
@@ -567,6 +573,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestReplicaSelector() {
 	// Test accessFollower state with kv.ReplicaReadFollower request type.
 	req = tikvrpc.NewReplicaReadRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{}, kv.ReplicaReadFollower, nil)
 	refreshEpochs(regionStore)
+	refreshLivenessStates(regionStore)
 	replicaSelector, err = newReplicaSelector(cache, regionLoc.Region, req)
 	s.Nil(err)
 	s.NotNil(replicaSelector)
@@ -680,10 +687,13 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	region, err := s.cache.LocateRegionByID(s.bo, s.regionID)
 	s.Nil(err)
 	s.NotNil(region)
+	regionStore := s.cache.GetCachedRegionWithRLock(region.Region).getStore()
+	s.NotNil(regionStore)
 
 	reloadRegion := func() {
 		s.regionRequestSender.replicaSelector.region.invalidate(Other)
 		region, _ = s.cache.LocateRegionByID(s.bo, s.regionID)
+		regionStore = s.cache.GetCachedRegionWithRLock(region.Region).getStore()
 	}
 
 	hasFakeRegionError := func(resp *tikvrpc.Response) bool {
@@ -715,6 +725,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	s.Equal(sender.replicaSelector.targetIdx, AccessIndex(1))
 	s.True(bo.GetTotalBackoffTimes() == 1)
 	s.cluster.StartStore(s.storeIDs[0])
+	atomic.StoreUint32(&regionStore.stores[0].livenessState, uint32(reachable))
 
 	// Leader is updated because of send success, so no backoff.
 	bo = retry.NewBackoffer(context.Background(), -1)
@@ -734,6 +745,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	s.True(hasFakeRegionError(resp))
 	s.Equal(bo.GetTotalBackoffTimes(), 1)
 	s.cluster.StartStore(s.storeIDs[1])
+	atomic.StoreUint32(&regionStore.stores[1].livenessState, uint32(reachable))
 
 	// Leader is changed. No backoff.
 	reloadRegion()
@@ -750,7 +762,7 @@ func (s *testRegionRequestToThreeStoresSuite) TestSendReqWithReplicaSelector() {
 	resp, _, err = sender.SendReq(bo, req, region.Region, time.Second)
 	s.Nil(err)
 	s.True(hasFakeRegionError(resp))
-	s.Equal(bo.GetTotalBackoffTimes(), 2) // The unreachable leader is skipped
+	s.Equal(bo.GetTotalBackoffTimes(), 3)
 	s.False(sender.replicaSelector.region.isValid())
 	s.cluster.ChangeLeader(s.regionID, s.peerIDs[0])
 
@@ -1098,3 +1110,81 @@ func (s *testRegionRequestToThreeStoresSuite) TestReplicaReadFallbackToLeaderReg
 	// after region error returned, the region should be invalidated.
 	s.False(region.isValid())
 }
+
+func (s *testRegionRequestToThreeStoresSuite) TestAccessFollowerAfter1TiKVDown() {
+	var leaderAddr string
+	s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
+		// Returns error when accesses non-leader.
+		if leaderAddr != addr {
+			return nil, context.DeadlineExceeded
+		}
+		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
+			Value: []byte("value"),
+		}}, nil
+	}}
+
+	req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{
+		Key: []byte("key"),
+	})
+	req.ReplicaReadType = kv.ReplicaReadMixed
+
+	loc, err := s.cache.LocateKey(s.bo, []byte("key"))
+	s.Nil(err)
+	region := s.cache.GetCachedRegionWithRLock(loc.Region)
+	s.NotNil(region)
+	regionStore := region.getStore()
+	leaderAddr = regionStore.stores[regionStore.workTiKVIdx].addr
+	s.NotEqual(leaderAddr, "")
+	for i := 0; i < 10; i++ {
+		bo := retry.NewBackofferWithVars(context.Background(), 100, nil)
+		resp, _, _, err := s.regionRequestSender.SendReqCtx(bo, req, loc.Region, time.Second, tikvrpc.TiKV)
+		s.Nil(err)
+		s.NotNil(resp)
+
+		// Since send req to follower will receive error, then all follower will be marked as unreachable and epoch stale.
+		allFollowerStoreEpochStale := true
+		for i, store := range regionStore.stores {
+			if i == int(regionStore.workTiKVIdx) {
+				continue
+			}
+			if store.epoch == regionStore.storeEpochs[i] {
+				allFollowerStoreEpochStale = false
+				break
+			} else {
+				s.Equal(store.getLivenessState(), unreachable)
+			}
+		}
+		if allFollowerStoreEpochStale {
+			break
+		}
+	}
+
+	// mock for GC leader reload all regions.
+	bo := retry.NewBackofferWithVars(context.Background(), 10, nil)
+	_, err = s.cache.BatchLoadRegionsWithKeyRange(bo, []byte(""), nil, 1)
+	s.Nil(err)
+
+	loc, err = s.cache.LocateKey(s.bo, []byte("key"))
+	s.Nil(err)
+	region = s.cache.GetCachedRegionWithRLock(loc.Region)
+	s.NotNil(region)
+	regionStore = region.getStore()
+	for i, store := range regionStore.stores {
+		if i == int(regionStore.workTiKVIdx) {
+			continue
+		}
+		// After reload region, the region epoch will be updated, but the store liveness state is still unreachable.
+		s.Equal(store.epoch, regionStore.storeEpochs[i])
+		s.Equal(store.getLivenessState(), unreachable)
+	}
+
+	for i := 0; i < 100; i++ {
+		bo := retry.NewBackofferWithVars(context.Background(), 1, nil)
+		resp, _, retryTimes, err := s.regionRequestSender.SendReqCtx(bo, req, loc.Region, time.Second, tikvrpc.TiKV)
+		s.Nil(err)
+		s.NotNil(resp)
+		// since all follower'store is unreachable, the request will be sent to leader, the backoff times should be 0.
+		s.Equal(0, bo.GetTotalBackoffTimes())
+		s.Equal(0, retryTimes)
+	}
+}