// OnFailedSync always returns a 10 second wait period between failed /syncs, never a fatal error.
977652
func (s *DefaultSyncer) OnFailedSync(res *RespSync, err error) (time.Duration, error) {
	return 10 * time.Second, nil
}
683122
func (p *grpcPeer) monitorLoopWait(ctx context.Context, cancel context.CancelFunc, connectivityState connectivity.State) (connectivity.State, bool) {
	changedC := make(chan bool, 1)
	go func() { changedC <- p.clientConn.WaitForStateChange(ctx, connectivityState) }()

	loop := false
	select {
	case changed := <-changedC:
		if cancel != nil {
			cancel()
		}
		if changed {
			connectivityState = p.clientConn.GetState()
		}
		loop = true
	case <-p.stoppingC:
	case <-p.t.once.Stopping():
		if cancel != nil {
			cancel()
		}
	}
	return connectivityState, loop
}
1015930
func (g *GossipSyncer) ProcessSyncTransition(newSyncType SyncerType) error {
	errChan := make(chan error, 1)
	select {
	case g.syncTransitionReqs <- &syncTransitionReq{
		newSyncType: newSyncType,
		errChan:     errChan,
	}:
	case <-time.After(syncTransitionTimeout):
		return ErrSyncTransitionTimeout
	case <-g.quit:
		return ErrGossipSyncerExiting
	}

	select {
	case err := <-errChan:
		return err
	case <-g.quit:
		return ErrGossipSyncerExiting
	}
}
788409
func (s *Counters) SetFailed(v int64) *Counters {
	s.Failed = &v
	return s
}
1212403
func (xc *CommonExchange) IsFailed() bool {
	xc.mtx.RLock()
	defer xc.mtx.RUnlock()
	return xc.lastFail.After(xc.lastUpdate)
}
1021183
func (m *SyncManager) forceHistoricalSync() *GossipSyncer {
	m.syncersMu.Lock()
	defer m.syncersMu.Unlock()

	// We'll sample from both sets of active and inactive syncers in the
	// event that we don't have any inactive syncers.
	return chooseRandomSyncer(m.gossipSyncers(), func(s *GossipSyncer) error {
		return s.historicalSync()
	})
}
1213829
func asyncNotifyEvent(d *ddlCtx, e *util.Event) {
	if d.ddlEventCh != nil {
		if d.lease == 0 {
			// If lease is 0, it's always used in test.
			select {
			case d.ddlEventCh <- e:
			default:
			}
			return
		}
		for i := 0; i < 10; i++ {
			select {
			case d.ddlEventCh <- e:
				return
			default:
				logutil.Logger(ddlLogCtx).Warn("[ddl] fail to notify DDL event", zap.String("event", e.String()))
				time.Sleep(time.Microsecond * 10)
			}
		}
	}
}
642595
func (e *Executor) FailNextCallToExecute(err error) {
	e.mu.Lock()
	e.nextExecuteErr = err
	e.mu.Unlock()
}
107115
func (a *api) RegistrySync() error {
	client, err := a.connectMaster()
	if err != nil {
		return err
	}
	return client.SyncRegistry()
}
878313
func (queue *Queue) RequeueFailed() error {
	l := queue.GetFailedLength()
	// TODO implement this in lua
	for l > 0 {
		err := queue.redisClient.RPopLPush(queueFailedKey(queue.Name), queueInputKey(queue.Name)).Err()
		if err != nil {
			return err
		}
		queue.incrRate(queueInputRateKey(queue.Name), 1)
		l--
	}
	return nil
}
686131
func (g *GossipSyncer) historicalSync() error {
	done := make(chan struct{})

	select {
	case g.historicalSyncReqs <- &historicalSyncReq{
		doneChan: done,
	}:
	case <-time.After(syncTransitionTimeout):
		return ErrSyncTransitionTimeout
	case <-g.quit:
		return ErrGossiperShuttingDown
	}

	select {
	case <-done:
		return nil
	case <-g.quit:
		return ErrGossiperShuttingDown
	}
}
316857
func (d *txDependency) waitForAndRetrieveValidationResult(ns string) error {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	err, ok := d.validationResultMap[ns]
	if ok {
		return err
	}

	for !ok {
		d.cond.Wait()
		err, ok = d.validationResultMap[ns]
	}

	return err
}
306220
func (b *Breaker) Failure(event Event, err error) bool {
	b.mu.Lock()
	defer b.mu.Unlock()

	b.recent++
	b.last = time.Now().UTC()
	return false // never drop events.
}
1070328
func (a *Allocator) WaitForInitialSync(ctx context.Context) error {
	select {
	case <-a.initialListDone:
	case <-ctx.Done():
		return fmt.Errorf("identity sync with kvstore was cancelled: %s", ctx.Err())
	}

	return nil
}
856386
func LastFailed(x int, attempts []LoginAttempt) bool {
	var failed int
	for i := len(attempts) - 1; i >= 0; i-- {
		if !attempts[i].Success {
			failed++
		} else {
			return false
		}
		if failed >= x {
			return true
		}
	}
	return false
}
1352328
func (cli *socketClient) FlushSync() error {
	reqRes := cli.queueRequest(types.ToRequestFlush())
	if err := cli.Error(); err != nil {
		return err
	}
	reqRes.Wait() // NOTE: if we don't flush the queue, its possible to get stuck here
	return cli.Error()
}
27026
func (c *Controller) GetFailureCount() int {
	c.mutex.RLock()
	defer c.mutex.RUnlock()

	return c.failureCount
}