fix(multidb): outcome classification and selection races from review

- retryable server replies (LOADING, READONLY, CLUSTERDOWN, ...) now
  classify as availability failures instead of healthy replies, so the
  breaker/detector see them and failover can trigger
- locally synthesized Redis errors (ErrCrossSlot) are neutral: they no
  longer count as proof of a healthy server
- single-command retries honor Cmder.NoRetry: streaming commands record
  the failure but are never replayed
- candidate snapshots, background checks and failover re-checks use a
  non-reserving circuit-state read so they cannot exhaust a recovering
  member's bounded half-open probe budget
- auto-fallback selection runs under failoverMu, closing the same
  removal race fixed for manual failover
- a successful SetActiveIndex probe resets the target's breaker so the
  switch sticks when the member recovered before the grace period

Author: ndyakov

multidb_core.go

@@ -53,6 +53,16 @@ func (db *multidbDatabase) closeClient() error {
 	return db.c.Close()
 }
 
+// selectable reports whether the database's circuit permits selecting it,
+// WITHOUT reserving a half-open probe slot. IsAllowed consumes one of the
+// breaker's bounded half-open requests, so it must only be called right
+// before actually executing a command; candidate snapshots, background
+// checks and failover re-checks use this instead, or repeated selections
+// would exhaust a recovering database's probe budget without ever probing it.
+func (db *multidbDatabase) selectable() bool {
+	return db.cb.CheckState() != imultidb.CircuitOpen
+}
+
 // probe runs the database's health checks under the configured policy,
 // bounded by HealthCheckTimeout, and feeds the result into the circuit
 // breaker and the OTel recorder.
@@ -252,7 +262,7 @@ func (c *multidbCore) candidates(exclude int) []MultiDBDatabaseState {
 		out = append(out, MultiDBDatabaseState{
 			Index:   i,
 			Weight:  db.weight,
-			Allowed: db.cb.IsAllowed(),
+			Allowed: db.selectable(),
 		})
 	}
 	return out
@@ -313,23 +323,26 @@ func (c *multidbCore) process(ctx context.Context, cmd Cmder) error {
 			cmd.SetErr(nil)
 		}
 		err := db.process(ctx, cmd)
-		if err == nil || isRedisReplyError(err) {
-			// A server reply — including error replies like WRONGTYPE and
-			// redis.Nil — proves the database is reachable and healthy.
+		switch classifyOutcome(err) {
+		case outcomeSuccess:
 			db.cb.RecordSuccess()
 			c.detector.RecordSuccess()
 			return err
-		}
-		if !shouldRetry(err, true) {
-			// Client-side errors (context cancellation, deterministic local
-			// rejections) are not database-health signals: return them to
-			// the caller without recording a failure or failing over.
+		case outcomeNeutral:
+			// Not a database-health signal: return to the caller without
+			// recording a failure or failing over.
 			return err
+		case outcomeFailure:
+			db.cb.RecordFailure()
+			c.detector.RecordFailure(err)
+			lastErr = err
+			if cmd.NoRetry() {
+				// Commands that stream into caller-owned writers/buffers
+				// must never be replayed after a partial read: the failure
+				// is recorded, but the error goes straight to the caller.
+				return err
+			}
 		}
-
-		db.cb.RecordFailure()
-		c.detector.RecordFailure(err)
-		lastErr = err
 	}
 	return lastErr
 }
@@ -341,6 +354,42 @@ func isRedisReplyError(err error) bool {
 	return errors.As(err, &redisErr)
 }
 
+// outcomeKind classifies a command outcome for breaker/detector recording.
+type outcomeKind int
+
+const (
+	// outcomeSuccess proves the database served the request (including
+	// definitive error replies like WRONGTYPE or redis.Nil).
+	outcomeSuccess outcomeKind = iota
+	// outcomeFailure is an availability signal: transport-level failures and
+	// retryable server replies (LOADING, READONLY, CLUSTERDOWN, ...).
+	outcomeFailure
+	// outcomeNeutral is not a database-health signal at all: client-side
+	// errors (context cancellation, deterministic local rejections) and
+	// locally synthesized Redis errors such as ErrCrossSlot.
+	outcomeNeutral
+)
+
+// classifyOutcome decides how a command outcome feeds the circuit breaker
+// and the failure detector. Order matters: retryable server replies (LOADING,
+// READONLY, ...) are availability failures even though they are RedisErrors,
+// and locally synthesized RedisErrors (ErrCrossSlot) must not count as proof
+// of a healthy server because no round trip happened.
+func classifyOutcome(err error) outcomeKind {
+	switch {
+	case err == nil:
+		return outcomeSuccess
+	case shouldRetry(err, true):
+		return outcomeFailure
+	case errors.Is(err, ErrCrossSlot):
+		return outcomeNeutral
+	case isRedisReplyError(err):
+		return outcomeSuccess
+	default:
+		return outcomeNeutral
+	}
+}
+
 const (
 	failoverReasonAutomatic = "automatic"
 	failoverReasonManual    = "manual"
@@ -357,7 +406,7 @@ func (c *multidbCore) tryFailover(ctx context.Context, from int) error {
 
 	// Re-check under the lock: a concurrent failover may already have fixed
 	// the active database.
-	if db, idx := c.activeSnapshot(); db != nil && idx != from && db.cb.IsAllowed() {
+	if db, idx := c.activeSnapshot(); db != nil && idx != from && db.selectable() {
 		return nil
 	}
 
@@ -464,6 +513,10 @@ func (c *multidbCore) setActiveIndex(ctx context.Context, index int, probe bool)
 		if !db.probe(ctx, c.opts.HealthCheckTimeout) {
 			return ErrTargetUnhealthy
 		}
+		// The operator asked for this database and a fresh probe just passed:
+		// reset its breaker so a still-open circuit (recovered before the
+		// grace period elapsed) does not immediately fail the switch away.
+		db.cb.Reset()
 	}
 	from := int(c.active.Load())
 	if from == index {
@@ -580,7 +633,7 @@ func (c *multidbCore) startBackgroundLoop() {
 
 			// Background-driven failover: the active index must move even
 			// with no command traffic.
-			if db, idx := c.activeSnapshot(); db != nil && !db.cb.IsAllowed() {
+			if db, idx := c.activeSnapshot(); db != nil && !db.selectable() {
 				_ = c.tryFailover(ctx, idx)
 			}
 
@@ -610,8 +663,13 @@ func (c *multidbCore) runHealthChecksOnce(ctx context.Context) {
 }
 
 // tryFallbackToPrimary switches back to a strictly-higher-weight database
-// whose circuit is closed again.
+// whose circuit is closed again. Selection and switch happen under
+// failoverMu so a concurrent RemoveDatabase (which also holds it) cannot
+// remove the selected member or shift the slice in between.
 func (c *multidbCore) tryFallbackToPrimary(ctx context.Context) {
+	c.failoverMu.Lock()
+	defer c.failoverMu.Unlock()
+
 	active, idx := c.activeSnapshot()
 	if active == nil {
 		return
@@ -633,9 +691,7 @@ func (c *multidbCore) tryFallbackToPrimary(ctx context.Context) {
 	if best < 0 {
 		return
 	}
-	c.failoverMu.Lock()
 	c.switchActive(ctx, idx, best, failoverReasonFallback, 0)
-	c.failoverMu.Unlock()
 }
 
 // newPubSub creates a PubSub whose connections always target the currently

multidb_test.go

@@ -50,11 +50,15 @@ func (hc *fakeHealthCheck) CheckClusterHealth(ctx context.Context, client *redis
 	return hc.healthy.Load(), nil
 }
 
+// customErr lets a test inject an arbitrary error from the hook.
+type customErr struct{ err error }
+
 // hookedDB is a process hook that short-circuits every command (never dials),
 // recording the commands it saw and failing while `fail` is set.
 type hookedDB struct {
 	name     string
 	fail     atomic.Bool
+	custom   atomic.Pointer[customErr]
 	commands atomic.Int64
 }
 
@@ -63,6 +67,10 @@ func (h *hookedDB) DialHook(next redis.DialHook) redis.DialHook { return next }
 func (h *hookedDB) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
 	return func(ctx context.Context, cmd redis.Cmder) error {
 		h.commands.Add(1)
+		if ce := h.custom.Load(); ce != nil {
+			cmd.SetErr(ce.err)
+			return ce.err
+		}
 		if h.fail.Load() {
 			// Wrap io.EOF so the failure classifies as a transport error
 			// (the kind that records on the breaker/detector and retries).
@@ -540,6 +548,102 @@ func TestMultiDBClientSideErrorsDoNotFailOver(t *testing.T) {
 	}
 }
 
+func TestMultiDBLocalRedisErrorIsNeutral(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)
+
+	opts := baseOptions()
+	opts.CommandRetries = 3
+	opts.CircuitBreakerConfig = &redis.MultiDBCircuitBreakerConfig{
+		FailureThreshold: 1,
+		SuccessThreshold: 1,
+		GracePeriod:      time.Hour,
+	}
+	mdb := newTestMultiDB(t, opts, db1, db2)
+	ctx := context.Background()
+
+	// A locally synthesized Redis error (no round trip) must surface to the
+	// caller without failover and without being recorded as a success.
+	db1.hook.custom.Store(&customErr{err: redis.ErrCrossSlot})
+	if err := mdb.Get(ctx, "k").Err(); !errors.Is(err, redis.ErrCrossSlot) {
+		t.Fatalf("Get: err = %v, want ErrCrossSlot", err)
+	}
+	if got := mdb.ActiveIndex(); got != 0 {
+		t.Fatalf("local error caused failover; active = %d", got)
+	}
+	if got := db1.hook.commands.Load(); got != 1 {
+		t.Fatalf("local error was retried; attempts = %d, want 1", got)
+	}
+}
+
+func TestMultiDBNoRetryCommandNotReplayed(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)
+
+	opts := baseOptions()
+	opts.CommandRetries = 3
+	opts.CircuitBreakerConfig = &redis.MultiDBCircuitBreakerConfig{
+		FailureThreshold: 1,
+		SuccessThreshold: 1,
+		GracePeriod:      time.Hour,
+	}
+	mdb := newTestMultiDB(t, opts, db1, db2)
+	ctx := context.Background()
+
+	db1.hook.fail.Store(true)
+
+	cmd := redis.NewRawWriteToCmd(ctx, io.Discard, "get", "k")
+	if err := mdb.Process(ctx, cmd); err == nil {
+		t.Fatal("NoRetry command should surface the transport failure")
+	}
+	// One attempt only: replaying a command that streams into caller-owned
+	// buffers could corrupt output.
+	if got := db1.hook.commands.Load(); got != 1 {
+		t.Fatalf("NoRetry command executed %d times, want 1", got)
+	}
+	if db2.hook.commands.Load() != 0 {
+		t.Error("NoRetry command was replayed on another member")
+	}
+}
+
+func TestMultiDBManualFailoverResetsBreaker(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)
+
+	opts := baseOptions()
+	opts.CommandRetries = 1
+	opts.CircuitBreakerConfig = &redis.MultiDBCircuitBreakerConfig{
+		FailureThreshold: 1,
+		SuccessThreshold: 1,
+		GracePeriod:      time.Hour, // breaker stays open without a reset
+	}
+	mdb := newTestMultiDB(t, opts, db1, db2)
+	ctx := context.Background()
+
+	// Open db2's breaker: force it active while failing, let a command fail.
+	db2.hook.fail.Store(true)
+	if err := mdb.ForceActiveIndex(ctx, 1); err != nil {
+		t.Fatalf("ForceActiveIndex: %v", err)
+	}
+	_ = mdb.Set(ctx, "k", "v", 0).Err() // opens db2's breaker, fails over back to db1
+
+	// db2 recovers before the grace period; the manual probe passes and must
+	// reset the still-open breaker, or the switch would immediately fail away.
+	db2.hook.fail.Store(false)
+	if err := mdb.SetActiveIndex(ctx, 1); err != nil {
+		t.Fatalf("SetActiveIndex after recovery: %v", err)
+	}
+	if got := mdb.ActiveIndex(); got != 1 {
+		t.Fatalf("active = %d, want 1", got)
+	}
+	if err := mdb.Set(ctx, "k", "v", 0).Err(); err != nil {
+		t.Fatalf("Set on manually selected member: %v", err)
+	}
+	if got := mdb.ActiveIndex(); got != 1 {
+		t.Fatalf("switch did not stick; active = %d", got)
+	}
+}
+
 func TestMultiDBValidation(t *testing.T) {
 	ctx := context.Background()