lotus/tools/stats/sync/sync.go

package sync

import (
	"context"
	"time"

	"github.com/filecoin-project/go-state-types/abi"

	"github.com/filecoin-project/lotus/api"
	"github.com/filecoin-project/lotus/build"
	"github.com/filecoin-project/lotus/chain/store"
	"github.com/filecoin-project/lotus/chain/types"
	"github.com/filecoin-project/lotus/tools/stats/headbuffer"
)

type SyncWaitApi interface {
	SyncState(context.Context) (*api.SyncState, error)
	ChainHead(context.Context) (*types.TipSet, error)
}

// SyncWait returns when ChainHead is within 20 epochs of the expected height
func SyncWait(ctx context.Context, napi SyncWaitApi) error {
	for {
		state, err := napi.SyncState(ctx)
		if err != nil {
			return err
		}

		if len(state.ActiveSyncs) == 0 {
			build.Clock.Sleep(time.Second)
			continue
		}

		head, err := napi.ChainHead(ctx)
		if err != nil {
			return err
		}

		working := -1
		for i, ss := range state.ActiveSyncs {
			switch ss.Stage {
			case api.StageSyncComplete:
			default:
				working = i
			case api.StageIdle:
				// not complete, not actively working
			}
		}

		if working == -1 {
			working = len(state.ActiveSyncs) - 1
		}

		ss := state.ActiveSyncs[working]

		if ss.Base == nil || ss.Target == nil {
			log.Infow(
				"syncing",
				"height", ss.Height,
				"stage", ss.Stage.String(),
			)
		} else {
			log.Infow(
				"syncing",
				"base", ss.Base.Key(),
				"target", ss.Target.Key(),
				"target_height", ss.Target.Height(),
				"height", ss.Height,
				"stage", ss.Stage.String(),
			)
		}

		if build.Clock.Now().Unix()-int64(head.MinTimestamp()) < int64(build.BlockDelaySecs)*30 {
			break
		}

		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-build.Clock.After(time.Duration(int64(build.BlockDelaySecs) * int64(time.Second))):
		}
	}

	return nil
}

type BufferedTipsetChannelApi interface {
	ChainNotify(context.Context) (<-chan []*api.HeadChange, error)
	Version(context.Context) (api.APIVersion, error)
	ChainGetTipSet(context.Context, types.TipSetKey) (*types.TipSet, error)
}

// BufferedTipsetChannel returns an unbuffered channel of tipsets. Buffering occurs internally to handle revert
// ChainNotify changes. The returned channel can output tipsets at the same height twice if a reorg larger the the
// provided `size` occurs.
func BufferedTipsetChannel(ctx context.Context, api BufferedTipsetChannelApi, lastHeight abi.ChainEpoch, size int) (<-chan *types.TipSet, error) {
	chmain := make(chan *types.TipSet)

	hb := headbuffer.NewHeadChangeStackBuffer(size)

	notif, err := api.ChainNotify(ctx)
	if err != nil {
		return nil, err
	}

	go func() {
		defer close(chmain)

		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case changes, ok := <-notif:
				if !ok {
					return
				}
				for _, change := range changes {
					log.Debugw("head event", "height", change.Val.Height(), "type", change.Type)

					switch change.Type {
					case store.HCCurrent:
						tipsets, err := loadTipsets(ctx, api, change.Val, lastHeight)
						if err != nil {
							log.Info(err)
							return
						}

						for _, tipset := range tipsets {
							chmain <- tipset
						}
					case store.HCApply:
						if out := hb.Push(change); out != nil {
							chmain <- out.Val
						}
					case store.HCRevert:
						hb.Pop()
					}
				}
			case <-ticker.C:
				log.Debug("running health check")

				cctx, cancel := context.WithTimeout(ctx, 5*time.Second)

				if _, err := api.Version(cctx); err != nil {
					log.Error("health check failed")
					cancel()
					return
				}

				cancel()

				log.Debug("node online")
			case <-ctx.Done():
				return
			}
		}
	}()

	return chmain, nil
}

func loadTipsets(ctx context.Context, api BufferedTipsetChannelApi, curr *types.TipSet, lowestHeight abi.ChainEpoch) ([]*types.TipSet, error) {
	log.Infow("loading tipsets", "to_height", lowestHeight, "from_height", curr.Height())
	tipsets := []*types.TipSet{}
	for {
		if curr.Height() == 0 {
			break
		}

		if curr.Height() <= lowestHeight {
			break
		}

		log.Debugw("walking back", "height", curr.Height())
		tipsets = append(tipsets, curr)

		tsk := curr.Parents()
		prev, err := api.ChainGetTipSet(ctx, tsk)
		if err != nil {
			return tipsets, err
		}

		curr = prev
	}

	for i, j := 0, len(tipsets)-1; i < j; i, j = i+1, j-1 {
		tipsets[i], tipsets[j] = tipsets[j], tipsets[i]
	}

	return tipsets, nil
}