// Replay only those messages since the last block.  `timeoutRoutine` should
// run concurrently to read off tickChan.
15129
func (cs *ConsensusState) catchupReplay(csHeight int64) error {

	// Set replayMode to true so we don't log signing errors.
	cs.replayMode = true
	defer func() { cs.replayMode = false }()

	// Ensure that #ENDHEIGHT for this height doesn't exist.
	// NOTE: This is just a sanity check. As far as we know things work fine
	// without it, and Handshake could reuse ConsensusState if it weren't for
	// this check (since we can crash after writing #ENDHEIGHT).
	//
	// Ignore data corruption errors since this is a sanity check.
	gr, found, err := cs.wal.SearchForEndHeight(csHeight, &WALSearchOptions{IgnoreDataCorruptionErrors: true})
	if err != nil {
		return err
	}
	if gr != nil {
		if err := gr.Close(); err != nil {
			return err
		}
	}
	if found {
		return fmt.Errorf("WAL should not contain #ENDHEIGHT %d", csHeight)
	}

	// Search for last height marker.
	//
	// Ignore data corruption errors in previous heights because we only care about last height
	gr, found, err = cs.wal.SearchForEndHeight(csHeight-1, &WALSearchOptions{IgnoreDataCorruptionErrors: true})
	if err == io.EOF {
		cs.Logger.Error("Replay: wal.group.Search returned EOF", "#ENDHEIGHT", csHeight-1)
	} else if err != nil {
		return err
	}
	if !found {
		return fmt.Errorf("Cannot replay height %d. WAL does not contain #ENDHEIGHT for %d", csHeight, csHeight-1)
	}
	defer gr.Close() // nolint: errcheck

	cs.Logger.Info("Catchup by replaying consensus messages", "height", csHeight)

	var msg *TimedWALMessage
	dec := WALDecoder{gr}

	for {
		msg, err = dec.Decode()
		if err == io.EOF {
			break
		} else if IsDataCorruptionError(err) {
			cs.Logger.Error("data has been corrupted in last height of consensus WAL", "err", err, "height", csHeight)
			return err
		} else if err != nil {
			return err
		}

		// NOTE: since the priv key is set when the msgs are received
		// it will attempt to eg double sign but we can just ignore it
		// since the votes will be replayed and we'll get to the next step
		if err := cs.readReplayMessage(msg, nil); err != nil {
			return err
		}
	}
	cs.Logger.Info("Replay: Done")
	return nil
}
1285069
func (s *BulkService) Timeout(timeout string) *BulkService {
	s.timeout = timeout
	return s
}
6036
func (dt *dynamicTimeout) Timeout() time.Duration {
	return time.Duration(atomic.LoadInt64(&dt.timeout))
}
1134161
func (s *OpenIndexService) MasterTimeout(masterTimeout string) *OpenIndexService {
	s.masterTimeout = masterTimeout
	return s
}
1102200
func (s *IngestPutPipelineService) Timeout(timeout string) *IngestPutPipelineService {
	s.timeout = timeout
	return s
}
964019
func (m *Message) Requeue(delay time.Duration) {
	m.doRequeue(delay, true)
}
1000244
func (m *Machine) OnTimerTick(tickNonce int64) error {
	// Silently skip unexpected, late or canceled ticks. This is fine.
	switch {
	case m.State.IsSuspended():
		return nil
	case m.State.LastTick.TickNonce != tickNonce:
		return nil
	}

	// Report error (to trigger a retry) if the tick happened unexpectedly soon.
	// Allow up to 50ms clock drift, but correct it to be 0. This is important for
	// getting the correct next tick time from an absolute schedule. If we pass
	// m.Now to the cron schedule uncorrected, we'll just get the time of the
	// already scheduled tick (the one we are handling now, since uncorrected
	// m.Now is before it).
	//
	// Note that m.Now is part of inputs and must not be mutated. We propagate
	// the corrected time via the call stack.
	now := m.Now
	switch delay := now.Sub(m.State.LastTick.When); {
	case delay < -50*time.Millisecond:
		return fmt.Errorf("tick happened %s before it was expected", -delay)
	case delay < 0:
		now = m.State.LastTick.When
	}

	// The scheduled time has come!
	m.State.Generation = m.nextGen()
	m.Actions = append(m.Actions, StartInvocationAction{
		Generation: m.State.Generation,
	})
	m.State.LastTick = TickLaterAction{}

	// Start waiting for a new tick right away if on an absolute schedule or just
	// keep the tick state clear for relative schedules: new tick will be set when
	// RewindIfNecessary() is manually called by whoever handles the cron.
	if m.Schedule.IsAbsolute() {
		m.rewindIfNecessary(now)
	}

	return nil
}
880735
func (r *SubReply) Timeout() bool {
	if r.Err == nil {
		return false
	}
	t, ok := r.Err.(*net.OpError)
	return ok && t.Timeout()
}
614789
func (s *ClusterRerouteService) Timeout(timeout string) *ClusterRerouteService {
	s.timeout = timeout
	return s
}
31965
func (cs *clientStream) retryLocked(lastErr error) error {
	for {
		cs.attempt.finish(lastErr)
		if err := cs.shouldRetry(lastErr); err != nil {
			cs.commitAttemptLocked()
			return err
		}
		if err := cs.newAttemptLocked(nil, nil); err != nil {
			return err
		}
		if lastErr = cs.replayBufferLocked(); lastErr == nil {
			return nil
		}
	}
}
216118
func retry(ctx context.Context, f func() bool, delay time.Duration) error {
	if f() {
		return nil
	}

	ticker := time.NewTicker(delay)
	errChan := make(chan error)

	go func() {
		defer close(errChan)

		for {
			select {
			case <-ctx.Done():
				errChan <- ctx.Err()
				return
			case <-ticker.C:
				if f() {
					return
				}
			}
		}
	}()

	return <-errChan
}
265756
func (c *Client) SetTimeout(max time.Duration) {
	atomic.StoreInt64((*int64)(&c.timeout), (int64)(max))
}
413920
func ReplayOplog(r io.Reader, controller ratecontroller.Controller, alwaysUpsert bool, host string) error {
	done := make(chan struct{})
	defer close(done)

	session, err := mgo.Dial(host)
	if err != nil {
		return err
	}
	defer session.Close()

	log.Println("Parsing BSON...")
	ops, parseErrors := parseBSON(done, r)
	timedOps := controlRate(done, ops, controller)
	batchedOps := batchOps(done, timedOps)

	applyOps := getApplyOpsFunc(session, alwaysUpsert)
	log.Println("Begin replaying...")

	if err := oplogReplay(batchedOps, applyOps); err != nil {
		return err
	}
	if err := <-parseErrors; err != nil {
		return err
	}
	return nil
}
591407
func (w *realTimeoutFactory) TimeoutCh() (<-chan time.Time, func() bool) {
	if w.timeout == 0 {
		return neverExitWatch, func() bool { return false }
	}
	t := time.NewTimer(w.timeout)
	return t.C, t.Stop
}
520083
func (s *PutScriptService) MasterTimeout(masterTimeout string) *PutScriptService {
	s.masterTimeout = masterTimeout
	return s
}
450925
func (s *Snapshotter) replay() error {
	// Seek to the beginning
	if _, err := s.fh.Seek(0, os.SEEK_SET); err != nil {
		return err
	}

	// Read each line
	reader := bufio.NewReader(s.fh)
	for {
		line, err := reader.ReadString('\n')
		if err != nil {
			break
		}

		// Skip the newline
		line = line[:len(line)-1]

		// Switch on the prefix
		if strings.HasPrefix(line, "alive: ") {
			info := strings.TrimPrefix(line, "alive: ")
			addrIdx := strings.LastIndex(info, " ")
			if addrIdx == -1 {
				s.logger.Printf("[WARN] serf: Failed to parse address: %v", line)
				continue
			}
			addr := info[addrIdx+1:]
			name := info[:addrIdx]
			s.aliveNodes[name] = addr

		} else if strings.HasPrefix(line, "not-alive: ") {
			name := strings.TrimPrefix(line, "not-alive: ")
			delete(s.aliveNodes, name)

		} else if strings.HasPrefix(line, "clock: ") {
			timeStr := strings.TrimPrefix(line, "clock: ")
			timeInt, err := strconv.ParseUint(timeStr, 10, 64)
			if err != nil {
				s.logger.Printf("[WARN] serf: Failed to convert clock time: %v", err)
				continue
			}
			s.lastClock = LamportTime(timeInt)

		} else if strings.HasPrefix(line, "event-clock: ") {
			timeStr := strings.TrimPrefix(line, "event-clock: ")
			timeInt, err := strconv.ParseUint(timeStr, 10, 64)
			if err != nil {
				s.logger.Printf("[WARN] serf: Failed to convert event clock time: %v", err)
				continue
			}
			s.lastEventClock = LamportTime(timeInt)

		} else if strings.HasPrefix(line, "query-clock: ") {
			timeStr := strings.TrimPrefix(line, "query-clock: ")
			timeInt, err := strconv.ParseUint(timeStr, 10, 64)
			if err != nil {
				s.logger.Printf("[WARN] serf: Failed to convert query clock time: %v", err)
				continue
			}
			s.lastQueryClock = LamportTime(timeInt)

		} else if strings.HasPrefix(line, "coordinate: ") {
			continue // Ignores any coordinate persistence from old snapshots, serf should re-converge
		} else if line == "leave" {
			// Ignore a leave if we plan on re-joining
			if s.rejoinAfterLeave {
				s.logger.Printf("[INFO] serf: Ignoring previous leave in snapshot")
				continue
			}
			s.aliveNodes = make(map[string]string)
			s.lastClock = 0
			s.lastEventClock = 0
			s.lastQueryClock = 0

		} else if strings.HasPrefix(line, "#") {
			// Skip comment lines

		} else {
			s.logger.Printf("[WARN] serf: Unrecognized snapshot line: %v", line)
		}
	}

	// Seek to the end
	if _, err := s.fh.Seek(0, os.SEEK_END); err != nil {
		return err
	}
	return nil
}
288295
func (c PubSubConn) ReceiveWithTimeout(timeout time.Duration) interface{} {
	return c.receiveInternal(ReceiveWithTimeout(c.Conn, timeout))
}