// rewind allows bufferedBody to be read again.
322494
func (b *bufferedBody) rewind() error {
	if b == nil {
		return nil
	}
	_, err := b.Seek(0, io.SeekStart)
	return err
}
1100878
func RewindChain(chainConn ChainConn, txNotifier *TxNotifier,
	currBestBlock BlockEpoch, targetHeight int32) (BlockEpoch, error) {

	newBestBlock := BlockEpoch{
		Height: currBestBlock.Height,
		Hash:   currBestBlock.Hash,
	}

	for height := currBestBlock.Height; height > targetHeight; height-- {
		hash, err := chainConn.GetBlockHash(int64(height - 1))
		if err != nil {
			return newBestBlock, fmt.Errorf("unable to "+
				"find blockhash for disconnected height=%d: %v",
				height, err)
		}

		Log.Infof("Block disconnected from main chain: "+
			"height=%v, sha=%v", height, newBestBlock.Hash)

		err = txNotifier.DisconnectTip(uint32(height))
		if err != nil {
			return newBestBlock, fmt.Errorf("unable to "+
				" disconnect tip for height=%d: %v",
				height, err)
		}
		newBestBlock.Height = height - 1
		newBestBlock.Hash = hash
	}
	return newBestBlock, nil
}
214
func (dec *ProtoDecoder) Reset(r io.Reader) {
	dec.pBuf.Reset()
	dec.bBuf.Reset()
	dec.r = r
}
1315632
func (b *BufferedSeriesIterator) PeekBack() (t int64, v float64, ok bool) {
	return b.buf.last()
}
848445
func (b *ByteBuf) Resume(capacity int) {
	b.buf = b.pool.Alloc(b.capacity)
}
143393
func Retry(ctx context.Context, f func() error) error {
	// container for atomic error value
	type errWrap struct {
		E error
	}

	// Try the function in a goroutine
	var errVal atomic.Value
	doneCh := make(chan struct{})
	go func() {
		defer close(doneCh)

		delay := time.Duration(0)
		for {
			// If our context ended, we want to exit right away.
			select {
			case <-ctx.Done():
				return
			case <-time.After(delay):
			}

			// Try the function call
			err := f()

			// return if we have no error, or a FatalError
			done := false
			switch e := err.(type) {
			case nil:
				done = true
			case Fatal:
				err = e.FatalError()
				done = true
			}

			errVal.Store(errWrap{err})

			if done {
				return
			}

			log.Printf("[WARN] retryable error: %v", err)

			delay *= 2

			if delay == 0 {
				delay = initialBackoffDelay
			}

			if delay > maxBackoffDelay {
				delay = maxBackoffDelay
			}

			log.Printf("[INFO] sleeping for %s", delay)
		}
	}()

	// Wait for completion
	select {
	case <-ctx.Done():
	case <-doneCh:
	}

	var lastErr error
	// Check if we got an error executing
	if ev, ok := errVal.Load().(errWrap); ok {
		lastErr = ev.E
	}

	// Check if we have a context error to check if we're interrupted or timeout
	switch ctx.Err() {
	case context.Canceled:
		return fmt.Errorf("interrupted - last error: %v", lastErr)
	case context.DeadlineExceeded:
		return fmt.Errorf("timeout - last error: %v", lastErr)
	}

	if lastErr != nil {
		return lastErr
	}
	return nil
}
451580
func (b *batch) Replay(w ethdb.Writer) error {
	return b.b.Replay(&replayer{writer: w})
}
1308483
func AcceptString(l Lexer, word string, rewind bool) (ok bool) {
	i := 0
	defer func() {
		if rewind {
			Trace("Rewinding AccepString(%q) (%d runes)\n", word, i)
			for j := i; j > 0; j-- {
				l.Backup()
			}
		}
		Trace("AcceptString returning %s\n", ok)
	}()

	for pos := 0; pos < len(word); {
		r, width := utf8.DecodeRuneInString(word[pos:])
		pos += width
		var n rune
		if pos == 0 {
			n = l.Current()
		} else {
			n = l.Next()
		}
		i++
		Trace("r (%q) == n (%q) %s ? \n", r, n, r == n)
		if r != n {
			rewind = true
			ok = false
			return
		}
	}
	ok = true
	return
}
1333252
func (b *BlockedEvals) Reblock(eval *structs.Evaluation, token string) {
	b.processBlock(eval, token)
}
784126
func (ix *Reindexer) Scroll(timeout string) *Reindexer {
	ix.scroll = timeout
	return ix
}
989695
func (i *inotify) Rewatch(path string, _, newevent Event) error {
	return i.watch(path, newevent)
}
307240
func (p *cancelReadCloser) Read(buf []byte) (n int, err error) {
	return p.pR.Read(buf)
}
1168339
func (b *Batch) Replay(r BatchReplay) error {
	for _, index := range b.index {
		switch index.keyType {
		case keyTypeVal:
			r.Put(index.k(b.data), index.v(b.data))
		case keyTypeDel:
			r.Delete(index.k(b.data))
		}
	}
	return nil
}
923964
func (s *Scanner) undo(scanCode int) {
	if s.redo {
		panic("json: invalid use of scanner")
	}
	s.redoCode = scanCode
	s.redoState = s.Step
	s.Step = stateRedo
	s.redo = true
}
513790
func (b *Buffer) Read(p []byte) (n int, err error) {
	if b.off >= len(b.buf) {
		// Buffer is empty, reset to recover space.
		b.Truncate(0)
		if len(p) == 0 {
			return
		}
		return 0, io.EOF
	}
	n = copy(p, b.buf[b.off:])
	b.off += n
	return
}
716777
func Retry(ctx context.Context, interval, timeout time.Duration, fn func() error) (err error) {
	j := &backoff.Backoff{Min: interval / 2, Max: interval, Factor: 1, Jitter: true}
	to := time.NewTimer(timeout)
	defer to.Stop()

	for {
		if err = fn(); err == nil {
			return nil
		}

		Logger.Error("operation failed to execute", zap.Error(err))

		// @check if the context has been cancelled or wait
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-to.C:
			return errors.New("operation timed out")
		case <-time.After(j.Duration()):
		}
	}
}
1306749
func (t *Tailer) Read(resumeMap map[string]client.MessageSet, filterFn client.NsFilterFunc) client.MessageChanFunc {
	return func(s client.Session, done chan struct{}) (chan client.MessageSet, error) {
		readFunc := t.reader.Read(resumeMap, filterFn)
		msgChan, err := readFunc(s, done)
		if err != nil {
			return nil, err
		}
		session := s.(*Session)
		out := make(chan client.MessageSet)
		go func() {
			defer close(out)
			// read until reader done
			for msg := range msgChan {
				out <- msg
			}
			// start tailing
			log.With("db", session.db).With("logical_decoding_slot", t.replicationSlot).Infoln("Listening for changes...")
			for {
				select {
				case <-done:
					log.With("db", session.db).Infoln("tailing stopping...")
					return
				case <-time.After(time.Second):
					msgSlice, err := t.pluckFromLogicalDecoding(s.(*Session), filterFn)
					if err != nil {
						log.With("db", session.db).Errorf("error plucking from logical decoding %v", err)
						continue
					}
					for _, msg := range msgSlice {
						out <- msg
					}
				}
			}
		}()

		return out, nil
	}
}