cerc-io/lotus
15
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Merge pull request #3442 from filecoin-project/fix/chainwatch/logging-sync-blocks

Browse Source 
Some more Chainwatch improvements
This commit is contained in:
Łukasz Magiera 2020-09-01 10:57:55 +02:00 committed by GitHub
commit 247b5205fb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 27 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/lotus-chainwatch/processor/processor.go
View File

@ -369,7 +369,9 @@ where rnk <= $1
maxBlock = bh.Height
}
}
log.Infow("Gathered Blocks to Process", "start", minBlock, "end", maxBlock)
if minBlock <= maxBlock {
log.Infow("Gathered Blocks to Process", "start", minBlock, "end", maxBlock)
}
return out, rows.Close()
}

5
cmd/lotus-chainwatch/syncer/blockssub.go
View File

@ -11,16 +11,17 @@ import (
func (s *Syncer) subBlocks(ctx context.Context) {
sub, err := s.node.SyncIncomingBlocks(ctx)
if err != nil {
log.Error(err)
log.Errorf("opening incoming block channel: %+v", err)
return
}
log.Infow("Capturing incoming blocks")
for bh := range sub {
err := s.storeHeaders(map[cid.Cid]*types.BlockHeader{
bh.Cid(): bh,
}, false, time.Now())
if err != nil {
log.Errorf("%+v", err)
log.Errorf("storing incoming block header: %+v", err)
}
}
}

28
cmd/lotus-chainwatch/syncer/sync.go
View File

@ -160,23 +160,37 @@ func (s *Syncer) Start(ctx context.Context) {
log.Fatal(err)
}
// capture all reported blocks
go s.subBlocks(ctx)
// we need to ensure that on a restart we don't reprocess the whole flarping chain
var sinceEpoch uint64
blkCID, height, err := s.mostRecentlySyncedBlockHeight()
if err != nil {
log.Fatalw("failed to find most recently synced block", "error", err)
} else {
if height > 0 {
log.Infow("Found starting point for syncing", "blockCID", blkCID.String(), "height", height)
sinceEpoch = uint64(height)
}
}
// continue to keep the block headers table up to date.
notifs, err := s.node.ChainNotify(ctx)
if err != nil {
log.Fatal(err)
}
// we need to ensure that on a restart we don't reprocess the whole flarping chain
blkCID, height, err := s.mostRecentlySyncedBlockHeight()
if err != nil {
log.Fatalw("failed to find most recently synced block", "error", err)
}
log.Infow("Found starting point for syncing", "blockCID", blkCID.String(), "height", height)
sinceEpoch := uint64(height)
go func() {
for notif := range notifs {
for _, change := range notif {
switch change.Type {
case store.HCCurrent:
// This case is important for capturing the initial state of a node
// which might be on a dead network with no new blocks being produced.
// It also allows a fresh Chainwatch instance to start walking the
// chain without waiting for a new block to come along.
fallthrough
case store.HCApply:
unsynced, err := s.unsyncedBlocks(ctx, change.Val, sinceEpoch)
if err != nil {