fix: fix force promote replication issues (#48936)

## Summary
- Strip replicate headers from incomplete broadcast messages during
force promote to re-append as primary messages
- Implement `GetSalvageCheckpoint` remote WAL path via gRPC, fixing
proxy infinite retry when handling `GetReplicateInfo` for non-failover
CDC sync
- Replace `FastLock`+panic with blocking `Lock` during force promote to
handle concurrent ack callbacks (e.g., replicated `DropLoadConfig`) that
still hold resource key locks
- Fix data race in `broadcastTask` between `MarkIgnore` and `Header`:
`MarkIgnore` (for force promote) mutates `b.msg` and the shared
properties map, breaking the immutability assumption. Split `Header()`
into locked/unlocked versions, deep copy properties in `MarkIgnore`, and
add locks to all unprotected `b.msg` readers
(`IsAlterReplicateConfigMessage`, `IsForcePromoteMessage`,
`isControlChannelAcked`)

## Test plan
- [x] Unit tests added for `ClearReplicateHeader`, `WithWALTerm`
idempotency, and force promote lock contention
- [x] Manual CDC replication test (`test_partition.py`) passed
- [x] Manual force promote test (`test_incomplete_broadcast_ddl.py`)
passed
- [x] Broadcaster unit tests pass with `-race` detector (3 sub-packages)

issue: https://github.com/milvus-io/milvus/issues/48821

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
Signed-off-by: Yihao Dai <yihao.dai@zilliz.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

client/go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/cockroachdb/errors v1.9.1
 	github.com/google/uuid v1.6.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
-	github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260304062516-e7e54d966ef2
+	github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260309063517-8b5776d31f2b
 	github.com/milvus-io/milvus/pkg/v2 v2.6.4-0.20251104142533-a2ce70d25256
 	github.com/quasilyte/go-ruleguard/dsl v0.3.23
 	github.com/samber/lo v1.52.0

client/go.sum

@@ -331,8 +331,8 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5
 github.com/mediocregopher/radix/v3 v3.4.2/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i4n7wVopoX3x7Bv8=
 github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260304062516-e7e54d966ef2 h1:OEqYnraL/Tf2nGMjRJji6xaaDe40zSQVBvruzEBZ1hU=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260304062516-e7e54d966ef2/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260309063517-8b5776d31f2b h1:tUp6a72FFeosfjN+8jy/Hax7BqVbtgsaMMVrAfzKnEc=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260309063517-8b5776d31f2b/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
 github.com/milvus-io/milvus/pkg/v2 v2.6.4-0.20251104142533-a2ce70d25256 h1:M2waty0w2k4YT2HHzJk3fx6EFPD4DKxNJatitIV+gGU=
 github.com/milvus-io/milvus/pkg/v2 v2.6.4-0.20251104142533-a2ce70d25256/go.mod h1:HT6Wxahwj/l8+i+D/C3iwDzCjDa36U9gyVw6CjjK4pE=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=

client/milvusclient/mock_milvus_server_test.go

@@ -3141,6 +3141,53 @@ func (_c *MilvusServiceServer_Dummy_Call) RunAndReturn(run func(context.Context,
 	return _c
 }
 
+// DumpMessages provides a mock function with given fields: _a0, _a1
+func (_m *MilvusServiceServer) DumpMessages(_a0 *milvuspb.DumpMessagesRequest, _a1 milvuspb.MilvusService_DumpMessagesServer) error {
+	ret := _m.Called(_a0, _a1)
+
+	if len(ret) == 0 {
+		panic("no return value specified for DumpMessages")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(*milvuspb.DumpMessagesRequest, milvuspb.MilvusService_DumpMessagesServer) error); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// MilvusServiceServer_DumpMessages_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'DumpMessages'
+type MilvusServiceServer_DumpMessages_Call struct {
+	*mock.Call
+}
+
+// DumpMessages is a helper method to define mock.On call
+//   - _a0 *milvuspb.DumpMessagesRequest
+//   - _a1 milvuspb.MilvusService_DumpMessagesServer
+func (_e *MilvusServiceServer_Expecter) DumpMessages(_a0 interface{}, _a1 interface{}) *MilvusServiceServer_DumpMessages_Call {
+	return &MilvusServiceServer_DumpMessages_Call{Call: _e.mock.On("DumpMessages", _a0, _a1)}
+}
+
+func (_c *MilvusServiceServer_DumpMessages_Call) Run(run func(_a0 *milvuspb.DumpMessagesRequest, _a1 milvuspb.MilvusService_DumpMessagesServer)) *MilvusServiceServer_DumpMessages_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(*milvuspb.DumpMessagesRequest), args[1].(milvuspb.MilvusService_DumpMessagesServer))
+	})
+	return _c
+}
+
+func (_c *MilvusServiceServer_DumpMessages_Call) Return(_a0 error) *MilvusServiceServer_DumpMessages_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *MilvusServiceServer_DumpMessages_Call) RunAndReturn(run func(*milvuspb.DumpMessagesRequest, milvuspb.MilvusService_DumpMessagesServer) error) *MilvusServiceServer_DumpMessages_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Flush provides a mock function with given fields: _a0, _a1
 func (_m *MilvusServiceServer) Flush(_a0 context.Context, _a1 *milvuspb.FlushRequest) (*milvuspb.FlushResponse, error) {
 	ret := _m.Called(_a0, _a1)
@@ -7571,8 +7618,7 @@ func (_c *MilvusServiceServer_Upsert_Call) RunAndReturn(run func(context.Context
 func NewMilvusServiceServer(t interface {
 	mock.TestingT
 	Cleanup(func())
-},
-) *MilvusServiceServer {
+}) *MilvusServiceServer {
 	mock := &MilvusServiceServer{}
 	mock.Mock.Test(t)
 

internal/streamingcoord/server/broadcaster/ack_callback_scheduler.go

@@ -184,15 +184,7 @@ func (s *ackCallbackScheduler) doForcePromoteFixIncompleteBroadcasts(bt *broadca
 	logger.Info("completed fixing incomplete broadcasts for force promote")
 
 	// Acquire resource key lock before executing ack callback.
-	// Force promote runs on a secondary cluster. By step 3, all incomplete tasks have reached
-	// tombstone (their ack callbacks finished, resource key locks released), and no new broadcasts
-	// can arrive (cluster is still secondary). So rkLocker has zero contenders — FastLock always succeeds.
-	s.rkLockerMu.Lock()
-	g, err := s.rkLocker.FastLock(bt.Header().ResourceKeys.Collect()...)
-	s.rkLockerMu.Unlock()
-	if err != nil {
-		panic(fmt.Sprintf("unreachable: FastLock failed during force promote with zero contenders: %v", err))
-	}
+	g := s.rkLocker.Lock(bt.Header().ResourceKeys.Collect()...)
 	s.doAckCallback(bt, g)
 }
 
@@ -238,6 +230,9 @@ func (s *ackCallbackScheduler) fixIncompleteBroadcastsForForcePromote(ctx contex
 		if pending == nil {
 			continue // no pending messages for this task
 		}
+		for i, msg := range pending.pendingMessages {
+			pending.pendingMessages[i] = message.ClearReplicateHeader(msg)
+		}
 		s.Logger().Info("Delegating incomplete task to broadcastScheduler",
 			zap.Uint64("broadcastID", task.Header().BroadcastID),
 			zap.String("messageType", task.msg.MessageType().String()),

internal/streamingcoord/server/broadcaster/broadcast_task.go

@@ -102,7 +102,7 @@ type broadcastTask struct {
 	*taskMetricsGuard
 
 	mu                       sync.Mutex
-	msg                      message.BroadcastMutableMessage
+	msg                      message.BroadcastMutableMessage // protected by mu since MarkIgnore may mutate it.
 	task                     *streamingpb.BroadcastTask
 	dirty                    bool // a flag to indicate that the task has been modified and needs to be saved into the recovery info.
 	done                     chan struct{}
@@ -133,7 +133,7 @@ func (b *broadcastTask) BroadcastResult() (message.BroadcastMutableMessage, map[
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	vchannels := b.msg.BroadcastHeader().VChannels
+	vchannels := b.header().VChannels
 	result := make(map[string]*types.AppendResult, len(vchannels))
 	for idx, vchannel := range vchannels {
 		if b.task.AckedCheckpoints == nil {
@@ -159,8 +159,16 @@ func (b *broadcastTask) BroadcastResult() (message.BroadcastMutableMessage, map[
 }
 
 // Header returns the header of the broadcast task.
+// Must acquire b.mu because MarkIgnore may replace b.msg concurrently.
 func (b *broadcastTask) Header() *message.BroadcastHeader {
-	// header is a immutable field, no need to lock.
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.header()
+}
+
+// header returns the header without acquiring the lock.
+// Caller must hold b.mu.
+func (b *broadcastTask) header() *message.BroadcastHeader {
 	return b.msg.BroadcastHeader()
 }
 
@@ -203,11 +211,15 @@ func (b *broadcastTask) PendingBroadcastMessages() []message.MutableMessage {
 
 // IsAlterReplicateConfigMessage returns true if this task is an AlterReplicateConfig message.
 func (b *broadcastTask) IsAlterReplicateConfigMessage() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
 	return b.msg.MessageType() == message.MessageTypeAlterReplicateConfig
 }
 
 // IsForcePromoteMessage returns true if this task is a force promote AlterReplicateConfig message.
 func (b *broadcastTask) IsForcePromoteMessage() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
 	if b.msg.MessageType() != message.MessageTypeAlterReplicateConfig {
 		return false
 	}
@@ -227,8 +239,17 @@ func (b *broadcastTask) MarkIgnore() error {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	// Parse the message as AlterReplicateConfig
-	msg := message.NewBroadcastMutableMessageBeforeAppend(b.task.Message.Payload, b.task.Message.Properties)
+	// Deep copy properties to avoid mutating the map shared by the old b.msg.
+	// Without this copy, concurrent readers of the old b.msg (e.g., doAckCallback
+	// reading BroadcastHeader via properties.Get) would race with the Set below.
+	origProps := b.task.Message.Properties
+	copiedProps := make(map[string]string, len(origProps))
+	for k, v := range origProps {
+		copiedProps[k] = v
+	}
+
+	// Parse the message as AlterReplicateConfig using the copied properties
+	msg := message.NewBroadcastMutableMessageBeforeAppend(b.task.Message.Payload, copiedProps)
 	alterMsg, err := message.AsMutableAlterReplicateConfigMessageV2(msg)
 	if err != nil {
 		return errors.Wrap(err, "failed to parse message as AlterReplicateConfigMessage")
@@ -237,11 +258,10 @@ func (b *broadcastTask) MarkIgnore() error {
 	// Get current header and set ignore to true
 	header := alterMsg.Header()
 	header.Ignore = true
-	alterMsg.OverwriteHeader(header)
+	alterMsg.OverwriteHeader(header) // writes to copiedProps, not origProps
 
-	// Re-create the broadcast message from updated payload and properties
-	// The OverwriteHeader call above modified the underlying messageImpl properties
-	updatedMsg := message.NewBroadcastMutableMessageBeforeAppend(b.task.Message.Payload, b.task.Message.Properties)
+	// Re-create the broadcast message from the copied (now modified) properties
+	updatedMsg := message.NewBroadcastMutableMessageBeforeAppend(b.task.Message.Payload, copiedProps)
 
 	// Update the task's in-memory message
 	b.task.Message = updatedMsg.IntoMessageProto()
@@ -343,7 +363,9 @@ func (b *broadcastTask) closeAllAcked() {
 // for the operation since 2.6.5, the control channel is always broadcasted.
 // so it's just a dummy function for compatibility.
 func (b *broadcastTask) isControlChannelAcked() bool {
-	for idx, vc := range b.Header().VChannels {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	for idx, vc := range b.header().VChannels {
 		if funcutil.IsControlChannel(vc) && b.task.AckedCheckpoints[idx] != nil {
 			return true
 		}
@@ -380,15 +402,15 @@ func (b *broadcastTask) copyAndSetAckedCheckpoints(msgs ...message.ImmutableMess
 	task := proto.Clone(b.task).(*streamingpb.BroadcastTask)
 	for _, msg := range msgs {
 		vchannel := msg.VChannel()
-		idx, err := findIdxOfVChannel(vchannel, b.Header().VChannels)
+		idx, err := findIdxOfVChannel(vchannel, b.header().VChannels)
 		if err != nil {
 			panic(err)
 		}
 		if len(task.AckedVchannelBitmap) == 0 {
-			task.AckedVchannelBitmap = make([]byte, len(b.Header().VChannels))
+			task.AckedVchannelBitmap = make([]byte, len(b.header().VChannels))
 		}
 		if len(task.AckedCheckpoints) == 0 {
-			task.AckedCheckpoints = make([]*streamingpb.AckedCheckpoint, len(b.Header().VChannels))
+			task.AckedCheckpoints = make([]*streamingpb.AckedCheckpoint, len(b.header().VChannels))
 		}
 		if cp := task.AckedCheckpoints[idx]; cp != nil && cp.TimeTick != 0 {
 			// after proto.Clone, the cp is always not nil, so we also need to check the time tick.
@@ -429,7 +451,7 @@ func (b *broadcastTask) FastAck(ctx context.Context, broadcastResult map[string]
 
 	b.ObserveBroadcastDone()
 
-	if b.Header().AckSyncUp {
+	if b.header().AckSyncUp {
 		// Because the ack sync up is enabled, the ack operation want to be synced up at comsuming side of streaming node,
 		// so we can not make a fast ack operation here to speed up the ack operation.
 		return nil
@@ -506,7 +528,7 @@ func (b *broadcastTask) saveTaskIfDirty(ctx context.Context, logger *log.MLogger
 	}
 	b.dirty = false
 	logger = logger.With(zap.String("state", b.task.State.String()), zap.Int("ackedVChannelCount", ackedCount(b.task)))
-	if err := resource.Resource().StreamingCatalog().SaveBroadcastTask(ctx, b.Header().BroadcastID, b.task); err != nil {
+	if err := resource.Resource().StreamingCatalog().SaveBroadcastTask(ctx, b.header().BroadcastID, b.task); err != nil {
 		logger.Warn("save broadcast task failed", zap.Error(err))
 		if ctx.Err() == nil {
 			panic("critical error: the save broadcast task is failed before the context is done")

internal/streamingcoord/server/broadcaster/force_promote_failover_test.go

@@ -776,6 +776,31 @@ func TestForcePromoteFailover(t *testing.T) {
 	})
 }
 
+func TestForcePromoteLockContention(t *testing.T) {
+	// Verifies that doForcePromoteFixIncompleteBroadcasts succeeds even when the
+	// resource key lock is temporarily held by another goroutine (e.g., a concurrent
+	// ack callback for a replicated DropLoadConfig). Previously this panicked with
+	// "FastLock failed during force promote with zero contenders".
+	initForcePromoteTestGlobals(t)
+
+	vchannels := []string{"cc_vcchan", "v1", "v2"}
+
+	// An incomplete DropCollection task holds the Cluster resource key lock
+	// via its ack callback. The force promote task needs the same lock.
+	env := setupForcePromoteTest(t,
+		[]*streamingpb.BroadcastTask{
+			createReplicatedDropCollectionTask(10, vchannels, []string{"cc_vcchan"}), // cc acked, v1/v2 pending
+			createReplicatedForcePromoteTask(100, vchannels, vchannels),              // all acked
+		},
+		nil,
+	)
+	defer env.close()
+
+	// Both tasks should complete without panic. Task 10's ack callback holds the
+	// resource key lock while running; force promote must wait for it instead of panicking.
+	env.waitForTombstone(t, 30*time.Second, 10, 100)
+}
+
 // --- Record filtering helpers ---
 
 func filterRecords(records []appendRecord, broadcastID uint64) []appendRecord {

internal/streamingnode/client/handler/handler_client_impl.go

@@ -106,7 +106,6 @@ func (hc *handlerClientImpl) GetReplicateCheckpoint(ctx context.Context, pchanne
 }
 
 // GetSalvageCheckpoint gets all salvage checkpoints of the wal.
-// Note: Currently only supports local WAL. Remote WAL support requires streaming proto update.
 func (hc *handlerClientImpl) GetSalvageCheckpoint(ctx context.Context, pchannel string) ([]*wal.ReplicateCheckpoint, error) {
 	if !hc.lifetime.Add(typeutil.LifetimeStateWorking) {
 		return nil, ErrClientClosed
@@ -126,9 +125,21 @@ func (hc *handlerClientImpl) GetSalvageCheckpoint(ctx context.Context, pchannel
 		if !shouldUseRemoteWAL(err) {
 			return nil, err
 		}
-		// The salvage checkpoint is only meaningful on the streaming node that performed the force promote,
-		// so this path should not be reached in practice.
-		return nil, errors.New("GetSalvageCheckpoint is not implemented for remote WAL")
+		handlerService, err := hc.service.GetService(ctx)
+		if err != nil {
+			return nil, err
+		}
+		resp, err := handlerService.GetSalvageCheckpoint(ctx, &streamingpb.GetSalvageCheckpointRequest{
+			Pchannel: types.NewProtoFromPChannelInfo(assign.Channel),
+		})
+		if err != nil {
+			return nil, err
+		}
+		result := make([]*wal.ReplicateCheckpoint, 0, len(resp.GetCheckpoints()))
+		for _, cp := range resp.GetCheckpoints() {
+			result = append(result, utility.NewReplicateCheckpointFromProto(cp))
+		}
+		return result, nil
 	})
 	if err != nil {
 		return nil, err

internal/streamingnode/client/handler/handler_client_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"google.golang.org/grpc"
@@ -180,6 +181,7 @@ func TestHandlerClient_GetSalvageCheckpoint(t *testing.T) {
 	ctx := context.Background()
 
 	// Remote WAL returns "not implemented"; Watch returns Canceled to exit the loop.
+	service.EXPECT().GetService(mock.Anything).Return(nil, errors.New("not implemented"))
 	cps, err := handler.GetSalvageCheckpoint(ctx, "pchannel")
 	assert.Error(t, err)
 	assert.Nil(t, cps)

internal/streamingnode/server/service/handler.go

@@ -3,6 +3,7 @@ package service
 import (
 	"context"
 
+	commonpb "github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/service/handler/consumer"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/service/handler/producer"
 	"github.com/milvus-io/milvus/internal/streamingnode/server/walmanager"
@@ -46,6 +47,20 @@ func (hs *handlerServiceImpl) GetReplicateCheckpoint(ctx context.Context, req *s
 	return &streamingpb.GetReplicateCheckpointResponse{Checkpoint: cp.IntoProto()}, nil
 }
 
+// GetSalvageCheckpoint returns all salvage checkpoints captured during force promote.
+func (hs *handlerServiceImpl) GetSalvageCheckpoint(ctx context.Context, req *streamingpb.GetSalvageCheckpointRequest) (*streamingpb.GetSalvageCheckpointResponse, error) {
+	wal, err := hs.walManager.GetAvailableWAL(types.NewPChannelInfoFromProto(req.GetPchannel()))
+	if err != nil {
+		return nil, err
+	}
+	cps := wal.GetSalvageCheckpoint()
+	protoCps := make([]*commonpb.ReplicateCheckpoint, 0, len(cps))
+	for _, cp := range cps {
+		protoCps = append(protoCps, cp.IntoProto())
+	}
+	return &streamingpb.GetSalvageCheckpointResponse{Checkpoints: protoCps}, nil
+}
+
 // Produce creates a new producer for the channel on this log node.
 func (hs *handlerServiceImpl) Produce(streamServer streamingpb.StreamingNodeHandlerService_ProduceServer) error {
 	p, err := producer.CreateProduceServer(hs.walManager, streamServer)

pkg/mocks/proto/mock_streamingpb/mock_StreamingNodeHandlerServiceClient.go

@@ -172,6 +172,80 @@ func (_c *MockStreamingNodeHandlerServiceClient_GetReplicateCheckpoint_Call) Run
 	return _c
 }
 
+// GetSalvageCheckpoint provides a mock function with given fields: ctx, in, opts
+func (_m *MockStreamingNodeHandlerServiceClient) GetSalvageCheckpoint(ctx context.Context, in *streamingpb.GetSalvageCheckpointRequest, opts ...grpc.CallOption) (*streamingpb.GetSalvageCheckpointResponse, error) {
+	_va := make([]interface{}, len(opts))
+	for _i := range opts {
+		_va[_i] = opts[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, in)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
+
+	if len(ret) == 0 {
+		panic("no return value specified for GetSalvageCheckpoint")
+	}
+
+	var r0 *streamingpb.GetSalvageCheckpointResponse
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *streamingpb.GetSalvageCheckpointRequest, ...grpc.CallOption) (*streamingpb.GetSalvageCheckpointResponse, error)); ok {
+		return rf(ctx, in, opts...)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *streamingpb.GetSalvageCheckpointRequest, ...grpc.CallOption) *streamingpb.GetSalvageCheckpointResponse); ok {
+		r0 = rf(ctx, in, opts...)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*streamingpb.GetSalvageCheckpointResponse)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *streamingpb.GetSalvageCheckpointRequest, ...grpc.CallOption) error); ok {
+		r1 = rf(ctx, in, opts...)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetSalvageCheckpoint'
+type MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call struct {
+	*mock.Call
+}
+
+// GetSalvageCheckpoint is a helper method to define mock.On call
+//   - ctx context.Context
+//   - in *streamingpb.GetSalvageCheckpointRequest
+//   - opts ...grpc.CallOption
+func (_e *MockStreamingNodeHandlerServiceClient_Expecter) GetSalvageCheckpoint(ctx interface{}, in interface{}, opts ...interface{}) *MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call {
+	return &MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call{Call: _e.mock.On("GetSalvageCheckpoint",
+		append([]interface{}{ctx, in}, opts...)...)}
+}
+
+func (_c *MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call) Run(run func(ctx context.Context, in *streamingpb.GetSalvageCheckpointRequest, opts ...grpc.CallOption)) *MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		variadicArgs := make([]grpc.CallOption, len(args)-2)
+		for i, a := range args[2:] {
+			if a != nil {
+				variadicArgs[i] = a.(grpc.CallOption)
+			}
+		}
+		run(args[0].(context.Context), args[1].(*streamingpb.GetSalvageCheckpointRequest), variadicArgs...)
+	})
+	return _c
+}
+
+func (_c *MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call) Return(_a0 *streamingpb.GetSalvageCheckpointResponse, _a1 error) *MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call {
+	_c.Call.Return(_a0, _a1)
+	return _c
+}
+
+func (_c *MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call) RunAndReturn(run func(context.Context, *streamingpb.GetSalvageCheckpointRequest, ...grpc.CallOption) (*streamingpb.GetSalvageCheckpointResponse, error)) *MockStreamingNodeHandlerServiceClient_GetSalvageCheckpoint_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Produce provides a mock function with given fields: ctx, opts
 func (_m *MockStreamingNodeHandlerServiceClient) Produce(ctx context.Context, opts ...grpc.CallOption) (streamingpb.StreamingNodeHandlerService_ProduceClient, error) {
 	_va := make([]interface{}, len(opts))

pkg/proto/streaming.proto

@@ -363,6 +363,10 @@ service StreamingNodeHandlerService {
     // from the correct position, ensuring no duplicate or missing messages.
     rpc GetReplicateCheckpoint(GetReplicateCheckpointRequest) returns (GetReplicateCheckpointResponse) {}
 
+    // GetSalvageCheckpoint returns all salvage checkpoints captured during force promote.
+    // Returns an empty list if no force promote has occurred.
+    rpc GetSalvageCheckpoint(GetSalvageCheckpointRequest) returns (GetSalvageCheckpointResponse) {}
+
     // Produce is a bi-directional streaming RPC to send messages to a channel.
     // All messages sent to a channel will be assigned a unique messageID.
     // The messageID is used to identify the message in the channel.
@@ -391,6 +395,16 @@ message GetReplicateCheckpointResponse {
   common.ReplicateCheckpoint checkpoint = 1;
 }
 
+// GetSalvageCheckpointRequest is the request of GetSalvageCheckpoint service.
+message GetSalvageCheckpointRequest {
+  PChannelInfo pchannel = 1;
+}
+
+// GetSalvageCheckpointResponse is the response of GetSalvageCheckpoint service.
+message GetSalvageCheckpointResponse {
+  repeated common.ReplicateCheckpoint checkpoints = 1;
+}
+
 // ProduceRequest is the request of the Produce RPC.
 // Channel name will be passthrough in the header of stream bu not in the
 // request body.

pkg/proto/streamingpb/streaming_grpc.pb.go

@@ -468,6 +468,7 @@ var StreamingCoordAssignmentService_ServiceDesc = grpc.ServiceDesc{
 
 const (
 	StreamingNodeHandlerService_GetReplicateCheckpoint_FullMethodName = "/milvus.proto.streaming.StreamingNodeHandlerService/GetReplicateCheckpoint"
+	StreamingNodeHandlerService_GetSalvageCheckpoint_FullMethodName   = "/milvus.proto.streaming.StreamingNodeHandlerService/GetSalvageCheckpoint"
 	StreamingNodeHandlerService_Produce_FullMethodName                = "/milvus.proto.streaming.StreamingNodeHandlerService/Produce"
 	StreamingNodeHandlerService_Consume_FullMethodName                = "/milvus.proto.streaming.StreamingNodeHandlerService/Consume"
 )
@@ -479,6 +480,9 @@ type StreamingNodeHandlerServiceClient interface {
 	// GetReplicateCheckpoint returns the WAL checkpoint that will be used to create scanner
 	// from the correct position, ensuring no duplicate or missing messages.
 	GetReplicateCheckpoint(ctx context.Context, in *GetReplicateCheckpointRequest, opts ...grpc.CallOption) (*GetReplicateCheckpointResponse, error)
+	// GetSalvageCheckpoint returns all salvage checkpoints captured during force promote.
+	// Returns an empty list if no force promote has occurred.
+	GetSalvageCheckpoint(ctx context.Context, in *GetSalvageCheckpointRequest, opts ...grpc.CallOption) (*GetSalvageCheckpointResponse, error)
 	// Produce is a bi-directional streaming RPC to send messages to a channel.
 	// All messages sent to a channel will be assigned a unique messageID.
 	// The messageID is used to identify the message in the channel.
@@ -513,6 +517,15 @@ func (c *streamingNodeHandlerServiceClient) GetReplicateCheckpoint(ctx context.C
 	return out, nil
 }
 
+func (c *streamingNodeHandlerServiceClient) GetSalvageCheckpoint(ctx context.Context, in *GetSalvageCheckpointRequest, opts ...grpc.CallOption) (*GetSalvageCheckpointResponse, error) {
+	out := new(GetSalvageCheckpointResponse)
+	err := c.cc.Invoke(ctx, StreamingNodeHandlerService_GetSalvageCheckpoint_FullMethodName, in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *streamingNodeHandlerServiceClient) Produce(ctx context.Context, opts ...grpc.CallOption) (StreamingNodeHandlerService_ProduceClient, error) {
 	stream, err := c.cc.NewStream(ctx, &StreamingNodeHandlerService_ServiceDesc.Streams[0], StreamingNodeHandlerService_Produce_FullMethodName, opts...)
 	if err != nil {
@@ -582,6 +595,9 @@ type StreamingNodeHandlerServiceServer interface {
 	// GetReplicateCheckpoint returns the WAL checkpoint that will be used to create scanner
 	// from the correct position, ensuring no duplicate or missing messages.
 	GetReplicateCheckpoint(context.Context, *GetReplicateCheckpointRequest) (*GetReplicateCheckpointResponse, error)
+	// GetSalvageCheckpoint returns all salvage checkpoints captured during force promote.
+	// Returns an empty list if no force promote has occurred.
+	GetSalvageCheckpoint(context.Context, *GetSalvageCheckpointRequest) (*GetSalvageCheckpointResponse, error)
 	// Produce is a bi-directional streaming RPC to send messages to a channel.
 	// All messages sent to a channel will be assigned a unique messageID.
 	// The messageID is used to identify the message in the channel.
@@ -606,6 +622,9 @@ type UnimplementedStreamingNodeHandlerServiceServer struct {
 func (UnimplementedStreamingNodeHandlerServiceServer) GetReplicateCheckpoint(context.Context, *GetReplicateCheckpointRequest) (*GetReplicateCheckpointResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetReplicateCheckpoint not implemented")
 }
+func (UnimplementedStreamingNodeHandlerServiceServer) GetSalvageCheckpoint(context.Context, *GetSalvageCheckpointRequest) (*GetSalvageCheckpointResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method GetSalvageCheckpoint not implemented")
+}
 func (UnimplementedStreamingNodeHandlerServiceServer) Produce(StreamingNodeHandlerService_ProduceServer) error {
 	return status.Errorf(codes.Unimplemented, "method Produce not implemented")
 }
@@ -642,6 +661,24 @@ func _StreamingNodeHandlerService_GetReplicateCheckpoint_Handler(srv interface{}
 	return interceptor(ctx, in, info, handler)
 }
 
+func _StreamingNodeHandlerService_GetSalvageCheckpoint_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(GetSalvageCheckpointRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StreamingNodeHandlerServiceServer).GetSalvageCheckpoint(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StreamingNodeHandlerService_GetSalvageCheckpoint_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StreamingNodeHandlerServiceServer).GetSalvageCheckpoint(ctx, req.(*GetSalvageCheckpointRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _StreamingNodeHandlerService_Produce_Handler(srv interface{}, stream grpc.ServerStream) error {
 	return srv.(StreamingNodeHandlerServiceServer).Produce(&streamingNodeHandlerServiceProduceServer{stream})
 }
@@ -705,6 +742,10 @@ var StreamingNodeHandlerService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "GetReplicateCheckpoint",
 			Handler:    _StreamingNodeHandlerService_GetReplicateCheckpoint_Handler,
 		},
+		{
+			MethodName: "GetSalvageCheckpoint",
+			Handler:    _StreamingNodeHandlerService_GetSalvageCheckpoint_Handler,
+		},
 	},
 	Streams: []grpc.StreamDesc{
 		{

pkg/streaming/util/message/message_impl.go

@@ -102,9 +102,6 @@ func (m *messageImpl) WithBarrierTimeTick(tt uint64) MutableMessage {
 
 // WithWALTerm sets the wal term of current message.
 func (m *messageImpl) WithWALTerm(term int64) MutableMessage {
-	if m.properties.Exist(messageWALTerm) {
-		panic("wal term already set in properties of message")
-	}
 	m.properties.Set(messageWALTerm, EncodeInt64(term))
 	return m
 }

pkg/streaming/util/message/message_test.go

@@ -154,3 +154,12 @@ func TestCheckIfMessageFromStreaming(t *testing.T) {
 
 func TestReplicateHeader(t *testing.T) {
 }
+
+func TestWithWALTermIdempotent(t *testing.T) {
+	msg := NewMutableMessageBeforeAppend([]byte("payload"), map[string]string{})
+	// Setting WAL term twice should not panic (was a panic before the fix).
+	msg.WithWALTerm(1)
+	assert.NotPanics(t, func() {
+		msg.WithWALTerm(2)
+	})
+}

pkg/streaming/util/message/utils.go

@@ -60,3 +60,18 @@ type ReplicateHeader struct {
 	TimeTick               uint64
 	VChannel               string
 }
+
+// ClearReplicateHeader removes replicate header from a mutable message.
+// Used during force promote fix to re-append as primary messages.
+func ClearReplicateHeader(msg MutableMessage) MutableMessage {
+	if msg == nil {
+		return nil
+	}
+	if impl, ok := msg.(*messageImpl); ok {
+		impl.properties.Delete(messageReplicateMesssageHeader)
+		return impl
+	}
+	raw := msg.Properties().ToRawMap()
+	delete(raw, messageReplicateMesssageHeader)
+	return NewMutableMessageBeforeAppend(msg.Payload(), raw)
+}

pkg/streaming/util/message/utils_test.go

@@ -0,0 +1,42 @@
+package message
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClearReplicateHeader(t *testing.T) {
+	t.Run("nil_message", func(t *testing.T) {
+		result := ClearReplicateHeader(nil)
+		assert.Nil(t, result)
+	})
+
+	t.Run("message_without_replicate_header", func(t *testing.T) {
+		msg := NewMutableMessageBeforeAppend([]byte("payload"), map[string]string{"key": "val"})
+		assert.Nil(t, msg.ReplicateHeader())
+
+		result := ClearReplicateHeader(msg)
+		assert.NotNil(t, result)
+		assert.Nil(t, result.ReplicateHeader())
+	})
+
+	t.Run("message_with_replicate_header_property", func(t *testing.T) {
+		// Directly set the replicate header property to simulate a replicated message.
+		props := map[string]string{
+			messageReplicateMesssageHeader: "some-header-data",
+		}
+		msg := NewMutableMessageBeforeAppend([]byte("payload"), props)
+
+		// Property exists before clear.
+		_, exists := msg.Properties().Get(messageReplicateMesssageHeader)
+		assert.True(t, exists)
+
+		result := ClearReplicateHeader(msg)
+		assert.NotNil(t, result)
+
+		// Property removed after clear.
+		_, exists = result.Properties().Get(messageReplicateMesssageHeader)
+		assert.False(t, exists)
+	})
+}

tests/go_client/go.mod

@@ -4,7 +4,7 @@ go 1.25.8
 
 require (
 	github.com/apache/arrow/go/v17 v17.0.0
-	github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260304062516-e7e54d966ef2
+	github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260309063517-8b5776d31f2b
 	github.com/milvus-io/milvus/client/v2 v2.0.0-20241125024034-0b9edb62a92d
 	github.com/milvus-io/milvus/pkg/v2 v2.6.4-0.20251104142533-a2ce70d25256
 	github.com/minio/minio-go/v7 v7.0.73

tests/go_client/go.sum

@@ -354,8 +354,8 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5
 github.com/mediocregopher/radix/v3 v3.4.2/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i4n7wVopoX3x7Bv8=
 github.com/microcosm-cc/bluemonday v1.0.2/go.mod h1:iVP4YcDBq+n/5fb23BhYFvIMq/leAFZyRl6bYmGDlGc=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260304062516-e7e54d966ef2 h1:OEqYnraL/Tf2nGMjRJji6xaaDe40zSQVBvruzEBZ1hU=
-github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260304062516-e7e54d966ef2/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260309063517-8b5776d31f2b h1:tUp6a72FFeosfjN+8jy/Hax7BqVbtgsaMMVrAfzKnEc=
+github.com/milvus-io/milvus-proto/go-api/v2 v2.6.6-0.20260309063517-8b5776d31f2b/go.mod h1:/6UT4zZl6awVeXLeE7UGDWZvXj3IWkRsh3mqsn0DiAs=
 github.com/milvus-io/milvus/pkg/v2 v2.6.4-0.20251104142533-a2ce70d25256 h1:M2waty0w2k4YT2HHzJk3fx6EFPD4DKxNJatitIV+gGU=
 github.com/milvus-io/milvus/pkg/v2 v2.6.4-0.20251104142533-a2ce70d25256/go.mod h1:HT6Wxahwj/l8+i+D/C3iwDzCjDa36U9gyVw6CjjK4pE=
 github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 h1:AMFGa4R4MiIpspGNG7Z948v4n35fFGB3RR3G/ry4FWs=