lotus/chain/sync.go

1239 lines
35 KiB
Go
Raw Normal View History

2019-07-05 14:29:17 +00:00
package chain
import (
"context"
"errors"
2019-07-05 14:29:17 +00:00
"fmt"
"sync"
"time"
2019-07-05 14:29:17 +00:00
2020-02-27 21:45:31 +00:00
"github.com/filecoin-project/specs-actors/actors/abi"
"github.com/filecoin-project/specs-actors/actors/builtin"
"github.com/Gurpartap/async"
amt "github.com/filecoin-project/go-amt-ipld/v2"
2020-01-07 16:23:12 +00:00
sectorbuilder "github.com/filecoin-project/go-sectorbuilder"
2020-02-12 23:52:36 +00:00
"github.com/filecoin-project/specs-actors/actors/builtin/power"
"github.com/filecoin-project/specs-actors/actors/crypto"
"github.com/filecoin-project/specs-actors/actors/util/adt"
2019-11-11 19:32:30 +00:00
"github.com/hashicorp/go-multierror"
2019-07-05 14:29:17 +00:00
"github.com/ipfs/go-cid"
dstore "github.com/ipfs/go-datastore"
bstore "github.com/ipfs/go-ipfs-blockstore"
2020-02-04 22:19:05 +00:00
cbor "github.com/ipfs/go-ipld-cbor"
logging "github.com/ipfs/go-log/v2"
"github.com/libp2p/go-libp2p-core/connmgr"
2019-07-30 16:39:07 +00:00
"github.com/libp2p/go-libp2p-core/peer"
2019-09-17 18:09:22 +00:00
cbg "github.com/whyrusleeping/cbor-gen"
2019-11-18 21:39:07 +00:00
"github.com/whyrusleeping/pubsub"
2019-10-11 02:17:24 +00:00
"go.opencensus.io/trace"
2019-09-17 18:09:22 +00:00
"golang.org/x/xerrors"
2019-11-11 19:32:30 +00:00
2020-02-12 23:52:36 +00:00
bls "github.com/filecoin-project/filecoin-ffi"
"github.com/filecoin-project/go-address"
2020-01-07 16:23:12 +00:00
2019-11-11 19:32:30 +00:00
"github.com/filecoin-project/lotus/api"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/chain/blocksync"
"github.com/filecoin-project/lotus/chain/gen"
2019-11-11 19:32:30 +00:00
"github.com/filecoin-project/lotus/chain/state"
"github.com/filecoin-project/lotus/chain/stmgr"
"github.com/filecoin-project/lotus/chain/store"
"github.com/filecoin-project/lotus/chain/types"
"github.com/filecoin-project/lotus/lib/sigs"
2019-07-05 14:29:17 +00:00
)
2019-07-26 04:54:22 +00:00
var log = logging.Logger("chain")
2019-12-01 23:11:43 +00:00
var LocalIncoming = "incoming"
2019-11-18 21:39:07 +00:00
2019-07-05 14:29:17 +00:00
type Syncer struct {
// The interface for accessing and putting tipsets into local storage
2019-07-26 04:54:22 +00:00
store *store.ChainStore
2019-07-05 14:29:17 +00:00
// the state manager handles making state queries
sm *stmgr.StateManager
2019-07-05 14:46:21 +00:00
// The known Genesis tipset
2019-07-26 04:54:22 +00:00
Genesis *types.TipSet
2019-07-05 14:29:17 +00:00
// TipSets known to be invalid
2019-10-09 08:50:57 +00:00
bad *BadBlockCache
2019-07-05 14:29:17 +00:00
// handle to the block sync service
2019-11-09 23:00:22 +00:00
Bsync *blocksync.BlockSync
2019-07-05 14:29:17 +00:00
2019-07-11 02:36:43 +00:00
self peer.ID
syncmgr *SyncManager
2019-11-18 21:39:07 +00:00
connmgr connmgr.ConnManager
2019-11-18 21:39:07 +00:00
incoming *pubsub.PubSub
receiptTracker *blockReceiptTracker
2019-07-05 14:29:17 +00:00
}
func NewSyncer(sm *stmgr.StateManager, bsync *blocksync.BlockSync, connmgr connmgr.ConnManager, self peer.ID) (*Syncer, error) {
gen, err := sm.ChainStore().GetGenesis()
2019-07-05 14:29:17 +00:00
if err != nil {
return nil, err
}
2019-07-26 04:54:22 +00:00
gent, err := types.NewTipSet([]*types.BlockHeader{gen})
2019-07-05 14:29:17 +00:00
if err != nil {
return nil, err
}
2019-11-15 21:35:29 +00:00
s := &Syncer{
bad: NewBadBlockCache(),
Genesis: gent,
Bsync: bsync,
store: sm.ChainStore(),
sm: sm,
self: self,
receiptTracker: newBlockReceiptTracker(),
connmgr: connmgr,
2019-11-18 21:39:07 +00:00
2019-11-20 19:44:38 +00:00
incoming: pubsub.New(50),
2019-11-15 21:35:29 +00:00
}
s.syncmgr = NewSyncManager(s.Sync)
return s, nil
2019-07-05 14:29:17 +00:00
}
2019-11-15 21:35:29 +00:00
func (syncer *Syncer) Start() {
syncer.syncmgr.Start()
}
func (syncer *Syncer) Stop() {
syncer.syncmgr.Stop()
2019-07-05 14:29:17 +00:00
}
// InformNewHead informs the syncer about a new potential tipset
// This should be called when connecting to new peers, and additionally
// when receiving new blocks from the network
func (syncer *Syncer) InformNewHead(from peer.ID, fts *store.FullTipSet) bool {
2019-10-10 11:13:26 +00:00
ctx := context.Background()
2019-07-05 14:29:17 +00:00
if fts == nil {
2019-11-16 23:41:14 +00:00
log.Errorf("got nil tipset in InformNewHead")
return false
2019-07-05 14:29:17 +00:00
}
for _, b := range fts.Blocks {
if reason, ok := syncer.bad.Has(b.Cid()); ok {
log.Warnf("InformNewHead called on block marked as bad: %s (reason: %s)", b.Cid(), reason)
return false
}
2019-10-15 12:19:10 +00:00
if err := syncer.ValidateMsgMeta(b); err != nil {
log.Warnf("invalid block received: %s", err)
return false
}
}
2019-12-01 23:11:43 +00:00
syncer.incoming.Pub(fts.TipSet().Blocks(), LocalIncoming)
2019-11-18 21:39:07 +00:00
2019-07-11 02:36:43 +00:00
if from == syncer.self {
// TODO: this is kindof a hack...
2019-11-07 00:18:06 +00:00
log.Debug("got block from ourselves")
2019-07-11 02:36:43 +00:00
2019-10-10 11:13:26 +00:00
if err := syncer.Sync(ctx, fts.TipSet()); err != nil {
log.Errorf("failed to sync our own block %s: %+v", fts.TipSet().Cids(), err)
return false
2019-07-11 02:36:43 +00:00
}
return true
2019-07-11 02:36:43 +00:00
}
2019-11-15 03:19:16 +00:00
// TODO: IMPORTANT(GARBAGE) this needs to be put in the 'temporary' side of
// the blockstore
if err := syncer.store.PersistBlockHeaders(fts.TipSet().Blocks()...); err != nil {
log.Warn("failed to persist incoming block header: ", err)
return false
2019-11-15 03:19:16 +00:00
}
2019-07-08 14:07:09 +00:00
syncer.Bsync.AddPeer(from)
2019-07-05 14:29:17 +00:00
bestPweight := syncer.store.GetHeaviestTipSet().Blocks()[0].ParentWeight
2019-11-10 23:06:06 +00:00
targetWeight := fts.TipSet().Blocks()[0].ParentWeight
if targetWeight.LessThan(bestPweight) {
2019-12-04 04:59:41 +00:00
var miners []string
for _, blk := range fts.TipSet().Blocks() {
miners = append(miners, blk.Miner.String())
}
2019-12-16 17:14:21 +00:00
log.Infof("incoming tipset from %s does not appear to be better than our best chain, ignoring for now", miners)
return false
}
2019-11-16 06:48:42 +00:00
syncer.syncmgr.SetPeerHead(ctx, from, fts.TipSet())
return true
2019-07-05 14:29:17 +00:00
}
2019-11-18 21:39:07 +00:00
func (syncer *Syncer) IncomingBlocks(ctx context.Context) (<-chan *types.BlockHeader, error) {
2019-12-01 23:11:43 +00:00
sub := syncer.incoming.Sub(LocalIncoming)
2019-11-18 21:39:07 +00:00
out := make(chan *types.BlockHeader, 10)
go func() {
2019-12-01 23:11:43 +00:00
defer syncer.incoming.Unsub(sub, LocalIncoming)
2019-11-19 19:49:11 +00:00
2019-11-18 21:39:07 +00:00
for {
select {
case r := <-sub:
hs := r.([]*types.BlockHeader)
for _, h := range hs {
select {
case out <- h:
case <-ctx.Done():
return
}
}
case <-ctx.Done():
return
}
}
}()
return out, nil
}
2019-10-15 12:19:10 +00:00
func (syncer *Syncer) ValidateMsgMeta(fblk *types.FullBlock) error {
if msgc := len(fblk.BlsMessages) + len(fblk.SecpkMessages); msgc > build.BlockMessageLimit {
return xerrors.Errorf("block %s has too many messages (%d)", fblk.Header.Cid(), msgc)
}
var bcids, scids []cbg.CBORMarshaler
for _, m := range fblk.BlsMessages {
c := cbg.CborCid(m.Cid())
bcids = append(bcids, &c)
}
for _, m := range fblk.SecpkMessages {
c := cbg.CborCid(m.Cid())
scids = append(scids, &c)
}
2019-11-15 03:19:16 +00:00
// TODO: IMPORTANT(GARBAGE). These message puts and the msgmeta
// computation need to go into the 'temporary' side of the blockstore when
// we implement that
blockstore := syncer.store.Blockstore()
bs := cbor.NewCborStore(blockstore)
smroot, err := computeMsgMeta(bs, bcids, scids)
if err != nil {
return xerrors.Errorf("validating msgmeta, compute failed: %w", err)
}
if fblk.Header.Messages != smroot {
return xerrors.Errorf("messages in full block did not match msgmeta root in header (%s != %s)", fblk.Header.Messages, smroot)
}
2019-11-15 03:19:16 +00:00
for _, m := range fblk.BlsMessages {
_, err := store.PutMessage(blockstore, m)
if err != nil {
return xerrors.Errorf("putting bls message to blockstore after msgmeta computation: %w", err)
}
}
for _, m := range fblk.SecpkMessages {
_, err := store.PutMessage(blockstore, m)
if err != nil {
return xerrors.Errorf("putting bls message to blockstore after msgmeta computation: %w", err)
}
}
return nil
}
func (syncer *Syncer) LocalPeer() peer.ID {
return syncer.self
}
func (syncer *Syncer) ChainStore() *store.ChainStore {
return syncer.store
}
func (syncer *Syncer) InformNewBlock(from peer.ID, blk *types.FullBlock) bool {
2019-07-05 14:29:17 +00:00
// TODO: search for other blocks that could form a tipset with this block
// and then send that tipset to InformNewHead
2019-07-26 04:54:22 +00:00
fts := &store.FullTipSet{Blocks: []*types.FullBlock{blk}}
return syncer.InformNewHead(from, fts)
2019-07-05 14:29:17 +00:00
}
func copyBlockstore(from, to bstore.Blockstore) error {
cids, err := from.AllKeysChan(context.TODO())
if err != nil {
return err
}
for c := range cids {
b, err := from.Get(c)
if err != nil {
return err
}
if err := to.Put(b); err != nil {
return err
}
}
return nil
}
// TODO: this function effectively accepts unchecked input from the network,
// either validate it here, or ensure that its validated elsewhere (maybe make
// sure the blocksync code checks it?)
// maybe this code should actually live in blocksync??
func zipTipSetAndMessages(bs cbor.IpldStore, ts *types.TipSet, allbmsgs []*types.Message, allsmsgs []*types.SignedMessage, bmi, smi [][]uint64) (*store.FullTipSet, error) {
if len(ts.Blocks()) != len(smi) || len(ts.Blocks()) != len(bmi) {
2019-07-05 14:29:17 +00:00
return nil, fmt.Errorf("msgincl length didnt match tipset size")
}
2019-07-26 04:54:22 +00:00
fts := &store.FullTipSet{}
2019-07-05 14:29:17 +00:00
for bi, b := range ts.Blocks() {
var smsgs []*types.SignedMessage
2019-09-17 01:56:37 +00:00
var smsgCids []cbg.CBORMarshaler
for _, m := range smi[bi] {
smsgs = append(smsgs, allsmsgs[m])
2019-09-17 01:56:37 +00:00
c := cbg.CborCid(allsmsgs[m].Cid())
smsgCids = append(smsgCids, &c)
2019-07-05 14:29:17 +00:00
}
var bmsgs []*types.Message
2019-09-17 01:56:37 +00:00
var bmsgCids []cbg.CBORMarshaler
for _, m := range bmi[bi] {
bmsgs = append(bmsgs, allbmsgs[m])
2019-09-17 01:56:37 +00:00
c := cbg.CborCid(allbmsgs[m].Cid())
bmsgCids = append(bmsgCids, &c)
}
if msgc := len(bmsgCids) + len(smsgCids); msgc > build.BlockMessageLimit {
return nil, fmt.Errorf("block %q has too many messages (%d)", b.Cid(), msgc)
}
mrcid, err := computeMsgMeta(bs, bmsgCids, smsgCids)
if err != nil {
return nil, err
}
if b.Messages != mrcid {
2019-07-05 14:29:17 +00:00
return nil, fmt.Errorf("messages didnt match message root in header")
}
fb := &types.FullBlock{
Header: b,
BlsMessages: bmsgs,
SecpkMessages: smsgs,
2019-07-05 14:29:17 +00:00
}
fts.Blocks = append(fts.Blocks, fb)
}
return fts, nil
}
func computeMsgMeta(bs cbor.IpldStore, bmsgCids, smsgCids []cbg.CBORMarshaler) (cid.Cid, error) {
ctx := context.TODO()
bmroot, err := amt.FromArray(ctx, bs, bmsgCids)
if err != nil {
return cid.Undef, err
}
smroot, err := amt.FromArray(ctx, bs, smsgCids)
if err != nil {
return cid.Undef, err
}
mrcid, err := bs.Put(ctx, &types.MsgMeta{
BlsMessages: bmroot,
SecpkMessages: smroot,
})
if err != nil {
return cid.Undef, xerrors.Errorf("failed to put msgmeta: %w", err)
}
return mrcid, nil
}
2019-12-16 19:22:56 +00:00
func (syncer *Syncer) FetchTipSet(ctx context.Context, p peer.ID, tsk types.TipSetKey) (*store.FullTipSet, error) {
if fts, err := syncer.tryLoadFullTipSet(tsk); err == nil {
2019-07-05 14:29:17 +00:00
return fts, nil
}
2019-12-16 19:22:56 +00:00
return syncer.Bsync.GetFullTipSet(ctx, p, tsk)
2019-07-05 14:29:17 +00:00
}
2019-12-16 19:22:56 +00:00
func (syncer *Syncer) tryLoadFullTipSet(tsk types.TipSetKey) (*store.FullTipSet, error) {
ts, err := syncer.store.LoadTipSet(tsk)
2019-07-05 14:29:17 +00:00
if err != nil {
return nil, err
}
2019-07-26 04:54:22 +00:00
fts := &store.FullTipSet{}
2019-07-05 14:29:17 +00:00
for _, b := range ts.Blocks() {
bmsgs, smsgs, err := syncer.store.MessagesForBlock(b)
2019-07-05 14:29:17 +00:00
if err != nil {
return nil, err
}
fb := &types.FullBlock{
Header: b,
BlsMessages: bmsgs,
SecpkMessages: smsgs,
2019-07-05 14:29:17 +00:00
}
fts.Blocks = append(fts.Blocks, fb)
}
return fts, nil
}
2019-10-10 11:13:26 +00:00
func (syncer *Syncer) Sync(ctx context.Context, maybeHead *types.TipSet) error {
ctx, span := trace.StartSpan(ctx, "chain.Sync")
defer span.End()
2019-11-16 01:05:16 +00:00
2019-11-10 23:06:06 +00:00
if span.IsRecordingEvents() {
span.AddAttributes(
trace.StringAttribute("tipset", fmt.Sprint(maybeHead.Cids())),
trace.Int64Attribute("height", int64(maybeHead.Height())),
)
}
2019-11-20 19:44:38 +00:00
if syncer.store.GetHeaviestTipSet().ParentWeight().GreaterThan(maybeHead.ParentWeight()) {
return nil
}
if syncer.Genesis.Equals(maybeHead) || syncer.store.GetHeaviestTipSet().Equals(maybeHead) {
2019-07-05 14:29:17 +00:00
return nil
}
if err := syncer.collectChain(ctx, maybeHead); err != nil {
2019-11-10 23:06:06 +00:00
span.AddAttributes(trace.StringAttribute("col_error", err.Error()))
2019-12-04 23:20:02 +00:00
span.SetStatus(trace.Status{
Code: 13,
Message: err.Error(),
})
return xerrors.Errorf("collectChain failed: %w", err)
2019-07-05 14:29:17 +00:00
}
2019-10-15 05:00:30 +00:00
if err := syncer.store.PutTipSet(ctx, maybeHead); err != nil {
2019-11-10 23:06:06 +00:00
span.AddAttributes(trace.StringAttribute("put_error", err.Error()))
2019-12-04 23:20:02 +00:00
span.SetStatus(trace.Status{
Code: 13,
Message: err.Error(),
})
return xerrors.Errorf("failed to put synced tipset to chainstore: %w", err)
2019-07-05 14:29:17 +00:00
}
peers := syncer.receiptTracker.GetPeers(maybeHead)
if len(peers) > 0 {
syncer.connmgr.TagPeer(peers[0], "new-block", 40)
for _, p := range peers[1:] {
syncer.connmgr.TagPeer(p, "new-block", 25)
}
}
2019-07-05 14:29:17 +00:00
return nil
}
func isPermanent(err error) bool {
return !errors.Is(err, ErrTemporal)
}
func (syncer *Syncer) ValidateTipSet(ctx context.Context, fts *store.FullTipSet) error {
ctx, span := trace.StartSpan(ctx, "validateTipSet")
defer span.End()
span.AddAttributes(trace.Int64Attribute("height", int64(fts.TipSet().Height())))
2019-07-05 14:29:17 +00:00
ts := fts.TipSet()
2019-07-05 14:46:21 +00:00
if ts.Equals(syncer.Genesis) {
2019-07-05 14:29:17 +00:00
return nil
}
for _, b := range fts.Blocks {
if err := syncer.ValidateBlock(ctx, b); err != nil {
if isPermanent(err) {
syncer.bad.Add(b.Cid(), err.Error())
}
return xerrors.Errorf("validating block %s: %w", b.Cid(), err)
2019-07-05 14:29:17 +00:00
}
2019-10-10 03:04:10 +00:00
if err := syncer.sm.ChainStore().AddToTipSetTracker(b.Header); err != nil {
return xerrors.Errorf("failed to add validated header to tipset tracker: %w", err)
}
2019-07-05 14:29:17 +00:00
}
return nil
}
func (syncer *Syncer) minerIsValid(ctx context.Context, maddr address.Address, baseTs *types.TipSet) error {
2020-02-12 23:52:36 +00:00
var spast power.State
2020-02-25 20:54:58 +00:00
_, err := syncer.sm.LoadActorState(ctx, builtin.StoragePowerActorAddr, &spast, baseTs)
if err != nil {
return err
}
2020-02-12 23:52:36 +00:00
var claim power.Claim
exist, err := adt.AsMap(syncer.store.Store(ctx), spast.Claims).Get(adt.AddrKey(maddr), &claim)
if err != nil {
2020-02-12 23:52:36 +00:00
return err
}
2020-02-12 23:52:36 +00:00
if !exist {
2019-11-14 16:14:52 +00:00
return xerrors.New("miner isn't valid")
}
return nil
}
var ErrTemporal = errors.New("temporal error")
// Should match up with 'Semantical Validation' in validation.md in the spec
func (syncer *Syncer) ValidateBlock(ctx context.Context, b *types.FullBlock) error {
ctx, span := trace.StartSpan(ctx, "validateBlock")
defer span.End()
if build.InsecurePoStValidation {
log.Warn("insecure test validation is enabled, if you see this outside of a test, it is a severe bug!")
}
2019-10-10 11:13:26 +00:00
2019-07-05 14:29:17 +00:00
h := b.Header
2019-10-02 20:03:27 +00:00
2019-12-16 19:22:56 +00:00
baseTs, err := syncer.store.LoadTipSet(types.NewTipSetKey(h.Parents...))
2019-10-02 20:03:27 +00:00
if err != nil {
return xerrors.Errorf("load parent tipset failed (%s): %w", h.Parents, err)
2019-10-02 20:03:27 +00:00
}
2019-11-11 19:26:14 +00:00
// fast checks first
2019-11-19 20:07:16 +00:00
if h.BlockSig == nil {
return xerrors.Errorf("block had nil signature")
}
2019-11-11 19:26:14 +00:00
if h.Timestamp > uint64(time.Now().Unix()+build.AllowableClockDrift) {
return xerrors.Errorf("block was from the future: %w", ErrTemporal)
2019-11-11 19:26:14 +00:00
}
2019-12-03 20:00:04 +00:00
if h.Timestamp > uint64(time.Now().Unix()) {
log.Warn("Got block from the future, but within threshold", h.Timestamp, time.Now().Unix())
}
2019-11-11 19:26:14 +00:00
2020-02-08 02:18:32 +00:00
if h.Timestamp < baseTs.MinTimestamp()+(build.BlockDelay*uint64(h.Height-baseTs.Height())) {
log.Warn("timestamp funtimes: ", h.Timestamp, baseTs.MinTimestamp(), h.Height, baseTs.Height())
2020-02-23 00:47:47 +00:00
diff := (baseTs.MinTimestamp() + (build.BlockDelay * uint64(h.Height-baseTs.Height()))) - h.Timestamp
2020-02-22 13:10:46 +00:00
return xerrors.Errorf("block was generated too soon (h.ts:%d < base.mints:%d + BLOCK_DELAY:%d * deltaH:%d; diff %d)", h.Timestamp, baseTs.MinTimestamp(), build.BlockDelay, h.Height-baseTs.Height(), diff)
2019-11-11 19:26:14 +00:00
}
winnerCheck := async.Err(func() error {
2020-03-02 01:09:38 +00:00
slashed, err := stmgr.GetMinerSlashed(ctx, syncer.sm, baseTs, h.Miner)
if err != nil {
return xerrors.Errorf("failed to check if block miner was slashed: %w", err)
}
2020-03-02 01:09:38 +00:00
if slashed {
return xerrors.Errorf("received block was from slashed or invalid miner")
}
mpow, tpow, err := stmgr.GetPower(ctx, syncer.sm, baseTs, h.Miner)
2019-11-11 19:26:14 +00:00
if err != nil {
return xerrors.Errorf("failed getting power: %w", err)
}
2019-11-21 22:21:45 +00:00
ssize, err := stmgr.GetMinerSectorSize(ctx, syncer.sm, baseTs, h.Miner)
if err != nil {
return xerrors.Errorf("failed to get sector size for block miner: %w", err)
}
2020-02-08 02:18:32 +00:00
snum := types.BigDiv(mpow, types.NewInt(uint64(ssize)))
2019-12-20 09:51:13 +00:00
if len(h.EPostProof.Candidates) == 0 {
return xerrors.Errorf("no candidates")
}
wins := make(map[uint64]bool)
2019-11-28 12:46:56 +00:00
for _, t := range h.EPostProof.Candidates {
if wins[t.ChallengeIndex] {
return xerrors.Errorf("block had duplicate epost candidates")
}
wins[t.ChallengeIndex] = true
if !types.IsTicketWinner(t.Partial, ssize, snum.Uint64(), tpow) {
2019-11-21 22:21:45 +00:00
return xerrors.Errorf("miner created a block but was not a winner")
}
2019-11-11 19:26:14 +00:00
}
return nil
})
msgsCheck := async.Err(func() error {
if err := syncer.checkBlockMessages(ctx, b, baseTs); err != nil {
return xerrors.Errorf("block had invalid messages: %w", err)
}
return nil
})
minerCheck := async.Err(func() error {
if err := syncer.minerIsValid(ctx, h.Miner, baseTs); err != nil {
return xerrors.Errorf("minerIsValid failed: %w", err)
}
return nil
})
// Stuff that needs stateroot / worker address
2019-10-10 11:13:26 +00:00
stateroot, precp, err := syncer.sm.TipSetState(ctx, baseTs)
2019-07-05 14:29:17 +00:00
if err != nil {
return xerrors.Errorf("get tipsetstate(%d, %s) failed: %w", h.Height, h.Parents, err)
}
2019-09-27 23:55:15 +00:00
if stateroot != h.ParentStateRoot {
msgs, err := syncer.store.MessagesForTipset(baseTs)
if err != nil {
log.Error("failed to load messages for tipset during tipset state mismatch error: ", err)
} else {
log.Warn("Messages for tipset with mismatching state:")
for i, m := range msgs {
mm := m.VMMessage()
log.Warnf("Message[%d]: from=%s to=%s method=%d params=%x", i, mm.From, mm.To, mm.Method, mm.Params)
}
}
2019-09-27 23:55:15 +00:00
return xerrors.Errorf("parent state root did not match computed state (%s != %s)", stateroot, h.ParentStateRoot)
}
2019-09-27 23:55:15 +00:00
if precp != h.ParentMessageReceipts {
return xerrors.Errorf("parent receipts root did not match computed value (%s != %s)", precp, h.ParentMessageReceipts)
}
2019-09-11 20:10:29 +00:00
waddr, err := stmgr.GetMinerWorkerRaw(ctx, syncer.sm, stateroot, h.Miner)
if err != nil {
return xerrors.Errorf("GetMinerWorkerRaw failed: %w", err)
}
2019-11-11 19:26:14 +00:00
blockSigCheck := async.Err(func() error {
if err := sigs.CheckBlockSignature(h, ctx, waddr); err != nil {
2019-11-11 19:26:14 +00:00
return xerrors.Errorf("check block signature failed: %w", err)
}
return nil
})
2019-11-11 19:26:14 +00:00
tktsCheck := async.Err(func() error {
2020-02-23 20:00:47 +00:00
vrfBase, err := syncer.sm.ChainStore().GetRandomness(ctx, baseTs.Cids(), crypto.DomainSeparationTag_TicketProduction, int64(baseTs.Height()), h.Miner.Bytes())
if err != nil {
return xerrors.Errorf("failed to get randomness for verifying election proof: %w", err)
}
2019-11-26 22:53:52 +00:00
2020-02-23 20:00:47 +00:00
err = gen.VerifyVRF(ctx, waddr, vrfBase, h.Ticket.VRFProof)
2019-11-26 22:53:52 +00:00
if err != nil {
2019-11-11 19:26:14 +00:00
return xerrors.Errorf("validating block tickets failed: %w", err)
}
return nil
})
2019-11-11 19:26:14 +00:00
eproofCheck := async.Err(func() error {
2019-11-21 22:21:45 +00:00
if err := syncer.VerifyElectionPoStProof(ctx, h, baseTs, waddr); err != nil {
return xerrors.Errorf("invalid election post: %w", err)
2019-11-11 19:26:14 +00:00
}
return nil
})
await := []async.ErrorFuture{
minerCheck,
tktsCheck,
blockSigCheck,
eproofCheck,
winnerCheck,
msgsCheck,
}
2019-11-11 19:30:49 +00:00
var merr error
for _, fut := range await {
if err := fut.AwaitContext(ctx); err != nil {
2019-11-14 16:14:52 +00:00
merr = multierror.Append(merr, err)
}
}
2019-11-11 19:30:49 +00:00
return merr
}
2019-11-21 22:21:45 +00:00
func (syncer *Syncer) VerifyElectionPoStProof(ctx context.Context, h *types.BlockHeader, baseTs *types.TipSet, waddr address.Address) error {
2020-02-23 20:00:47 +00:00
rand, err := syncer.sm.ChainStore().GetRandomness(ctx, baseTs.Cids(), crypto.DomainSeparationTag_ElectionPoStChallengeSeed, int64(h.Height-build.EcRandomnessLookback), h.Miner.Bytes())
2019-11-21 22:21:45 +00:00
if err != nil {
return xerrors.Errorf("failed to get randomness for verifying election proof: %w", err)
}
2020-02-23 20:00:47 +00:00
if err := VerifyElectionPoStVRF(ctx, h.EPostProof.PostRand, rand, waddr); err != nil {
2019-11-21 22:21:45 +00:00
return xerrors.Errorf("checking eproof failed: %w", err)
}
ssize, err := stmgr.GetMinerSectorSize(ctx, syncer.sm, baseTs, h.Miner)
if err != nil {
return xerrors.Errorf("failed to get sector size for miner: %w", err)
}
2020-02-27 21:45:31 +00:00
mid, err := address.IDFromAddress(h.Miner)
if err != nil {
return xerrors.Errorf("failed to get ID from miner address %s: %w", h.Miner, err)
}
var winners []abi.PoStCandidate
2019-11-28 12:46:56 +00:00
for _, t := range h.EPostProof.Candidates {
2020-02-27 21:45:31 +00:00
winners = append(winners, abi.PoStCandidate{
PartialTicket: t.Partial,
SectorID: abi.SectorID{
Number: t.SectorID,
Miner: abi.ActorID(mid),
},
ChallengeIndex: int64(t.ChallengeIndex),
2019-11-21 22:21:45 +00:00
})
}
2019-12-20 09:51:13 +00:00
if len(winners) == 0 {
return xerrors.Errorf("no candidates")
}
2019-11-21 22:21:45 +00:00
sectorInfo, err := stmgr.GetSectorsForElectionPost(ctx, syncer.sm, baseTs, h.Miner)
if err != nil {
return xerrors.Errorf("getting election post sector set: %w", err)
}
if build.InsecurePoStValidation {
2020-02-27 21:45:31 +00:00
if len(h.EPostProof.Proofs) == 0 {
return xerrors.Errorf("[TESTING] No election post proof given")
}
if string(h.EPostProof.Proofs[0].ProofBytes) == "valid proof" {
return nil
}
return xerrors.Errorf("[TESTING] election post was invalid")
}
2020-02-27 21:45:31 +00:00
rt, _, err := api.ProofTypeFromSectorSize(ssize)
if err != nil {
return err
}
candidates := make([]abi.PoStCandidate, 0, len(h.EPostProof.Candidates))
for _, c := range h.EPostProof.Candidates {
candidates = append(candidates, abi.PoStCandidate{
RegisteredProof: rt,
PartialTicket: c.Partial,
SectorID: abi.SectorID{Number: c.SectorID}, // this should not be an ID, we already know who the miner is...
ChallengeIndex: int64(c.ChallengeIndex),
})
}
// TODO: why do we need this here?
challengeCount := sectorbuilder.ElectionPostChallengeCount(uint64(len(sectorInfo)), 0)
pvi := abi.PoStVerifyInfo{
Randomness: h.EPostProof.PostRand,
Candidates: candidates,
Proofs: h.EPostProof.Proofs,
EligibleSectors: sectorInfo,
Prover: abi.ActorID(mid),
ChallengeCount: challengeCount,
}
ok, err := sectorbuilder.ProofVerifier.VerifyElectionPost(ctx, pvi)
2019-11-21 22:21:45 +00:00
if err != nil {
return xerrors.Errorf("failed to verify election post: %w", err)
}
if !ok {
return xerrors.Errorf("election post was invalid")
}
return nil
}
func (syncer *Syncer) checkBlockMessages(ctx context.Context, b *types.FullBlock, baseTs *types.TipSet) error {
2019-12-17 11:20:09 +00:00
{
var sigCids []cid.Cid // this is what we get for people not wanting the marshalcbor method on the cid type
var pubks []bls.PublicKey
for _, m := range b.BlsMessages {
sigCids = append(sigCids, m.Cid())
pubk, err := syncer.sm.GetBlsPublicKey(ctx, m.From, baseTs)
if err != nil {
return xerrors.Errorf("failed to load bls public to validate block: %w", err)
}
pubks = append(pubks, pubk)
}
if err := syncer.verifyBlsAggregate(ctx, b.Header.BLSAggregate, sigCids, pubks); err != nil {
return xerrors.Errorf("bls aggregate signature was invalid: %w", err)
}
}
2019-09-27 23:55:15 +00:00
nonces := make(map[address.Address]uint64)
balances := make(map[address.Address]types.BigInt)
2019-07-05 14:29:17 +00:00
stateroot, _, err := syncer.sm.TipSetState(ctx, baseTs)
if err != nil {
return err
}
2020-02-04 22:19:05 +00:00
cst := cbor.NewCborStore(syncer.store.Blockstore())
2019-09-27 23:55:15 +00:00
st, err := state.LoadStateTree(cst, stateroot)
if err != nil {
2019-09-27 23:55:15 +00:00
return xerrors.Errorf("failed to load base state tree: %w", err)
2019-07-05 14:29:17 +00:00
}
2019-09-27 23:55:15 +00:00
checkMsg := func(m *types.Message) error {
if m.To == address.Undef {
return xerrors.New("'To' address cannot be empty")
}
2019-09-27 23:55:15 +00:00
if _, ok := nonces[m.From]; !ok {
act, err := st.GetActor(m.From)
if err != nil {
return xerrors.Errorf("failed to get actor: %w", err)
}
nonces[m.From] = act.Nonce
balances[m.From] = act.Balance
}
2019-09-27 23:55:15 +00:00
if nonces[m.From] != m.Nonce {
return xerrors.Errorf("wrong nonce (exp: %d, got: %d)", nonces[m.From], m.Nonce)
2019-07-05 14:29:17 +00:00
}
2019-09-27 23:55:15 +00:00
nonces[m.From]++
2019-07-05 14:29:17 +00:00
2019-09-27 23:55:15 +00:00
if balances[m.From].LessThan(m.RequiredFunds()) {
return xerrors.Errorf("not enough funds for message execution")
}
2019-07-05 14:29:17 +00:00
2019-09-27 23:55:15 +00:00
balances[m.From] = types.BigSub(balances[m.From], m.RequiredFunds())
return nil
2019-07-05 14:29:17 +00:00
}
var blsCids []cbg.CBORMarshaler
2019-09-27 23:55:15 +00:00
for i, m := range b.BlsMessages {
if err := checkMsg(m); err != nil {
return xerrors.Errorf("block had invalid bls message at index %d: %w", i, err)
2019-09-27 23:55:15 +00:00
}
c := cbg.CborCid(m.Cid())
blsCids = append(blsCids, &c)
2019-07-05 14:29:17 +00:00
}
var secpkCids []cbg.CBORMarshaler
2019-09-27 23:55:15 +00:00
for i, m := range b.SecpkMessages {
if err := checkMsg(&m.Message); err != nil {
return xerrors.Errorf("block had invalid secpk message at index %d: %w", i, err)
2019-09-27 23:55:15 +00:00
}
kaddr, err := syncer.sm.ResolveToKeyAddress(ctx, m.Message.From, baseTs)
if err != nil {
return xerrors.Errorf("failed to resolve key addr: %w", err)
}
if err := sigs.Verify(&m.Signature, kaddr, m.Message.Cid().Bytes()); err != nil {
return xerrors.Errorf("secpk message %s has invalid signature: %w", m.Cid(), err)
}
c := cbg.CborCid(m.Cid())
secpkCids = append(secpkCids, &c)
}
bmroot, err := amt.FromArray(ctx, cst, blsCids)
if err != nil {
return xerrors.Errorf("failed to build amt from bls msg cids: %w", err)
}
smroot, err := amt.FromArray(ctx, cst, secpkCids)
if err != nil {
return xerrors.Errorf("failed to build amt from bls msg cids: %w", err)
}
mrcid, err := cst.Put(ctx, &types.MsgMeta{
BlsMessages: bmroot,
SecpkMessages: smroot,
})
if err != nil {
return err
}
if b.Header.Messages != mrcid {
return fmt.Errorf("messages didnt match message root in header")
2019-07-05 14:29:17 +00:00
}
return nil
}
2020-02-12 23:52:36 +00:00
func (syncer *Syncer) verifyBlsAggregate(ctx context.Context, sig crypto.Signature, msgs []cid.Cid, pubks []bls.PublicKey) error {
2019-12-05 01:18:30 +00:00
_, span := trace.StartSpan(ctx, "syncer.verifyBlsAggregate")
2019-10-13 09:08:34 +00:00
defer span.End()
span.AddAttributes(
trace.Int64Attribute("msgCount", int64(len(msgs))),
)
var wg sync.WaitGroup
digests := make([]bls.Digest, len(msgs))
for i := 0; i < 10; i++ {
wg.Add(1)
go func(w int) {
defer wg.Done()
for j := 0; (j*10)+w < len(msgs); j++ {
digests[j*10+w] = bls.Hash(bls.Message(msgs[j*10+w].Bytes()))
}
}(i)
}
wg.Wait()
var bsig bls.Signature
copy(bsig[:], sig.Data)
2019-10-11 03:26:54 +00:00
if !bls.Verify(&bsig, digests, pubks) {
return xerrors.New("bls aggregate signature failed to verify")
}
return nil
}
type syncStateKey struct{}
2019-11-15 21:35:29 +00:00
func extractSyncState(ctx context.Context) *SyncerState {
v := ctx.Value(syncStateKey{})
2019-11-15 21:35:29 +00:00
if v != nil {
return v.(*SyncerState)
}
return nil
}
func (syncer *Syncer) collectHeaders(ctx context.Context, from *types.TipSet, to *types.TipSet) ([]*types.TipSet, error) {
ctx, span := trace.StartSpan(ctx, "collectHeaders")
defer span.End()
2019-11-15 21:35:29 +00:00
ss := extractSyncState(ctx)
span.AddAttributes(
trace.Int64Attribute("fromHeight", int64(from.Height())),
trace.Int64Attribute("toHeight", int64(to.Height())),
)
2019-12-16 19:22:56 +00:00
for _, pcid := range from.Parents().Cids() {
if reason, ok := syncer.bad.Has(pcid); ok {
2019-11-10 23:06:06 +00:00
for _, b := range from.Cids() {
syncer.bad.Add(b, fmt.Sprintf("linked to %s", pcid))
2019-11-10 23:06:06 +00:00
}
return nil, xerrors.Errorf("chain linked to block marked previously as bad (%s, %s) (reason: %s)", from.Cids(), pcid, reason)
2019-11-10 23:06:06 +00:00
}
}
2019-07-31 07:13:49 +00:00
blockSet := []*types.TipSet{from}
2019-07-05 14:29:17 +00:00
2019-07-31 07:13:49 +00:00
at := from.Parents()
2019-07-05 14:29:17 +00:00
// we want to sync all the blocks until the height above the block we have
untilHeight := to.Height() + 1
2019-11-15 21:35:29 +00:00
ss.SetHeight(blockSet[len(blockSet)-1].Height())
var acceptedBlocks []cid.Cid
loop:
for blockSet[len(blockSet)-1].Height() > untilHeight {
2019-12-16 19:22:56 +00:00
for _, bc := range at.Cids() {
if reason, ok := syncer.bad.Has(bc); ok {
for _, b := range acceptedBlocks {
syncer.bad.Add(b, fmt.Sprintf("chain contained %s", bc))
}
return nil, xerrors.Errorf("chain contained block marked previously as bad (%s, %s) (reason: %s)", from.Cids(), bc, reason)
2019-10-09 08:50:57 +00:00
}
}
// If, for some reason, we have a suffix of the chain locally, handle that here
2019-07-31 07:13:49 +00:00
ts, err := syncer.store.LoadTipSet(at)
if err == nil {
2019-12-16 19:22:56 +00:00
acceptedBlocks = append(acceptedBlocks, at.Cids()...)
blockSet = append(blockSet, ts)
at = ts.Parents()
continue
}
if !xerrors.Is(err, bstore.ErrNotFound) {
2019-07-31 07:13:49 +00:00
log.Warn("loading local tipset: %s", err)
}
2019-07-26 18:16:57 +00:00
2019-07-31 07:13:49 +00:00
// NB: GetBlocks validates that the blocks are in-fact the ones we
// requested, and that they are correctly linked to eachother. It does
// not validate any state transitions
window := 500
2019-09-09 20:03:10 +00:00
if gap := int(blockSet[len(blockSet)-1].Height() - untilHeight); gap < window {
window = gap
}
blks, err := syncer.Bsync.GetBlocks(ctx, at, window)
2019-07-31 07:13:49 +00:00
if err != nil {
// Most likely our peers aren't fully synced yet, but forwarded
// new block message (ideally we'd find better peers)
2019-07-26 18:16:57 +00:00
log.Errorf("failed to get blocks: %+v", err)
2019-07-26 18:16:57 +00:00
span.AddAttributes(trace.StringAttribute("error", err.Error()))
2019-07-31 07:13:49 +00:00
// This error will only be logged above,
return nil, xerrors.Errorf("failed to get blocks: %w", err)
}
log.Info("Got blocks: ", blks[0].Height(), len(blks))
2019-07-26 18:16:57 +00:00
2019-07-31 07:13:49 +00:00
for _, b := range blks {
if b.Height() < untilHeight {
2019-09-06 20:03:28 +00:00
break loop
}
2019-10-09 08:50:57 +00:00
for _, bc := range b.Cids() {
if reason, ok := syncer.bad.Has(bc); ok {
for _, b := range acceptedBlocks {
syncer.bad.Add(b, fmt.Sprintf("chain contained %s", bc))
}
return nil, xerrors.Errorf("chain contained block marked previously as bad (%s, %s) (reason: %s)", from.Cids(), bc, reason)
2019-10-09 08:50:57 +00:00
}
}
2019-07-31 07:13:49 +00:00
blockSet = append(blockSet, b)
}
2019-07-26 18:16:57 +00:00
2019-12-16 19:22:56 +00:00
acceptedBlocks = append(acceptedBlocks, at.Cids()...)
2019-11-15 21:35:29 +00:00
ss.SetHeight(blks[len(blks)-1].Height())
2019-07-31 07:13:49 +00:00
at = blks[len(blks)-1].Parents()
}
// We have now ascertained that this is *not* a 'fast forward'
2019-12-16 19:22:56 +00:00
if !types.CidArrsEqual(blockSet[len(blockSet)-1].Parents().Cids(), to.Cids()) {
last := blockSet[len(blockSet)-1]
2019-12-16 19:22:56 +00:00
if last.Parents() == to.Parents() {
// common case: receiving a block thats potentially part of the same tipset as our best block
return blockSet, nil
}
log.Warnf("(fork detected) synced header chain (%s - %d) does not link to our best block (%s - %d)", from.Cids(), from.Height(), to.Cids(), to.Height())
fork, err := syncer.syncFork(ctx, last, to)
if err != nil {
2019-11-10 23:06:06 +00:00
if xerrors.Is(err, ErrForkTooLong) {
// TODO: we're marking this block bad in the same way that we mark invalid blocks bad. Maybe distinguish?
log.Warn("adding forked chain to our bad tipset cache")
for _, b := range from.Blocks() {
syncer.bad.Add(b.Cid(), "fork past finality")
2019-11-10 23:06:06 +00:00
}
}
return nil, xerrors.Errorf("failed to sync fork: %w", err)
}
blockSet = append(blockSet, fork...)
2019-07-31 07:13:49 +00:00
}
2019-07-26 18:16:57 +00:00
2019-07-31 07:13:49 +00:00
return blockSet, nil
}
2019-07-26 18:16:57 +00:00
2019-11-10 23:06:06 +00:00
var ErrForkTooLong = fmt.Errorf("fork longer than threshold")
func (syncer *Syncer) syncFork(ctx context.Context, from *types.TipSet, to *types.TipSet) ([]*types.TipSet, error) {
2019-10-08 00:28:13 +00:00
tips, err := syncer.Bsync.GetBlocks(ctx, from.Parents(), build.ForkLengthThreshold)
if err != nil {
return nil, err
}
nts, err := syncer.store.LoadTipSet(to.Parents())
if err != nil {
return nil, xerrors.Errorf("failed to load next local tipset: %w", err)
}
for cur := 0; cur < len(tips); {
if nts.Height() == 0 {
if !syncer.Genesis.Equals(nts) {
return nil, xerrors.Errorf("somehow synced chain that linked back to a different genesis (bad genesis: %s)", nts.Key())
}
return nil, xerrors.Errorf("synced chain forked at genesis, refusing to sync")
}
if nts.Equals(tips[cur]) {
return tips[:cur+1], nil
}
if nts.Height() < tips[cur].Height() {
cur++
} else {
nts, err = syncer.store.LoadTipSet(nts.Parents())
if err != nil {
return nil, xerrors.Errorf("loading next local tipset: %w", err)
}
}
}
2019-11-10 23:06:06 +00:00
return nil, ErrForkTooLong
}
2019-10-10 11:13:26 +00:00
func (syncer *Syncer) syncMessagesAndCheckState(ctx context.Context, headers []*types.TipSet) error {
2019-11-15 21:35:29 +00:00
ss := extractSyncState(ctx)
ss.SetHeight(0)
return syncer.iterFullTipsets(ctx, headers, func(ctx context.Context, fts *store.FullTipSet) error {
log.Debugw("validating tipset", "height", fts.TipSet().Height(), "size", len(fts.TipSet().Cids()))
2019-10-10 11:13:26 +00:00
if err := syncer.ValidateTipSet(ctx, fts); err != nil {
log.Errorf("failed to validate tipset: %+v", err)
return xerrors.Errorf("message processing failed: %w", err)
}
2019-09-06 20:03:28 +00:00
2019-11-15 21:35:29 +00:00
ss.SetHeight(fts.TipSet().Height())
return nil
})
}
// fills out each of the given tipsets with messages and calls the callback with it
func (syncer *Syncer) iterFullTipsets(ctx context.Context, headers []*types.TipSet, cb func(context.Context, *store.FullTipSet) error) error {
ctx, span := trace.StartSpan(ctx, "iterFullTipsets")
defer span.End()
span.AddAttributes(trace.Int64Attribute("num_headers", int64(len(headers))))
windowSize := 200
for i := len(headers) - 1; i >= 0; {
fts, err := syncer.store.TryFillTipSet(headers[i])
if err != nil {
return err
}
if fts != nil {
if err := cb(ctx, fts); err != nil {
return err
}
i--
continue
}
2019-07-26 18:16:57 +00:00
batchSize := windowSize
if i < batchSize {
batchSize = i
}
next := headers[i-batchSize]
bstips, err := syncer.Bsync.GetChainMessages(ctx, next, uint64(batchSize+1))
2019-07-31 07:13:49 +00:00
if err != nil {
return xerrors.Errorf("message processing failed: %w", err)
}
2019-07-31 07:13:49 +00:00
for bsi := 0; bsi < len(bstips); bsi++ {
2019-09-06 20:03:28 +00:00
// temp storage so we don't persist data we dont want to
ds := dstore.NewMapDatastore()
bs := bstore.NewBlockstore(ds)
blks := cbor.NewCborStore(bs)
2019-09-06 20:03:28 +00:00
this := headers[i-bsi]
2019-07-31 07:13:49 +00:00
bstip := bstips[len(bstips)-(bsi+1)]
2019-09-17 01:56:37 +00:00
fts, err := zipTipSetAndMessages(blks, this, bstip.BlsMessages, bstip.SecpkMessages, bstip.BlsMsgIncludes, bstip.SecpkMsgIncludes)
if err != nil {
log.Warnw("zipping failed", "error", err, "bsi", bsi, "i", i,
"height", this.Height(), "bstip-height", bstip.Blocks[0].Height,
"bstips", bstips, "next-height", i+batchSize)
2019-07-31 07:13:49 +00:00
return xerrors.Errorf("message processing failed: %w", err)
}
2019-07-26 18:16:57 +00:00
if err := cb(ctx, fts); err != nil {
return err
}
2019-07-26 18:16:57 +00:00
2019-09-06 20:03:28 +00:00
if err := persistMessages(bs, bstip); err != nil {
return err
}
2019-09-06 20:03:28 +00:00
if err := copyBlockstore(bs, syncer.store.Blockstore()); err != nil {
return xerrors.Errorf("message processing failed: %w", err)
}
2019-07-05 14:29:17 +00:00
}
i -= windowSize
2019-07-31 07:13:49 +00:00
}
2019-07-05 14:29:17 +00:00
2019-07-31 07:13:49 +00:00
return nil
}
2019-07-05 14:29:17 +00:00
2019-11-09 23:00:22 +00:00
func persistMessages(bs bstore.Blockstore, bst *blocksync.BSTipSet) error {
2019-09-06 20:03:28 +00:00
for _, m := range bst.BlsMessages {
//log.Infof("putting BLS message: %s", m.Cid())
if _, err := store.PutMessage(bs, m); err != nil {
log.Errorf("failed to persist messages: %+v", err)
2019-09-06 20:03:28 +00:00
return xerrors.Errorf("BLS message processing failed: %w", err)
}
2019-09-06 20:03:28 +00:00
}
for _, m := range bst.SecpkMessages {
2020-02-12 23:52:36 +00:00
if m.Signature.Type != crypto.SigTypeSecp256k1 {
return xerrors.Errorf("unknown signature type on message %s: %q", m.Cid(), m.Signature.Type)
2019-09-06 20:03:28 +00:00
}
//log.Infof("putting secp256k1 message: %s", m.Cid())
if _, err := store.PutMessage(bs, m); err != nil {
log.Errorf("failed to persist messages: %+v", err)
2019-09-06 20:03:28 +00:00
return xerrors.Errorf("secp256k1 message processing failed: %w", err)
}
}
return nil
}
func (syncer *Syncer) collectChain(ctx context.Context, ts *types.TipSet) error {
ctx, span := trace.StartSpan(ctx, "collectChain")
defer span.End()
2019-11-15 21:35:29 +00:00
ss := extractSyncState(ctx)
2019-11-15 21:35:29 +00:00
ss.Init(syncer.store.GetHeaviestTipSet(), ts)
headers, err := syncer.collectHeaders(ctx, ts, syncer.store.GetHeaviestTipSet())
2019-07-31 07:13:49 +00:00
if err != nil {
2019-12-04 03:56:29 +00:00
ss.Error(err)
2019-07-31 07:13:49 +00:00
return err
}
span.AddAttributes(trace.Int64Attribute("syncChainLength", int64(len(headers))))
2019-10-10 03:04:10 +00:00
if !headers[0].Equals(ts) {
log.Errorf("collectChain headers[0] should be equal to sync target. Its not: %s != %s", headers[0].Cids(), ts.Cids())
}
2019-11-15 21:35:29 +00:00
ss.SetStage(api.StagePersistHeaders)
2019-11-12 10:18:46 +00:00
toPersist := make([]*types.BlockHeader, 0, len(headers)*build.BlocksPerEpoch)
2019-07-31 07:13:49 +00:00
for _, ts := range headers {
2019-11-12 10:18:46 +00:00
toPersist = append(toPersist, ts.Blocks()...)
}
if err := syncer.store.PersistBlockHeaders(toPersist...); err != nil {
2019-12-04 03:56:29 +00:00
err = xerrors.Errorf("failed to persist synced blocks to the chainstore: %w", err)
ss.Error(err)
return err
2019-07-05 14:29:17 +00:00
}
2019-11-12 10:18:46 +00:00
toPersist = nil
2019-07-05 14:29:17 +00:00
2019-11-15 21:35:29 +00:00
ss.SetStage(api.StageMessages)
2019-10-10 11:13:26 +00:00
if err := syncer.syncMessagesAndCheckState(ctx, headers); err != nil {
2019-12-04 03:56:29 +00:00
err = xerrors.Errorf("collectChain syncMessages: %w", err)
ss.Error(err)
return err
2019-07-31 07:13:49 +00:00
}
2019-11-15 21:35:29 +00:00
ss.SetStage(api.StageSyncComplete)
2019-11-07 00:18:06 +00:00
log.Debugw("new tipset", "height", ts.Height(), "tipset", types.LogCids(ts.Cids()))
2019-07-31 07:13:49 +00:00
return nil
2019-07-05 14:36:08 +00:00
}
2020-02-23 20:00:47 +00:00
func VerifyElectionPoStVRF(ctx context.Context, evrf []byte, rand []byte, worker address.Address) error {
if err := gen.VerifyVRF(ctx, worker, rand, evrf); err != nil {
2019-11-21 22:21:45 +00:00
return xerrors.Errorf("failed to verify post_randomness vrf: %w", err)
}
return nil
}
2019-11-16 01:05:16 +00:00
func (syncer *Syncer) State() []SyncerState {
var out []SyncerState
for _, ss := range syncer.syncmgr.syncStates {
out = append(out, ss.Snapshot())
}
return out
}
func (syncer *Syncer) MarkBad(blk cid.Cid) {
syncer.bad.Add(blk, "manually marked bad")
}
func (syncer *Syncer) CheckBadBlockCache(blk cid.Cid) (string, bool) {
return syncer.bad.Has(blk)
}