enhance: fast-fail for proxy-to-QueryNode gRPC retry and delegator stall detection (#49102)

issue: #48185

- Add retry.WithMaxAttemptsContext to cap inner gRPC client retry from
caller's context, enabling proxy LBPolicy to failover to another replica
immediately instead of burning the full backoff budget (~13s) on a dead
QueryNode.
- Apply retry cap=1 to searchShard, queryShard (covers requery path),
and HighlightTask.getHighlightOnShardleader.
- Fast-fail on connection closing error in gRPC client to avoid
unnecessary retries on dead connections.
- Fast-fail for delegator forward delete on ServerIDMismatch.
- waitTSafe stall detection using ContextCond with configurable
stallTimeout (queryNode.waitTsafeStallTimeout, default 3s).
- Mark segments offline on ServerIDMismatch in Search/Query/QueryStream
paths (in addition to existing ErrNodeNotFound handling).
- Fix nil pointer panic in IsServerIDMismatchErr and
IsCrossClusterRoutingErr when called with nil error.
- Add cache for ParamItem.GetAsDurationByParse.

Signed-off-by: chyezh <chyezh@outlook.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

internal/proxy/task.go

@@ -41,6 +41,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/retry"
 	"github.com/milvus-io/milvus/pkg/v2/util/timestamptz"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
@@ -3747,6 +3748,7 @@ func (t *HighlightTask) PreExecute(ctx context.Context) error {
 }
 
 func (t *HighlightTask) getHighlightOnShardleader(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channel string) error {
+	ctx = retry.WithMaxAttemptsContext(ctx, 1)
 	t.Channel = channel
 	resp, err := qn.GetHighlight(ctx, t.GetHighlightRequest)
 	if err != nil {

internal/proxy/task_query.go

@@ -35,6 +35,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/retry"
 	"github.com/milvus-io/milvus/pkg/v2/util/timerecord"
 	"github.com/milvus-io/milvus/pkg/v2/util/timestamptz"
 	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
@@ -986,6 +987,7 @@ func (t *queryTask) IsSubTask() bool {
 }
 
 func (t *queryTask) queryShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channel string) error {
+	ctx = retry.WithMaxAttemptsContext(ctx, 1)
 	needOverrideMvcc := false
 	mvccTs := t.MvccTimestamp
 	if len(t.channelsMvcc) > 0 {

internal/proxy/task_search.go

@@ -39,6 +39,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/metric"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/retry"
 	"github.com/milvus-io/milvus/pkg/v2/util/timerecord"
 	"github.com/milvus-io/milvus/pkg/v2/util/timestamptz"
 	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
@@ -1140,6 +1141,7 @@ func (t *searchTask) PostExecute(ctx context.Context) error {
 }
 
 func (t *searchTask) searchShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channel string) error {
+	ctx = retry.WithMaxAttemptsContext(ctx, 1)
 	searchReq := shallowcopy.ShallowCopySearchRequest(t.SearchRequest, nodeID)
 	req := &querypb.SearchRequest{
 		Req:             searchReq,

internal/querynodev2/delegator/delegator.go

@@ -32,8 +32,6 @@ import (
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
@@ -44,6 +42,7 @@ import (
 	"github.com/milvus-io/milvus/internal/querynodev2/segments"
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/internal/util/function"
+	"github.com/milvus-io/milvus/internal/util/grpcclient"
 	"github.com/milvus-io/milvus/internal/util/reduce"
 	"github.com/milvus-io/milvus/internal/util/searchutil/optimizers"
 	"github.com/milvus-io/milvus/internal/util/shallowcopy"
@@ -62,6 +61,8 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/util/metautil"
 	"github.com/milvus-io/milvus/pkg/v2/util/metric"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/retry"
+	"github.com/milvus-io/milvus/pkg/v2/util/syncutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/timerecord"
 	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
@@ -140,7 +141,7 @@ type shardDelegator struct {
 
 	sf          conc.Singleflight[struct{}]
 	loader      segments.Loader
-	tsCond      *sync.Cond
+	tsCond      *syncutil.ContextCond
 	latestTsafe *atomic.Uint64
 	// queryHook
 	queryHook      optimizers.QueryHook
@@ -293,9 +294,9 @@ func (sd *shardDelegator) executeSearchSubTasks(
 
 	results, err := executeSubTasks(ctx, tasks, NewRowCountBasedEvaluator(sealedRowCount),
 		func(ctx context.Context, req *querypb.SearchRequest, worker cluster.Worker) (*internalpb.SearchResults, error) {
+			ctx = retry.WithMaxAttemptsContext(ctx, 3)
 			resp, err := worker.SearchSegments(ctx, req)
-			st, ok := status.FromError(err)
-			if ok && st.Code() == codes.Unavailable {
+			if errors.Is(err, merr.ErrNodeNotFound) || grpcclient.IsServerIDMismatchErr(err) {
 				sd.markSegmentOffline(req.GetSegmentIDs()...)
 			}
 			return resp, err
@@ -394,7 +395,6 @@ func (sd *shardDelegator) search(ctx context.Context, req *querypb.SearchRequest
 		log.Warn("failed to optimize search params", zap.Error(err))
 		return nil, err
 	}
-
 	return sd.executeSearchSubTasks(ctx, req, sealed, growing, sealedRowCount)
 }
 
@@ -623,9 +623,9 @@ func (sd *shardDelegator) QueryStream(ctx context.Context, req *querypb.QueryReq
 	}
 
 	_, err = executeSubTasks(ctx, tasks, NewRowCountBasedEvaluator(sealedRowCount), func(ctx context.Context, req *querypb.QueryRequest, worker cluster.Worker) (*internalpb.RetrieveResults, error) {
+		ctx = retry.WithMaxAttemptsContext(ctx, 3)
 		err := worker.QueryStreamSegments(ctx, req, srv)
-		status, ok := status.FromError(err)
-		if ok && status.Code() == codes.Unavailable {
+		if errors.Is(err, merr.ErrNodeNotFound) || grpcclient.IsServerIDMismatchErr(err) {
 			sd.markSegmentOffline(req.GetSegmentIDs()...)
 		}
 		return nil, err
@@ -723,9 +723,9 @@ func (sd *shardDelegator) Query(ctx context.Context, req *querypb.QueryRequest)
 	}
 
 	results, err := executeSubTasks(ctx, tasks, NewRowCountBasedEvaluator(sealedRowCount), func(ctx context.Context, req *querypb.QueryRequest, worker cluster.Worker) (*internalpb.RetrieveResults, error) {
+		ctx = retry.WithMaxAttemptsContext(ctx, 3)
 		resp, err := worker.QuerySegments(ctx, req)
-		status, ok := status.FromError(err)
-		if ok && status.Code() == codes.Unavailable {
+		if errors.Is(err, merr.ErrNodeNotFound) || grpcclient.IsServerIDMismatchErr(err) {
 			sd.markSegmentOffline(req.GetSegmentIDs()...)
 		}
 		return resp, err
@@ -811,6 +811,7 @@ func (sd *shardDelegator) GetStatistics(ctx context.Context, req *querypb.GetSta
 	}
 
 	results, err := executeSubTasks(ctx, tasks, NewRowCountBasedEvaluator(sealedRowCount), func(ctx context.Context, req *querypb.GetStatisticsRequest, worker cluster.Worker) (*internalpb.GetStatisticsResponse, error) {
+		ctx = retry.WithMaxAttemptsContext(ctx, 1)
 		return worker.GetStatistics(ctx, req)
 	}, "GetStatistics", log)
 	if err != nil {
@@ -1030,12 +1031,16 @@ func (sd *shardDelegator) waitTSafe(ctx context.Context, ts uint64) (uint64, err
 	defer sp.End()
 	log := sd.getLogger(ctx)
 
-	// already safe to search
+	// Fast path: tSafe already meets the guarantee timestamp.
 	latestTSafe := sd.latestTsafe.Load()
 	if latestTSafe >= ts {
 		return latestTSafe, nil
 	}
-	// check whethertsafe downgraded
+
+	// Slow path: tSafe has not yet reached the guarantee timestamp,
+	// need to wait for tSafe to advance via condition variable.
+
+	// check whether tsafe downgraded
 	if paramtable.Get().QueryNodeCfg.DowngradeTsafe.GetAsBool() {
 		log.WithRateGroup("downgradeTsafe", 1, 60).RatedWarn(10, "downgrade tsafe", zap.Uint64("latestTSafe", latestTSafe), zap.Uint64("ts", ts))
 		return latestTSafe, nil
@@ -1055,33 +1060,41 @@ func (sd *shardDelegator) waitTSafe(ctx context.Context, ts uint64) (uint64, err
 		return 0, WrapErrTsLagTooLarge(lag, maxLag)
 	}
 
-	ch := make(chan struct{})
-	go func() {
-		sd.tsCond.L.Lock()
-		defer sd.tsCond.L.Unlock()
+	// Stall detection: if tSafe does not advance within stallTimeout, return
+	// a retryable error so the proxy can failover to another replica.
+	// This handles cases where the WAL consumption pipeline is blocked
+	// (e.g. forward delete retrying against a dead QueryNode).
+	stallTimeout := paramtable.Get().QueryNodeCfg.WaitTsafeStallTimeout.GetAsDurationByParse()
 
-		for sd.latestTsafe.Load() < ts &&
-			ctx.Err() == nil &&
-			sd.Serviceable() {
-			sd.tsCond.Wait()
-		}
-		close(ch)
-	}()
-
-	for {
-		select {
-		// timeout
-		case <-ctx.Done():
-			// notify wait goroutine to quit
-			sd.tsCond.Broadcast()
-			return 0, ctx.Err()
-		case <-ch:
-			if !sd.Serviceable() {
-				return 0, merr.WrapErrChannelNotAvailable(sd.vchannelName, "delegator closed during wait tsafe")
+	// Standard condition variable pattern: Lock → for !condition { Wait } → Unlock
+	sd.tsCond.L.Lock()
+	for sd.latestTsafe.Load() < ts && sd.Serviceable() {
+		stallCtx, stallCancel := context.WithTimeout(ctx, stallTimeout)
+		err := sd.tsCond.Wait(stallCtx)
+		stallCancel()
+		if err != nil {
+			// Wait returned without holding the lock.
+			if ctx.Err() != nil {
+				return 0, ctx.Err()
 			}
-			return sd.latestTsafe.Load(), nil
+			// No broadcast within stallTimeout — tSafe is stalled.
+			log.Warn("tsafe stall detected, fast-fail to allow proxy failover",
+				zap.Uint64("currentTsafe", sd.latestTsafe.Load()),
+				zap.Uint64("targetTs", ts),
+				zap.Duration("stallTimeout", stallTimeout),
+			)
+			return 0, WrapErrTsLagTooLarge(gt.Sub(st), stallTimeout)
 		}
+		// Woken by broadcast with lock re-acquired, loop back to re-check condition.
 	}
+	current := sd.latestTsafe.Load()
+	serviceable := sd.Serviceable()
+	sd.tsCond.L.Unlock()
+
+	if !serviceable {
+		return 0, merr.WrapErrChannelNotAvailable(sd.vchannelName, "delegator closed during wait tsafe")
+	}
+	return current, nil
 }
 
 // GetLatestRequiredMVCCTimeTick returns the latest required mvcc timestamp for the delegator.
@@ -1112,35 +1125,39 @@ func (sd *shardDelegator) updateLatestRequiredMVCCTimestamp(ts uint64) {
 // updateTSafe read current tsafe value from tsafeManager.
 func (sd *shardDelegator) UpdateTSafe(tsafe uint64) {
 	log := sd.getLogger(context.Background()).WithRateGroup(fmt.Sprintf("UpdateTSafe-%s", sd.vchannelName), 1, 60)
-	sd.tsCond.L.Lock()
 	log.RatedInfo(10, "update tsafe",
 		zap.Int64("collectionID", sd.collectionID),
 		zap.String("vchannel", sd.vchannelName),
 		zap.Time("tsafe", tsoutil.PhysicalTime(tsafe)),
 		zap.Time("latestTSafe", tsoutil.PhysicalTime(sd.latestTsafe.Load())))
-	if tsafe > sd.latestTsafe.Load() {
-		sd.latestTsafe.Store(tsafe)
-		sd.tsCond.Broadcast()
+	if tsafe <= sd.latestTsafe.Load() {
+		return
+	}
 
-		// Check if caught up with streaming data
-		if sd.catchingUpStreamingData.Load() {
-			lagThreshold := paramtable.Get().QueryNodeCfg.CatchUpStreamingDataTsLag.GetAsDurationByParse()
-			if lagThreshold > 0 {
-				tsafeTime := tsoutil.PhysicalTime(tsafe)
-				lag := time.Since(tsafeTime)
-				caughtUp := lag <= lagThreshold
-				log.RatedInfo(10, "delegator catching up streaming data progress",
-					zap.String("channel", sd.vchannelName),
-					zap.Duration("lag", lag),
-					zap.Duration("threshold", lagThreshold),
-					zap.Bool("caughtUp", caughtUp))
-				if caughtUp {
-					sd.catchingUpStreamingData.Store(false)
-				}
+	// Store and broadcast under lock to prevent lost wakeups:
+	// without the lock, a waiter could observe the old tSafe, then enter
+	// Wait() after the broadcast has already fired, missing the notification.
+	sd.tsCond.LockAndBroadcast()
+	sd.latestTsafe.Store(tsafe)
+	sd.tsCond.L.Unlock()
+
+	// Check if caught up with streaming data (all fields are atomic, no lock needed)
+	if sd.catchingUpStreamingData.Load() {
+		lagThreshold := paramtable.Get().QueryNodeCfg.CatchUpStreamingDataTsLag.GetAsDurationByParse()
+		if lagThreshold > 0 {
+			tsafeTime := tsoutil.PhysicalTime(tsafe)
+			lag := time.Since(tsafeTime)
+			caughtUp := lag <= lagThreshold
+			log.RatedInfo(10, "delegator catching up streaming data progress",
+				zap.String("channel", sd.vchannelName),
+				zap.Duration("lag", lag),
+				zap.Duration("threshold", lagThreshold),
+				zap.Bool("caughtUp", caughtUp))
+			if caughtUp {
+				sd.catchingUpStreamingData.Store(false)
 			}
 		}
 	}
-	sd.tsCond.L.Unlock()
 }
 
 func (sd *shardDelegator) GetTSafe() uint64 {
@@ -1198,6 +1215,7 @@ func (sd *shardDelegator) UpdateSchema(ctx context.Context, schema *schemapb.Col
 	}
 
 	_, err = executeSubTasks(ctx, tasks, nil, func(ctx context.Context, req *querypb.UpdateSchemaRequest, worker cluster.Worker) (*StatusWrapper, error) {
+		ctx = retry.WithMaxAttemptsContext(ctx, 1)
 		status, err := worker.UpdateSchema(ctx, req)
 		return (*StatusWrapper)(status), err
 	}, "UpdateSchema", log)
@@ -1215,8 +1233,9 @@ func (w *StatusWrapper) GetStatus() *commonpb.Status {
 func (sd *shardDelegator) Close() {
 	sd.lifetime.SetState(lifetime.Stopped)
 	sd.lifetime.Close()
-	// broadcast to all waitTsafe goroutine to quit
-	sd.tsCond.Broadcast()
+	// broadcast to all waitTsafe to quit
+	sd.tsCond.LockAndBroadcast()
+	sd.tsCond.L.Unlock()
 	sd.lifetime.Wait()
 
 	// Stop background snapshot loop before refunding candidates
@@ -1384,8 +1403,7 @@ func NewShardDelegator(ctx context.Context, collectionID UniqueID, replicaID Uni
 		sd.idfOracle.Start()
 	}
 
-	m := sync.Mutex{}
-	sd.tsCond = sync.NewCond(&m)
+	sd.tsCond = syncutil.NewContextCond(&sync.Mutex{})
 	log.Info("finish build new shardDelegator")
 	return sd, nil
 }

internal/querynodev2/delegator/delegator_data.go

@@ -37,6 +37,7 @@ import (
 	"github.com/milvus-io/milvus/internal/querynodev2/pkoracle"
 	"github.com/milvus-io/milvus/internal/querynodev2/segments"
 	"github.com/milvus-io/milvus/internal/storage"
+	"github.com/milvus-io/milvus/internal/util/grpcclient"
 	"github.com/milvus-io/milvus/pkg/v2/common"
 	"github.com/milvus-io/milvus/pkg/v2/log"
 	"github.com/milvus-io/milvus/pkg/v2/metrics"
@@ -320,6 +321,10 @@ func (sd *shardDelegator) applyDelete(ctx context.Context,
 						// cancel other request
 						cancel()
 						return false, err
+					} else if grpcclient.IsServerIDMismatchErr(err) {
+						log.Warn("try to delete data on mismatched node, node has been replaced", zap.Error(err))
+						cancel()
+						return false, err
 					} else if errors.IsAny(err, merr.ErrSegmentNotFound, merr.ErrSegmentNotLoaded) {
 						log.Warn("try to delete data of released segment")
 						return false, nil

internal/querynodev2/delegator/delegator_test.go

@@ -51,6 +51,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/util/merr"
 	"github.com/milvus-io/milvus/pkg/v2/util/metric"
 	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+	"github.com/milvus-io/milvus/pkg/v2/util/syncutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
@@ -2192,7 +2193,7 @@ func TestDelegatorCatchingUpStreamingData(t *testing.T) {
 			vchannelName:               "test-channel",
 			latestTsafe:                atomic.NewUint64(0),
 			catchingUpStreamingData:    atomic.NewBool(true),
-			tsCond:                     sync.NewCond(&sync.Mutex{}),
+			tsCond:                     syncutil.NewContextCond(&sync.Mutex{}),
 			latestRequiredMVCCTimeTick: atomic.NewUint64(0),
 		}
 
@@ -2216,7 +2217,7 @@ func TestDelegatorCatchingUpStreamingData(t *testing.T) {
 			vchannelName:               "test-channel",
 			latestTsafe:                atomic.NewUint64(0),
 			catchingUpStreamingData:    atomic.NewBool(true),
-			tsCond:                     sync.NewCond(&sync.Mutex{}),
+			tsCond:                     syncutil.NewContextCond(&sync.Mutex{}),
 			latestRequiredMVCCTimeTick: atomic.NewUint64(0),
 		}
 
@@ -2240,7 +2241,7 @@ func TestDelegatorCatchingUpStreamingData(t *testing.T) {
 			vchannelName:               "test-channel",
 			latestTsafe:                atomic.NewUint64(0),
 			catchingUpStreamingData:    atomic.NewBool(true),
-			tsCond:                     sync.NewCond(&sync.Mutex{}),
+			tsCond:                     syncutil.NewContextCond(&sync.Mutex{}),
 			latestRequiredMVCCTimeTick: atomic.NewUint64(0),
 		}
 

internal/util/grpcclient/client.go

@@ -419,6 +419,10 @@ func (c *ClientBase[T]) checkGrpcErr(ctx context.Context, err error) (needRetry,
 	// grpc err
 	log.Warn("call received grpc error", zap.Error(err))
 	switch {
+	case IsConnectionClosingErr(err):
+		// Connection is being torn down, retry is pointless.
+		// Fast-fail and force reconnection for next call.
+		return false, true, true, err
 	case funcutil.IsGrpcErr(err, codes.Canceled, codes.DeadlineExceeded):
 		// canceled or deadline exceeded
 		return true, c.needResetCancel(), false, err
@@ -545,7 +549,7 @@ func (c *ClientBase[T]) call(ctx context.Context, caller func(client T) (any, er
 			return true, err
 		}
 		return false, nil
-	}, retry.Attempts(uint(c.MaxAttempts)),
+	}, retry.Attempts(retry.MaxAttemptsFromContextOrDefault(ctx, uint(c.MaxAttempts))),
 		// Because the previous InitialBackoff and MaxBackoff were float, and the unit was s.
 		// For compatibility, this is multiplied by 1000.
 		retry.Sleep(time.Duration(c.InitialBackoff*1000)*time.Millisecond),
@@ -615,13 +619,23 @@ func (c *ClientBase[T]) SetSession(sess *sessionutil.Session) {
 }
 
 func IsCrossClusterRoutingErr(err error) bool {
+	if err == nil {
+		return false
+	}
 	// GRPC utilizes `status.Status` to encapsulate errors,
 	// hence it is not viable to employ the `errors.Is` for assessment.
 	return strings.Contains(err.Error(), merr.ErrServiceCrossClusterRouting.Error())
 }
 
 func IsServerIDMismatchErr(err error) bool {
+	if err == nil {
+		return false
+	}
 	// GRPC utilizes `status.Status` to encapsulate errors,
 	// hence it is not viable to employ the `errors.Is` for assessment.
 	return strings.Contains(err.Error(), merr.ErrNodeNotMatch.Error())
 }
+
+func IsConnectionClosingErr(err error) bool {
+	return errors.Is(err, grpc.ErrClientConnClosing)
+}

internal/util/grpcclient/client_test.go

@@ -630,3 +630,24 @@ func TestClientBase_ServerIDMismatch_NodeFastFail(t *testing.T) {
 	// The caller should be invoked exactly once (no retries)
 	assert.Equal(t, 1, callCount)
 }
+
+func TestIsConnectionClosingErr(t *testing.T) {
+	// Positive case — the exact exported sentinel
+	assert.True(t, IsConnectionClosingErr(grpc.ErrClientConnClosing))
+
+	// Positive case — wrapped sentinel still matches via errors.Is
+	err := errors.Wrap(grpc.ErrClientConnClosing, "outer context")
+	assert.True(t, IsConnectionClosingErr(err))
+
+	// Positive case — status with same code and message (proto.Equal match)
+	err = status.Error(codes.Canceled, "grpc: the client connection is closing")
+	assert.True(t, IsConnectionClosingErr(err))
+
+	// Negative — normal canceled
+	err = status.Error(codes.Canceled, "context canceled")
+	assert.False(t, IsConnectionClosingErr(err))
+
+	// Negative — non-grpc error
+	err = errors.New("random error")
+	assert.False(t, IsConnectionClosingErr(err))
+}

pkg/util/paramtable/component_param.go

@@ -3446,6 +3446,7 @@ type queryNodeConfig struct {
 	// tsafe
 	MaxTimestampLag           ParamItem `refreshable:"true"`
 	DowngradeTsafe            ParamItem `refreshable:"true"`
+	WaitTsafeStallTimeout     ParamItem `refreshable:"true"`
 	CatchUpStreamingDataTsLag ParamItem `refreshable:"true"`
 
 	// delete buffer
@@ -4412,6 +4413,15 @@ Max read concurrency must greater than or equal to 1, and less than or equal to
 	}
 	p.DowngradeTsafe.Init(base.mgr)
 
+	p.WaitTsafeStallTimeout = ParamItem{
+		Key:          "queryNode.waitTsafeStallTimeout",
+		Version:      "2.6.15",
+		Doc:          "If tSafe does not advance within this duration during waitTSafe, the delegator returns an error to allow proxy failover to another replica",
+		DefaultValue: "3s",
+		Export:       false,
+	}
+	p.WaitTsafeStallTimeout.Init(base.mgr)
+
 	p.CatchUpStreamingDataTsLag = ParamItem{
 		Key:          "queryNode.delegator.catchUpStreamingDataTsLag",
 		Version:      "2.6.8",

pkg/util/paramtable/param_item.go

@@ -327,7 +327,12 @@ func (pi *ParamItem) GetAsRoleDetails() map[string](map[string]([](map[string]st
 }
 
 func (pi *ParamItem) GetAsDurationByParse() time.Duration {
-	val, _ := pi.get()
+	if val, exist := pi.manager.GetCachedValue(pi.Key); exist {
+		if durationVal, ok := val.(time.Duration); ok {
+			return durationVal
+		}
+	}
+	val, raw, _ := pi.getWithRaw()
 	durationVal, err := time.ParseDuration(val)
 	if err != nil {
 		durationVal, err = time.ParseDuration(pi.DefaultValue)
@@ -335,6 +340,7 @@ func (pi *ParamItem) GetAsDurationByParse() time.Duration {
 			panic(fmt.Sprintf("unreachable: parse duration from default value failed, %s, err: %s", pi.DefaultValue, err.Error()))
 		}
 	}
+	pi.manager.CASCachedValue(pi.Key, raw, durationVal)
 	return durationVal
 }
 

pkg/util/retry/options.go

@@ -11,7 +11,37 @@
 
 package retry
 
-import "time"
+import (
+	"context"
+	"time"
+)
+
+// maxAttemptsKey is the context key for storing max retry attempts.
+type maxAttemptsKey struct{}
+
+// WithMaxAttemptsContext stores the max retry attempts in the context.
+func WithMaxAttemptsContext(ctx context.Context, attempts uint) context.Context {
+	return context.WithValue(ctx, maxAttemptsKey{}, attempts)
+}
+
+// MaxAttemptsFromContext reads the max retry attempts from the context.
+// Returns (0, false) if not set.
+func MaxAttemptsFromContext(ctx context.Context) (uint, bool) {
+	raw := ctx.Value(maxAttemptsKey{})
+	if raw == nil {
+		return 0, false
+	}
+	return raw.(uint), true
+}
+
+// MaxAttemptsFromContextOrDefault reads the max retry attempts from the context,
+// returning defaultVal if no value is set.
+func MaxAttemptsFromContextOrDefault(ctx context.Context, defaultVal uint) uint {
+	if v, ok := MaxAttemptsFromContext(ctx); ok {
+		return v
+	}
+	return defaultVal
+}
 
 type config struct {
 	attempts     uint

pkg/util/retry/options_test.go

@@ -0,0 +1,53 @@
+// Copyright (C) 2019-2020 Zilliz. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software distributed under the License
+// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+// or implied. See the License for the specific language governing permissions and limitations under the License.
+
+package retry
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMaxAttemptsFromContext(t *testing.T) {
+	// No value set — returns 0, false
+	ctx := context.Background()
+	v, ok := MaxAttemptsFromContext(ctx)
+	assert.False(t, ok)
+	assert.Equal(t, uint(0), v)
+
+	// Value set — returns it
+	ctx = WithMaxAttemptsContext(ctx, 3)
+	v, ok = MaxAttemptsFromContext(ctx)
+	assert.True(t, ok)
+	assert.Equal(t, uint(3), v)
+
+	// Zero is a valid explicit value
+	ctx = WithMaxAttemptsContext(context.Background(), 0)
+	v, ok = MaxAttemptsFromContext(ctx)
+	assert.True(t, ok)
+	assert.Equal(t, uint(0), v)
+}
+
+func TestMaxAttemptsFromContextOrDefault(t *testing.T) {
+	// No value in ctx — returns default
+	ctx := context.Background()
+	assert.Equal(t, uint(10), MaxAttemptsFromContextOrDefault(ctx, 10))
+
+	// Value in ctx — returns ctx value, ignoring default
+	ctx = WithMaxAttemptsContext(ctx, 3)
+	assert.Equal(t, uint(3), MaxAttemptsFromContextOrDefault(ctx, 10))
+
+	// Zero in ctx — returns 0, not default
+	ctx = WithMaxAttemptsContext(context.Background(), 0)
+	assert.Equal(t, uint(0), MaxAttemptsFromContextOrDefault(ctx, 10))
+}