lotus/chain/messagepool/provider.go

package messagepool

import (
	"context"
	"time"

	"github.com/ipfs/go-cid"
	pubsub "github.com/libp2p/go-libp2p-pubsub"
	"golang.org/x/xerrors"

	"github.com/filecoin-project/go-address"
	"github.com/filecoin-project/lotus/chain/stmgr"
	"github.com/filecoin-project/lotus/chain/store"
	"github.com/filecoin-project/lotus/chain/types"
)

var (
	HeadChangeCoalesceMinDelay      = 2 * time.Second
	HeadChangeCoalesceMaxDelay      = 6 * time.Second
	HeadChangeCoalesceMergeInterval = time.Second
)

type Provider interface {
	SubscribeHeadChanges(func(rev, app []*types.TipSet) error) *types.TipSet
	PutMessage(m types.ChainMsg) (cid.Cid, error)
	PubSubPublish(string, []byte) error
	GetActorAfter(address.Address, *types.TipSet) (*types.Actor, error)
	StateAccountKeyAtFinality(context.Context, address.Address, *types.TipSet) (address.Address, error)
	MessagesForBlock(*types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error)
	MessagesForTipset(*types.TipSet) ([]types.ChainMsg, error)
	LoadTipSet(tsk types.TipSetKey) (*types.TipSet, error)
	ChainComputeBaseFee(ctx context.Context, ts *types.TipSet) (types.BigInt, error)
}

type mpoolProvider struct {
	sm *stmgr.StateManager
	ps *pubsub.PubSub
}

var _ Provider = (*mpoolProvider)(nil)

func NewProvider(sm *stmgr.StateManager, ps *pubsub.PubSub) Provider {
	return &mpoolProvider{sm: sm, ps: ps}
}

func (mpp *mpoolProvider) SubscribeHeadChanges(cb func(rev, app []*types.TipSet) error) *types.TipSet {
	mpp.sm.ChainStore().SubscribeHeadChanges(
		store.WrapHeadChangeCoalescer(
			cb,
			HeadChangeCoalesceMinDelay,
			HeadChangeCoalesceMaxDelay,
			HeadChangeCoalesceMergeInterval,
		))
	return mpp.sm.ChainStore().GetHeaviestTipSet()
}

func (mpp *mpoolProvider) PutMessage(m types.ChainMsg) (cid.Cid, error) {
	return mpp.sm.ChainStore().PutMessage(m)
}

func (mpp *mpoolProvider) PubSubPublish(k string, v []byte) error {
	return mpp.ps.Publish(k, v) //nolint
}

func (mpp *mpoolProvider) GetActorAfter(addr address.Address, ts *types.TipSet) (*types.Actor, error) {
	stcid, _, err := mpp.sm.TipSetState(context.TODO(), ts)
	if err != nil {
		return nil, xerrors.Errorf("computing tipset state for GetActor: %w", err)
	}
	st, err := mpp.sm.StateTree(stcid)
	if err != nil {
		return nil, xerrors.Errorf("failed to load state tree: %w", err)
	}
	return st.GetActor(addr)
}

func (mpp *mpoolProvider) StateAccountKeyAtFinality(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
	return mpp.sm.ResolveToKeyAddressAtFinality(ctx, addr, ts)
}

func (mpp *mpoolProvider) MessagesForBlock(h *types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error) {
	return mpp.sm.ChainStore().MessagesForBlock(h)
}

func (mpp *mpoolProvider) MessagesForTipset(ts *types.TipSet) ([]types.ChainMsg, error) {
	return mpp.sm.ChainStore().MessagesForTipset(ts)
}

func (mpp *mpoolProvider) LoadTipSet(tsk types.TipSetKey) (*types.TipSet, error) {
	return mpp.sm.ChainStore().LoadTipSet(tsk)
}

func (mpp *mpoolProvider) ChainComputeBaseFee(ctx context.Context, ts *types.TipSet) (types.BigInt, error) {
	baseFee, err := mpp.sm.ChainStore().ComputeBaseFee(ctx, ts)
	if err != nil {
		return types.NewInt(0), xerrors.Errorf("computing base fee at %s: %w", ts, err)
	}
	return baseFee, nil
}