diff --git a/internal/cdc/replication/replicatestream/replicate_stream_client_impl.go b/internal/cdc/replication/replicatestream/replicate_stream_client_impl.go
index 065479eec9..5a09e8ceaf 100644
--- a/internal/cdc/replication/replicatestream/replicate_stream_client_impl.go
+++ b/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)
diff --git a/internal/cdc/replication/replicatestream/replicate_stream_client_impl_test.go b/internal/cdc/replication/replicatestream/replicate_stream_client_impl_test.go
index fefcc31f8e..ab85cce1d3 100644
--- a/internal/cdc/replication/replicatestream/replicate_stream_client_impl_test.go
+++ b/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))
+}
diff --git a/internal/distributed/streaming/replicate_service.go b/internal/distributed/streaming/replicate_service.go
index bf72f56f22..1e7ff4cfba 100644
--- a/internal/distributed/streaming/replicate_service.go
+++ b/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)
diff --git a/internal/distributed/streaming/replicate_service_test.go b/internal/distributed/streaming/replicate_service_test.go
index 1193c4f00a..924dae5900 100644
--- a/internal/distributed/streaming/replicate_service_test.go
+++ b/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{
diff --git a/internal/streamingcoord/client/assignment/assignment_test.go b/internal/streamingcoord/client/assignment/assignment_test.go
index f83ccdd4f8..2b553a3e8e 100644
--- a/internal/streamingcoord/client/assignment/assignment_test.go
+++ b/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) {
diff --git a/internal/streamingcoord/client/assignment/watcher.go b/internal/streamingcoord/client/assignment/watcher.go
index 67a8337efa..cd396833ba 100644
--- a/internal/streamingcoord/client/assignment/watcher.go
+++ b/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
 		}
diff --git a/pkg/util/replicateutil/config_helper.go b/pkg/util/replicateutil/config_helper.go
index 7c28360714..c7eded5970 100644
--- a/pkg/util/replicateutil/config_helper.go
+++ b/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]
 }