fix: return the replicate configuration of an unconfigured cluster instead of blocking (#52687)

issue: #52686

## Problem

`GetReplicateConfiguration` never returns on a cluster that has never
been given a
replicate configuration — it blocks until the caller's deadline and
surfaces as
`DEADLINE_EXCEEDED`, which reads as an unhealthy cluster. Every freshly
installed
cluster is in that state, so reading the topology before configuring
replication
appears to fail.

A cluster with no replication has no configuration to report, and that
is nil all the
way down: the balancer pushes a nil `ReplicateConfiguration` when
`cm.replicateConfig`
is nil, `NewConfigHelper` maps a nil config to a nil helper, and the
watcher's wait
condition treated that nil helper as "no assignment received yet":

```go
for (…Version.Global == -1 && …Version.Local == -1) ||
    w.lastVersionedAssignment.ReplicateConfigHelper == nil {
```

The second term can never be satisfied there. The sibling accessors on
the same watcher,
`GetLatestStreamingVersion` and `GetLatestDiscover`, wait only on the
version.

Sending an empty configuration from the balancer instead is not viable:
`NewConfigHelper`
rejects a configuration with no clusters (`primary count is not 1`), so
`MustNewConfigHelper` would panic in every client. A nil helper *is* the
representation
of "no replication configured".

#47517 made this deterministic rather than intermittent: the public API
now always takes
the fresh-read path, which builds a throwaway watcher and waits for one
assignment push,
so a genuinely fresh cluster fails every time while a
previously-configured one still
answers from its long-lived watcher.

## Change

- `watcher.GetLatestReplicateConfiguration` waits only until an
assignment has been
  received, and returns the possibly-nil helper.
- `ConfigHelper.GetReplicateConfiguration` and `GetCluster` are
nil-receiver safe, so
  callers can use the result directly.
- `overwriteReplicateMessage` rejects a replicate message as a replicate
violation when
the cluster has no configuration. This state is reached in the normal
bootstrap order,
when the primary has been given the replicate configuration and the
secondary has not
yet; it previously blocked on the configuration query and is now
rejected explicitly.
- `replicateStreamClient` no longer resets its reconnect backoff as soon
as
`CreateReplicateStream` succeeds. Against a secondary that accepts the
stream but
rejects every message, the retry interval never left its 100 ms initial
value and each
round re-sent the pending messages from the head. The backoff is now
reset only once a
connection has proven healthy: the target confirmed at least one
message, or the stream
was closed by the idle timeout because there was nothing to send.
Rejected connections
  back off from 100 ms up to 10 s.

## Tests

- `blocks_when_version_set_but_config_nil` asserted the old behaviour
(deadline). It is
replaced by `returns_nil_when_version_set_but_config_nil`, which asserts
a prompt nil
  result and that the read accessors stay usable on it.
- New `blocks_until_first_assignment_arrives` keeps the genuine "not
known yet" case
covered: with no assignment at all the call still blocks, and returns
once one arrives.
- New `TestReplicateServiceNoConfig`: the configuration query answers
`(nil, nil)` and
  `Append` returns `STREAMING_CODE_REPLICATE_VIOLATION`.
- New `TestReplicateStreamClient_BackoffNotResetWhileTargetRejects`: the
stream opens but
`Recv` fails with `FailedPrecondition`; after five reconnects the
backoff was never
reset. `TestReplicateStreamClient_BackoffResetAfterIdleTimeout`: `Recv`
fails with the
  stream idle timeout; the backoff is reset on every round.
- `go test ./internal/cdc/replication/replicatestream/
./internal/distributed/streaming/
./internal/streamingcoord/client/assignment/` and `golangci-lint` on the
same packages
  pass on current master.

Signed-off-by: zhenshan.cao <[email protected]>
Signed-off-by: Zack <[email protected]>
Co-authored-by: Zack <[email protected]>

internal/cdc/replication/replicatestream/replicate_stream_client_impl.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
@@ -48,6 +49,10 @@ type replicateStreamClient struct {
 	channel         *meta.ReplicateChannel
 	pendingMessages MsgQueue
 	metrics         ReplicateMetrics
+	// connConfirmed records whether the target confirmed at least one message
+	// on the current connection. It decides whether the reconnect backoff is
+	// reset once the connection ends.
+	connConfirmed atomic.Bool
 
 	ctx        context.Context
 	cancel     context.CancelFunc
@@ -80,6 +85,16 @@ func NewReplicateStreamClient(ctx context.Context, c cluster.MilvusClient, chann
 	return rs
 }
 
+// newReconnectBackOff builds the backoff that paces reconnects of a replicate
+// stream. It is a variable so tests can observe when the backoff is reset.
+var newReconnectBackOff = func() backoff.BackOff {
+	b := backoff.NewExponentialBackOff()
+	b.InitialInterval = 100 * time.Millisecond
+	b.MaxInterval = 10 * time.Second
+	b.MaxElapsedTime = 0
+	return b
+}
+
 func (r *replicateStreamClient) startInternal() {
 	defer func() {
 		mlog.Info(r.ctx, "replicate stream client closed",
@@ -89,10 +104,7 @@ func (r *replicateStreamClient) startInternal() {
 		close(r.finishedCh)
 	}()
 
-	backoff := backoff.NewExponentialBackOff()
-	backoff.InitialInterval = 100 * time.Millisecond
-	backoff.MaxInterval = 10 * time.Second
-	backoff.MaxElapsedTime = 0
+	backoff := newReconnectBackOff()
 
 	for {
 		restart := r.startReplicating(backoff)
@@ -124,7 +136,7 @@ func (r *replicateStreamClient) startReplicating(backoff backoff.BackOff) (needR
 
 	logger.Info(r.ctx, "replicate stream client service started")
 	r.metrics.OnConnect()
-	backoff.Reset()
+	r.connConfirmed.Store(false)
 
 	// reset client and pending messages
 	r.client = client
@@ -144,6 +156,16 @@ func (r *replicateStreamClient) startReplicating(backoff backoff.BackOff) (needR
 	<-sendCh
 	<-recvCh // wait for send/recv loops to exit
 
+	// The backoff is reset only once this connection has proven healthy: the
+	// target confirmed at least one message, or the stream was closed by the
+	// idle timeout because there was nothing to send. Resetting right after
+	// CreateReplicateStream would pin the retry interval at its initial value
+	// against a target that accepts the stream but rejects every message, such
+	// as a secondary that has not been given its replicate configuration yet.
+	if r.connConfirmed.Load() || isStreamIdleTimeout(chErr) {
+		backoff.Reset()
+	}
+
 	if r.ctx.Err() != nil {
 		logger.Info(r.ctx, "close replicate stream client due to ctx done")
 		return false
@@ -308,6 +330,7 @@ func (r *replicateStreamClient) recvLoop(ctx context.Context) (err error) {
 			}
 			lastConfirmedMessageInfo := resp.GetReplicateConfirmedMessageInfo()
 			if lastConfirmedMessageInfo != nil {
+				r.connConfirmed.Store(true)
 				messages := r.pendingMessages.CleanupConfirmedMessages(lastConfirmedMessageInfo.GetConfirmedTimeTick())
 				for _, msg := range messages {
 					r.metrics.OnConfirmed(msg)

internal/cdc/replication/replicatestream/replicate_stream_client_impl_test.go

@@ -20,9 +20,11 @@ import (
 	"context"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"google.golang.org/grpc"
@@ -378,3 +380,75 @@ func TestIsStreamIdleTimeout(t *testing.T) {
 	// Non-gRPC errors should not match
 	assert.False(t, isStreamIdleTimeout(assert.AnError))
 }
+
+// countingBackOff records how often the reconnect loop resets it and never
+// delays, so a test can drive many reconnects quickly.
+type countingBackOff struct {
+	resets atomic.Int32
+}
+
+func (b *countingBackOff) NextBackOff() time.Duration { return 0 }
+
+func (b *countingBackOff) Reset() { b.resets.Add(1) }
+
+// runReconnects starts a replicate stream client against a target whose stream
+// opens but whose Recv fails with recvErr, lets it reconnect rounds times, and
+// returns how often the backoff was reset in the meantime.
+func runReconnects(t *testing.T, recvErr error, rounds int) int32 {
+	bo := &countingBackOff{}
+	origNewBackOff := newReconnectBackOff
+	newReconnectBackOff = func() backoff.BackOff { return bo }
+	defer func() { newReconnectBackOff = origNewBackOff }()
+
+	done := make(chan struct{})
+	var doneOnce sync.Once
+	var connects atomic.Int32
+	mockMilvusClient := cluster.NewMockMilvusClient(t)
+	mockMilvusClient.EXPECT().CreateReplicateStream(mock.Anything).RunAndReturn(func(ctx context.Context, opts ...grpc.CallOption) (milvuspb.MilvusService_CreateReplicateStreamClient, error) {
+		if connects.Add(1) >= int32(rounds) {
+			doneOnce.Do(func() { close(done) })
+		}
+		sc := newMockReplicateStreamClient(t)
+		sc.recvError = recvErr
+		return sc, nil
+	}).Maybe()
+
+	wal := mock_streaming.NewMockWALAccesser(t)
+	streaming.SetWALForTest(wal)
+
+	channel := &meta.ReplicateChannel{
+		Key: "test-replicate-key",
+		Value: &streamingpb.ReplicatePChannelMeta{
+			SourceChannelName: "test-source-channel",
+			TargetChannelName: "test-target-channel",
+			TargetCluster: &commonpb.MilvusCluster{
+				ClusterId:       "test-cluster",
+				ConnectionParam: &commonpb.ConnectionParam{Uri: "localhost:19530", Token: "test-token"},
+			},
+		},
+	}
+	replicateClient := NewReplicateStreamClient(context.Background(), mockMilvusClient, channel)
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("replicate stream client did not reconnect enough times")
+	}
+	replicateClient.Close()
+	return bo.resets.Load()
+}
+
+// The reconnect backoff must keep growing while the target opens the stream but
+// rejects what is sent on it, as a secondary without a replicate configuration
+// does; a reset on every successful CreateReplicateStream would pin the retry
+// interval at its initial value for as long as the target stays unconfigured.
+func TestReplicateStreamClient_BackoffNotResetWhileTargetRejects(t *testing.T) {
+	rejected := status.Error(codes.FailedPrecondition, "cluster has no replicate configuration, cannot receive replicate message")
+	assert.Equal(t, int32(0), runReconnects(t, rejected, 5))
+}
+
+// An idle timeout means the stream was healthy and simply had nothing to carry,
+// so the backoff is reset and the next reconnect is not delayed.
+func TestReplicateStreamClient_BackoffResetAfterIdleTimeout(t *testing.T) {
+	idle := status.Error(codes.Unknown, "rpc error: stream timeout")
+	assert.GreaterOrEqual(t, runReconnects(t, idle, 5), int32(4))
+}

internal/distributed/streaming/replicate_service.go

@@ -120,6 +120,9 @@ func (s replicateService) overwriteReplicateMessage(ctx context.Context, msg mes
 	if err != nil {
 		return nil, err
 	}
+	if cfg == nil {
+		return nil, status.NewReplicateViolation("cluster has no replicate configuration, cannot receive replicate message")
+	}
 
 	// Get target vchannel on current cluster that should be written to
 	currentCluster := cfg.GetCluster(s.clusterID)

internal/distributed/streaming/replicate_service_test.go

@@ -1341,6 +1341,35 @@ func TestReplicateServiceGetConfigError(t *testing.T) {
 	assert.Contains(t, err.Error(), "config unavailable")
 }
 
+// A cluster that has never been given a replicate configuration answers the
+// configuration query with a nil helper. Appending a replicate message to it is
+// rejected as a replicate violation instead of being dereferenced.
+func TestReplicateServiceNoConfig(t *testing.T) {
+	c := mock_client.NewMockClient(t)
+	as := mock_client.NewMockAssignmentService(t)
+	c.EXPECT().Assignment().Return(as).Maybe()
+	h := mock_handler.NewMockHandlerClient(t)
+
+	as.EXPECT().GetReplicateConfiguration(mock.Anything).Return(nil, nil)
+
+	rs := &replicateService{
+		walAccesserImpl: &walAccesserImpl{
+			lifetime:             typeutil.NewLifetime(),
+			clusterID:            "by-dev",
+			streamingCoordClient: c,
+			handlerClient:        h,
+			producers:            make(map[string]*producer.ResumableProducer),
+		},
+	}
+
+	replicateMsgs := createReplicateCreateCollectionMessages()
+	_, err := rs.Append(context.Background(), replicateMsgs[0])
+	assert.Error(t, err)
+	se := status.AsStreamingError(err)
+	assert.NotNil(t, se)
+	assert.Equal(t, streamingpb.StreamingCode_STREAMING_CODE_REPLICATE_VIOLATION, se.Code)
+}
+
 func createSimpleReplicateAlterConfigMessages(newConfig *commonpb.ReplicateConfiguration) []message.ReplicateMutableMessage {
 	alterMsg := message.NewAlterReplicateConfigMessageBuilderV2().
 		WithHeader(&message.AlterReplicateConfigMessageHeader{

internal/streamingcoord/client/assignment/assignment_test.go

@@ -378,21 +378,59 @@ func TestWatcher_GetLatestReplicateConfiguration(t *testing.T) {
 		assert.NoError(t, <-errCh)
 	})
 
-	t.Run("blocks_when_version_set_but_config_nil", func(t *testing.T) {
+	t.Run("returns_nil_when_version_set_but_config_nil", func(t *testing.T) {
+		// A cluster that has never been given a replicate configuration reports a
+		// nil config in its assignment. That is an answer, not a missing update:
+		// the call must return promptly with a nil helper instead of blocking
+		// until the deadline.
 		w := newWatcher()
 
-		// Update with valid version but nil config
 		w.Update(types.VersionedStreamingNodeAssignments{
 			Version:               typeutil.VersionInt64Pair{Global: 1, Local: 1},
 			Assignments:           map[int64]types.StreamingNodeAssignment{},
 			ReplicateConfigHelper: nil,
 		})
 
-		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
 
-		_, err := w.GetLatestReplicateConfiguration(ctx)
-		assert.ErrorIs(t, err, context.DeadlineExceeded)
+		config, err := w.GetLatestReplicateConfiguration(ctx)
+		assert.NoError(t, err)
+		assert.Nil(t, config)
+		// The nil helper stays usable for the read accessors.
+		assert.Nil(t, config.GetReplicateConfiguration())
+		assert.Nil(t, config.GetCluster("any"))
+	})
+
+	t.Run("blocks_until_first_assignment_arrives", func(t *testing.T) {
+		// With no assignment received at all the call still blocks: that is a
+		// genuine "not known yet", unlike a received-but-nil configuration.
+		w := newWatcher()
+
+		resultCh := make(chan error, 1)
+		go func() {
+			_, err := w.GetLatestReplicateConfiguration(context.Background())
+			resultCh <- err
+		}()
+
+		select {
+		case <-resultCh:
+			t.Fatal("GetLatestReplicateConfiguration should block before the first assignment")
+		case <-time.After(50 * time.Millisecond):
+		}
+
+		w.Update(types.VersionedStreamingNodeAssignments{
+			Version:               typeutil.VersionInt64Pair{Global: 1, Local: 1},
+			Assignments:           map[int64]types.StreamingNodeAssignment{},
+			ReplicateConfigHelper: nil,
+		})
+
+		select {
+		case err := <-resultCh:
+			assert.NoError(t, err)
+		case <-time.After(time.Second):
+			t.Fatal("GetLatestReplicateConfiguration should have returned")
+		}
 	})
 
 	t.Run("returns_error_on_context_cancel", func(t *testing.T) {

internal/streamingcoord/client/assignment/watcher.go

@@ -57,10 +57,15 @@ func (w *watcher) GetLatestDiscover(ctx context.Context) (*types.VersionedStream
 	return &last, nil
 }
 
+// GetLatestReplicateConfiguration returns the replicate configuration of the
+// latest assignment. It waits until an assignment has been received, and only
+// for that: a nil ConfigHelper is a valid answer meaning "this cluster has no
+// replication configured", which is the steady state of every cluster that has
+// never been given one. Waiting for a non-nil helper would block forever there
+// and surface as DEADLINE_EXCEEDED, indistinguishable from an unhealthy cluster.
 func (w *watcher) GetLatestReplicateConfiguration(ctx context.Context) (*replicateutil.ConfigHelper, error) {
 	w.cond.L.Lock()
-	for (w.lastVersionedAssignment.Version.Global == -1 && w.lastVersionedAssignment.Version.Local == -1) ||
-		w.lastVersionedAssignment.ReplicateConfigHelper == nil {
+	for w.lastVersionedAssignment.Version.Global == -1 && w.lastVersionedAssignment.Version.Local == -1 {
 		if err := w.cond.Wait(ctx); err != nil {
 			return nil, err
 		}

pkg/util/replicateutil/config_helper.go

@@ -128,6 +128,10 @@ type ConfigHelper struct {
 
 // GetReplicateConfiguration returns the replicate configuration of the graph.
 func (g *ConfigHelper) GetReplicateConfiguration() *commonpb.ReplicateConfiguration {
+	if g == nil {
+		// A nil helper is how "no replication configured" is represented.
+		return nil
+	}
 	return g.cfg
 }
 
@@ -145,6 +149,9 @@ func (g *ConfigHelper) IsJoinReplication() bool {
 
 // GetCluster returns the cluster from the graph.
 func (g *ConfigHelper) GetCluster(clusterID string) *MilvusCluster {
+	if g == nil {
+		return nil
+	}
 	return g.vs[clusterID]
 }