Merge pull request #3668 from filecoin-project/feat/sync-fetching-messages-state

Add StageFetchingMessages to sync status

api/api_full.go

@@ -2,6 +2,7 @@ package api
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/filecoin-project/specs-actors/actors/runtime/proof"
@@ -709,8 +710,28 @@ const (
 	StageMessages
 	StageSyncComplete
 	StageSyncErrored
+	StageFetchingMessages
 )
 
+func (v SyncStateStage) String() string {
+	switch v {
+	case StageHeaders:
+		return "header sync"
+	case StagePersistHeaders:
+		return "persisting headers"
+	case StageMessages:
+		return "message sync"
+	case StageSyncComplete:
+		return "complete"
+	case StageSyncErrored:
+		return "error"
+	case StageFetchingMessages:
+		return "fetching messages"
+	default:
+		return fmt.Sprintf("<unknown: %d>", v)
+	}
+}
+
 type MpoolChange int
 
 const (

chain/sync.go

@@ -1439,6 +1439,7 @@ func (syncer *Syncer) syncMessagesAndCheckState(ctx context.Context, headers []*
 
 // 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 {
+	ss := extractSyncState(ctx)
 	ctx, span := trace.StartSpan(ctx, "iterFullTipsets")
 	defer span.End()
 
@@ -1466,6 +1467,7 @@ mainLoop:
 
 		nextI := (i + 1) - batchSize // want to fetch batchSize values, 'i' points to last one we want to fetch, so its 'inclusive' of our request, thus we need to add one to our request start index
 
+		ss.SetStage(api.StageFetchingMessages)
 		var bstout []*blocksync.CompactedMessages
 		for len(bstout) < batchSize {
 			next := headers[nextI]
@@ -1485,6 +1487,7 @@ mainLoop:
 			bstout = append(bstout, bstips...)
 			nextI += len(bstips)
 		}
+		ss.SetStage(api.StageMessages)
 
 		for bsi := 0; bsi < len(bstout); bsi++ {
 			// temp storage so we don't persist data we dont want to

chain/syncstate.go

@@ -1,7 +1,6 @@
 package chain
 
 import (
-	"fmt"
 	"sync"
 	"time"
 
@@ -12,23 +11,6 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 )
 
-func SyncStageString(v api.SyncStateStage) string {
-	switch v {
-	case api.StageHeaders:
-		return "header sync"
-	case api.StagePersistHeaders:
-		return "persisting headers"
-	case api.StageMessages:
-		return "message sync"
-	case api.StageSyncComplete:
-		return "complete"
-	case api.StageSyncErrored:
-		return "error"
-	default:
-		return fmt.Sprintf("<unknown: %d>", v)
-	}
-}
-
 type SyncerState struct {
 	lk      sync.Mutex
 	Target  *types.TipSet

cli/sync.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/chain"
 )
 
 var syncCmd = &cli.Command{
@@ -61,7 +60,7 @@ var syncStatusCmd = &cli.Command{
 			fmt.Printf("\tBase:\t%s\n", base)
 			fmt.Printf("\tTarget:\t%s (%d)\n", target, theight)
 			fmt.Printf("\tHeight diff:\t%d\n", heightDiff)
-			fmt.Printf("\tStage: %s\n", chain.SyncStageString(ss.Stage))
+			fmt.Printf("\tStage: %s\n", ss.Stage)
 			fmt.Printf("\tHeight: %d\n", ss.Height)
 			if ss.End.IsZero() {
 				if !ss.Start.IsZero() {
@@ -186,7 +185,7 @@ func SyncWait(ctx context.Context, napi api.FullNode) error {
 			theight = ss.Target.Height()
 		}
 
-		fmt.Printf("\r\x1b[2KWorker %d: Target Height: %d\tTarget: %s\tState: %s\tHeight: %d", working, theight, target, chain.SyncStageString(ss.Stage), ss.Height)
+		fmt.Printf("\r\x1b[2KWorker %d: Target Height: %d\tTarget: %s\tState: %s\tHeight: %d", working, theight, target, ss.Stage, ss.Height)
 
 		if time.Now().Unix()-int64(head.MinTimestamp()) < int64(build.BlockDelaySecs) {
 			fmt.Println("\nDone!")

tools/stats/rpc.go

@@ -14,7 +14,6 @@ import (
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/api/client"
 	"github.com/filecoin-project/lotus/build"
-	"github.com/filecoin-project/lotus/chain"
 	"github.com/filecoin-project/lotus/chain/store"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/node/repo"
@@ -72,7 +71,7 @@ sync_complete:
 						"target_height", w.Target.Height(),
 						"height", w.Height,
 						"error", w.Message,
-						"stage", chain.SyncStageString(w.Stage),
+						"stage", w.Stage.String(),
 					)
 				} else {
 					log.Infow(
@@ -82,7 +81,7 @@ sync_complete:
 						"target", w.Target.Key(),
 						"target_height", w.Target.Height(),
 						"height", w.Height,
-						"stage", chain.SyncStageString(w.Stage),
+						"stage", w.Stage.String(),
 					)
 				}