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 }