fix(multidb): autopipeliner-vs-cluster-member lifecycle from review

- a closed autopipeliner no longer blocks AddDatabase(cluster) forever
  (the cached slot is checked for liveness, matching getOrCreate)
- the cluster-member check moved inside the pipeliner build function and
  AddDatabase holds autopipelinerMu across check+add, so creation and
  cluster additions are fully serialized

Author: ndyakov

multidb.go

@@ -461,16 +461,21 @@ func (c *MultiDBClient) ForceActiveIndex(ctx context.Context, index int) error {
 }
 
 // AddDatabase implements MultiDBCtrl. Adding a cluster member is refused
-// while an autopipeliner instance exists: the cached autopipeliner was built
-// without the cluster-specific safeguards and a later failover onto the new
-// member would route merged batches around them. Close the autopipeliners
-// first, add the member, then recreate them.
+// while a live autopipeliner instance exists: the cached autopipeliner was
+// built without the cluster-specific safeguards and a later failover onto the
+// new member would route merged batches around them. Close the autopipeliners
+// first, add the member, then recreate them (closed instances do not block).
 func (c *MultiDBClient) AddDatabase(ctx context.Context, cfg MultiDBClientConfig) (int, error) {
 	if cfg.ClusterOptions != nil {
+		// Hold autopipelinerMu across the check AND the membership change so
+		// a concurrent first AutoPipeline call (which creates the pipeliner
+		// under the same mutex and re-checks membership there) cannot
+		// interleave between them.
 		c.autopipelinerMu.Lock()
-		hasAP := c.autopipeliner != nil || c.asyncAutopipeliner != nil
-		c.autopipelinerMu.Unlock()
-		if hasAP {
+		defer c.autopipelinerMu.Unlock()
+		hasLiveAP := (c.autopipeliner != nil && !c.autopipeliner.closed.Load()) ||
+			(c.asyncAutopipeliner != nil && !c.asyncAutopipeliner.closed.Load())
+		if hasLiveAP {
 			return -1, errors.New("redis: multidb: close autopipeliners before adding a cluster member database")
 		}
 	}

multidb_autopipeline_test.go

@@ -231,6 +231,42 @@ func TestMultiDBAutoPipelineRefusesClusterMembers(t *testing.T) {
 	}
 }
 
+func TestMultiDBAddClusterMemberVsAutoPipeliner(t *testing.T) {
+	db1 := newTestDB("db1", "127.0.0.1:1", 2.0, true)
+	mdb := newTestMultiDB(t, baseOptions(), db1)
+	ctx := context.Background()
+
+	ap, err := mdb.AutoPipeline()
+	if err != nil {
+		t.Fatalf("AutoPipeline: %v", err)
+	}
+
+	clusterCfg := redis.MultiDBClientConfig{
+		ClusterOptions:         &redis.ClusterOptions{Addrs: []string{"127.0.0.1:2"}},
+		Weight:                 1.0,
+		HealthChecks:           []redis.MultiDBHealthCheck{newFakeHealthCheck(true)},
+		SkipInitialHealthCheck: true,
+	}
+
+	// Live autopipeliner blocks cluster additions.
+	if _, err := mdb.AddDatabase(ctx, clusterCfg); err == nil {
+		t.Fatal("AddDatabase(cluster) should be refused while an autopipeliner is live")
+	}
+
+	// A CLOSED autopipeliner must not block forever.
+	if err := ap.Close(); err != nil {
+		t.Fatalf("AutoPipeliner.Close: %v", err)
+	}
+	if _, err := mdb.AddDatabase(ctx, clusterCfg); err != nil {
+		t.Fatalf("AddDatabase(cluster) after closing the autopipeliner: %v", err)
+	}
+
+	// And with a cluster member present, new autopipeliners are refused.
+	if _, err := mdb.AutoPipeline(); err == nil {
+		t.Error("AutoPipeline should be refused once a cluster member exists")
+	}
+}
+
 func TestMultiDBAutoPipelineRoutesAndFailsOver(t *testing.T) {
 	db1 := newTestDB("db1", "127.0.0.1:1", 2.0, true)
 	db2 := newTestDB("db2", "127.0.0.1:2", 1.0, true)

multidb_pipeline.go

@@ -314,12 +314,17 @@ func (c *MultiDBClient) AutoPipeline() (*AutoPipeliner, error) {
 //
 // EXPERIMENTAL: this API is subject to change, use with caution.
 func (c *MultiDBClient) AutoPipelineWithOptions(config *AutoPipelineOptions) (*AutoPipeliner, error) {
-	if c.core.hasClusterMember() {
-		return nil, errMultiDBAutoPipelineCluster
-	}
+	// The cluster-member check runs inside the build function, under
+	// autopipelinerMu: AddDatabase performs its cluster-vs-autopipeliner
+	// check under the same mutex, so the two cannot interleave.
 	return getOrCreateAutoPipeliner(c.autopipelinerMu, &c.autopipeliner, &c.autopipelinerClosed, nil, config,
 		DefaultBlockingAutoPipelineOptions,
-		func(cfg *AutoPipelineOptions) (*AutoPipeliner, error) { return newAutoPipeliner(c, cfg, true) })
+		func(cfg *AutoPipelineOptions) (*AutoPipeliner, error) {
+			if c.core.hasClusterMember() {
+				return nil, errMultiDBAutoPipelineCluster
+			}
+			return newAutoPipeliner(c, cfg, true)
+		})
 }
 
 // AsyncAutoPipeline returns the deferred autopipeliner for this MultiDB
@@ -334,10 +339,12 @@ func (c *MultiDBClient) AsyncAutoPipeline() (*AutoPipeliner, error) {
 //
 // EXPERIMENTAL: this API is subject to change, use with caution.
 func (c *MultiDBClient) AsyncAutoPipelineWithOptions(config *AutoPipelineOptions) (*AutoPipeliner, error) {
-	if c.core.hasClusterMember() {
-		return nil, errMultiDBAutoPipelineCluster
-	}
 	return getOrCreateAutoPipeliner(c.autopipelinerMu, &c.asyncAutopipeliner, &c.autopipelinerClosed, nil, config,
 		DefaultAutoPipelineOptions,
-		func(cfg *AutoPipelineOptions) (*AutoPipeliner, error) { return newAutoPipeliner(c, cfg, false) })
+		func(cfg *AutoPipelineOptions) (*AutoPipeliner, error) {
+			if c.core.hasClusterMember() {
+				return nil, errMultiDBAutoPipelineCluster
+			}
+			return newAutoPipeliner(c, cfg, false)
+		})
 }