lotus/chain/store/messages.go

337 lines
8.5 KiB
Go
Raw Normal View History

package store
import (
"context"
2022-06-14 15:00:51 +00:00
"github.com/ipfs/go-cid"
cbor "github.com/ipfs/go-ipld-cbor"
ipld "github.com/ipfs/go-ipld-format"
block "github.com/ipfs/go-libipfs/blocks"
cbg "github.com/whyrusleeping/cbor-gen"
2022-06-14 15:00:51 +00:00
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
blockadt "github.com/filecoin-project/specs-actors/actors/util/adt"
bstore "github.com/filecoin-project/lotus/blockstore"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/types"
)
type storable interface {
ToStorageBlock() (block.Block, error)
}
2021-12-17 09:42:09 +00:00
func PutMessage(ctx context.Context, bs bstore.Blockstore, m storable) (cid.Cid, error) {
b, err := m.ToStorageBlock()
if err != nil {
return cid.Undef, err
}
2021-12-17 09:42:09 +00:00
if err := bs.Put(ctx, b); err != nil {
return cid.Undef, err
}
return b.Cid(), nil
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) PutMessage(ctx context.Context, m storable) (cid.Cid, error) {
return PutMessage(ctx, cs.chainBlockstore, m)
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) GetCMessage(ctx context.Context, c cid.Cid) (types.ChainMsg, error) {
m, err := cs.GetMessage(ctx, c)
if err == nil {
return m, nil
}
if !ipld.IsNotFound(err) {
log.Warnf("GetCMessage: unexpected error getting unsigned message: %s", err)
}
2021-12-17 09:42:09 +00:00
return cs.GetSignedMessage(ctx, c)
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) GetMessage(ctx context.Context, c cid.Cid) (*types.Message, error) {
var msg *types.Message
2021-12-17 09:42:09 +00:00
err := cs.chainLocalBlockstore.View(ctx, c, func(b []byte) (err error) {
msg, err = types.DecodeMessage(b)
return err
})
return msg, err
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) GetSignedMessage(ctx context.Context, c cid.Cid) (*types.SignedMessage, error) {
var msg *types.SignedMessage
2021-12-17 09:42:09 +00:00
err := cs.chainLocalBlockstore.View(ctx, c, func(b []byte) (err error) {
msg, err = types.DecodeSignedMessage(b)
return err
})
return msg, err
}
func (cs *ChainStore) readAMTCids(root cid.Cid) ([]cid.Cid, error) {
ctx := context.TODO()
// block headers use adt0, for now.
a, err := blockadt.AsArray(cs.ActorStore(ctx), root)
if err != nil {
return nil, xerrors.Errorf("amt load: %w", err)
}
var (
cids []cid.Cid
cborCid cbg.CborCid
)
if err := a.ForEach(&cborCid, func(i int64) error {
c := cid.Cid(cborCid)
cids = append(cids, c)
return nil
}); err != nil {
return nil, xerrors.Errorf("failed to traverse amt: %w", err)
}
if uint64(len(cids)) != a.Length() {
return nil, xerrors.Errorf("found %d cids, expected %d", len(cids), a.Length())
}
return cids, nil
}
type BlockMessages struct {
Miner address.Address
BlsMessages []types.ChainMsg
SecpkMessages []types.ChainMsg
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) BlockMsgsForTipset(ctx context.Context, ts *types.TipSet) ([]BlockMessages, error) {
2021-09-02 16:07:23 +00:00
// returned BlockMessages match block order in tipset
applied := make(map[address.Address]uint64)
cst := cbor.NewCborStore(cs.stateBlockstore)
st, err := state.LoadStateTree(cst, ts.Blocks()[0].ParentStateRoot)
if err != nil {
return nil, xerrors.Errorf("failed to load state tree at tipset %s: %w", ts, err)
}
selectMsg := func(m *types.Message) (bool, error) {
var sender address.Address
if ts.Height() >= build.UpgradeHyperdriveHeight {
sender, err = st.LookupID(m.From)
if err != nil {
return false, err
}
} else {
sender = m.From
}
// The first match for a sender is guaranteed to have correct nonce -- the block isn't valid otherwise
if _, ok := applied[sender]; !ok {
applied[sender] = m.Nonce
}
if applied[sender] != m.Nonce {
return false, nil
}
applied[sender]++
return true, nil
}
var out []BlockMessages
for _, b := range ts.Blocks() {
2021-12-17 09:42:09 +00:00
bms, sms, err := cs.MessagesForBlock(ctx, b)
if err != nil {
return nil, xerrors.Errorf("failed to get messages for block: %w", err)
}
bm := BlockMessages{
Miner: b.Miner,
BlsMessages: make([]types.ChainMsg, 0, len(bms)),
SecpkMessages: make([]types.ChainMsg, 0, len(sms)),
}
for _, bmsg := range bms {
b, err := selectMsg(bmsg.VMMessage())
if err != nil {
return nil, xerrors.Errorf("failed to decide whether to select message for block: %w", err)
}
if b {
bm.BlsMessages = append(bm.BlsMessages, bmsg)
}
}
for _, smsg := range sms {
b, err := selectMsg(smsg.VMMessage())
if err != nil {
return nil, xerrors.Errorf("failed to decide whether to select message for block: %w", err)
}
if b {
bm.SecpkMessages = append(bm.SecpkMessages, smsg)
}
}
out = append(out, bm)
}
return out, nil
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) MessagesForTipset(ctx context.Context, ts *types.TipSet) ([]types.ChainMsg, error) {
bmsgs, err := cs.BlockMsgsForTipset(ctx, ts)
if err != nil {
return nil, err
}
var out []types.ChainMsg
for _, bm := range bmsgs {
for _, blsm := range bm.BlsMessages {
out = append(out, blsm)
}
for _, secm := range bm.SecpkMessages {
out = append(out, secm)
}
}
return out, nil
}
type mmCids struct {
bls []cid.Cid
secpk []cid.Cid
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) ReadMsgMetaCids(ctx context.Context, mmc cid.Cid) ([]cid.Cid, []cid.Cid, error) {
o, ok := cs.mmCache.Get(mmc)
if ok {
mmcids := o.(*mmCids)
return mmcids.bls, mmcids.secpk, nil
}
cst := cbor.NewCborStore(cs.chainLocalBlockstore)
var msgmeta types.MsgMeta
2021-12-17 09:42:09 +00:00
if err := cst.Get(ctx, mmc, &msgmeta); err != nil {
return nil, nil, xerrors.Errorf("failed to load msgmeta (%s): %w", mmc, err)
}
blscids, err := cs.readAMTCids(msgmeta.BlsMessages)
if err != nil {
return nil, nil, xerrors.Errorf("loading bls message cids for block: %w", err)
}
secpkcids, err := cs.readAMTCids(msgmeta.SecpkMessages)
if err != nil {
return nil, nil, xerrors.Errorf("loading secpk message cids for block: %w", err)
}
cs.mmCache.Add(mmc, &mmCids{
bls: blscids,
secpk: secpkcids,
})
return blscids, secpkcids, nil
}
func (cs *ChainStore) ReadReceipts(ctx context.Context, root cid.Cid) ([]types.MessageReceipt, error) {
a, err := blockadt.AsArray(cs.ActorStore(ctx), root)
if err != nil {
return nil, err
}
receipts := make([]types.MessageReceipt, 0, a.Length())
var rcpt types.MessageReceipt
if err := a.ForEach(&rcpt, func(i int64) error {
if int64(len(receipts)) != i {
return xerrors.Errorf("missing receipt %d", i)
}
receipts = append(receipts, rcpt)
return nil
}); err != nil {
return nil, err
}
return receipts, nil
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) MessagesForBlock(ctx context.Context, b *types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error) {
blscids, secpkcids, err := cs.ReadMsgMetaCids(ctx, b.Messages)
if err != nil {
return nil, nil, err
}
2021-12-17 09:42:09 +00:00
blsmsgs, err := cs.LoadMessagesFromCids(ctx, blscids)
if err != nil {
return nil, nil, xerrors.Errorf("loading bls messages for block: %w", err)
}
2021-12-17 09:42:09 +00:00
secpkmsgs, err := cs.LoadSignedMessagesFromCids(ctx, secpkcids)
if err != nil {
return nil, nil, xerrors.Errorf("loading secpk messages for block: %w", err)
}
return blsmsgs, secpkmsgs, nil
}
func (cs *ChainStore) SecpkMessagesForBlock(ctx context.Context, b *types.BlockHeader) ([]*types.SignedMessage, error) {
_, secpkcids, err := cs.ReadMsgMetaCids(ctx, b.Messages)
if err != nil {
return nil, err
}
secpkmsgs, err := cs.LoadSignedMessagesFromCids(ctx, secpkcids)
if err != nil {
return nil, xerrors.Errorf("loading secpk messages for block: %w", err)
}
return secpkmsgs, nil
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) GetParentReceipt(ctx context.Context, b *types.BlockHeader, i int) (*types.MessageReceipt, error) {
// block headers use adt0, for now.
a, err := blockadt.AsArray(cs.ActorStore(ctx), b.ParentMessageReceipts)
if err != nil {
return nil, xerrors.Errorf("amt load: %w", err)
}
var r types.MessageReceipt
if found, err := a.Get(uint64(i), &r); err != nil {
return nil, err
} else if !found {
return nil, xerrors.Errorf("failed to find receipt %d", i)
}
return &r, nil
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) LoadMessagesFromCids(ctx context.Context, cids []cid.Cid) ([]*types.Message, error) {
msgs := make([]*types.Message, 0, len(cids))
for i, c := range cids {
2021-12-17 09:42:09 +00:00
m, err := cs.GetMessage(ctx, c)
if err != nil {
return nil, xerrors.Errorf("failed to get message: (%s):%d: %w", c, i, err)
}
msgs = append(msgs, m)
}
return msgs, nil
}
2021-12-17 09:42:09 +00:00
func (cs *ChainStore) LoadSignedMessagesFromCids(ctx context.Context, cids []cid.Cid) ([]*types.SignedMessage, error) {
msgs := make([]*types.SignedMessage, 0, len(cids))
for i, c := range cids {
2021-12-17 09:42:09 +00:00
m, err := cs.GetSignedMessage(ctx, c)
if err != nil {
return nil, xerrors.Errorf("failed to get message: (%s):%d: %w", c, i, err)
}
msgs = append(msgs, m)
}
return msgs, nil
}