backend: handle LOADING the same way as MASTERDOWN

Author: fancy-rabbit

pkg/proxy/backend.go

@@ -22,6 +22,7 @@ import (
 const (
 	stateConnected = 1
 	stateDataStale = 2
+	stateLoading   = 3
 )
 
 type BackendConn struct {
@@ -137,6 +138,55 @@ func (bc *BackendConn) KeepAlive() bool {
 					bc, bc.addr, bc.database)
 			}
 		}
+
+	case stateLoading:
+		m := &Request{}
+		m.Multi = []*redis.Resp{
+			redis.NewBulkBytes([]byte("INFO")),
+		}
+		m.Batch = &sync.WaitGroup{}
+		bc.PushBack(m)
+
+		keepAliveCallback <- func() {
+			m.Batch.Wait()
+			var err = func() error {
+				if err := m.Err; err != nil {
+					return err
+				}
+				switch resp := m.Resp; {
+				case resp == nil:
+					return ErrRespIsRequired
+				case resp.IsError():
+					return fmt.Errorf("bad info resp: %s", resp.Value)
+				case resp.IsBulkBytes():
+					var info = make(map[string]string)
+					for _, line := range strings.Split(string(resp.Value), "\n") {
+						kv := strings.SplitN(line, ":", 2)
+						if len(kv) != 2 {
+							continue
+						}
+						if key := strings.TrimSpace(kv[0]); key != "" {
+							info[key] = strings.TrimSpace(kv[1])
+						}
+					}
+					if info["loading"] == "1" {
+						return nil
+					}
+					if bc.state.CompareAndSwap(stateLoading, stateConnected) {
+						log.Warnf("backend conn [%p] to %s, db-%d state = Connected (keepalive)",
+							bc, bc.addr, bc.database)
+					}
+					return nil
+				default:
+					return fmt.Errorf("bad info resp: should be string, but got %s", resp.Type)
+				}
+			}()
+			if err != nil && bc.closed.IsFalse() {
+				log.WarnErrorf(err, "backend conn [%p] to %s, db-%d recover from Loading failed",
+					bc, bc.addr, bc.database)
+			}
+		}
+
 	}
 	return true
 }
@@ -266,6 +316,7 @@ func (bc *BackendConn) run() {
 }
 
 var errMasterDown = []byte("MASTERDOWN")
+var errLoading = []byte("LOADING")
 
 func (bc *BackendConn) loopReader(tasks <-chan *Request, c *redis.Conn, round int) (err error) {
 	defer func() {
@@ -288,6 +339,11 @@ func (bc *BackendConn) loopReader(tasks <-chan *Request, c *redis.Conn, round in
 					log.Warnf("backend conn [%p] to %s, db-%d state = DataStale, caused by 'MASTERDOWN'",
 						bc, bc.addr, bc.database)
 				}
+			case bytes.HasPrefix(resp.Value, errLoading):
+				if bc.state.CompareAndSwap(stateConnected, stateLoading) {
+					log.Warnf("backend conn [%p] to %s, db-%d state = Loading",
+						bc, bc.addr, bc.database)
+				}
 			}
 		}
 		bc.setResponse(r, resp, nil)