polygon/bridge: fix snapshot stores, logging, replay blocks if needed on startup

polygon/bridge/service.go

@@ -168,8 +168,8 @@ func (s *Service) Run(ctx context.Context) error {
 	}
 
 	// start syncing
-	s.logger.Debug(
-		bridgeLogPrefix("running bridge component"),
+	s.logger.Info(
+		bridgeLogPrefix("running bridge service component"),
 		"lastFetchedEventId", lastFetchedEventId,
 		"lastProcessedEventId", lastProcessedEventId,
 		"lastProcessedBlockNum", lastProcessedBlockInfo.BlockNum,
@@ -238,7 +238,7 @@ func (s *Service) Run(ctx context.Context) error {
 
 		select {
 		case <-logTicker.C:
-			s.logger.Debug(
+			s.logger.Info(
 				bridgeLogPrefix("fetched new events periodic progress"),
 				"count", len(events),
 				"lastFetchedEventId", lastFetchedEventId,
@@ -257,20 +257,20 @@ func (s *Service) InitialBlockReplayNeeded(ctx context.Context) (uint64, bool, e
 	lastFrozen := s.store.LastFrozenEventBlockNum()
 
 	if blockInfo := s.lastProcessedBlockInfo.Load(); blockInfo != nil && blockInfo.BlockNum > lastFrozen {
-		return 0, false, nil
+		return blockInfo.BlockNum, false, nil
 	}
 
 	blockInfo, ok, err := s.store.LastProcessedBlockInfo(ctx)
 	if err != nil {
 		return 0, false, err
 	}
 	if ok && blockInfo.BlockNum > lastFrozen {
-		// we have all info, no need to replay
-		return 0, false, nil
+		// we have all info, no need to replay initial block
+		return blockInfo.BlockNum, false, nil
 	}
 
 	// replay the last block we have in event snapshots
-	return s.store.LastFrozenEventBlockNum(), true, nil
+	return lastFrozen, true, nil
 }
 
 func (s *Service) ReplayInitialBlock(ctx context.Context, block *types.Block) error {

polygon/bridge/snapshot_store.go

@@ -83,7 +83,7 @@ func (s *SnapshotStore) LastFrozenEventBlockNum() uint64 {
 	if len(segments) == 0 {
 		return 0
 	}
-	// find the last segment which has a built index
+	// find the last segment which has a built non-empty index
 	var lastSegment *snapshotsync.VisibleSegment
 	for i := len(segments) - 1; i >= 0; i-- {
 		if segments[i].Src().Index() != nil {
@@ -161,7 +161,7 @@ func (s *SnapshotStore) LastFrozenEventId() uint64 {
 	if len(segments) == 0 {
 		return 0
 	}
-	// find the last segment which has a built index
+	// find the last segment which has a built non-empty index
 	var lastSegment *snapshotsync.VisibleSegment
 	for i := len(segments) - 1; i >= 0; i-- {
 		if segments[i].Src().Index() != nil {

polygon/heimdall/entity_fetcher.go

@@ -94,8 +94,23 @@ func (f *EntityFetcher[TEntity]) FetchEntitiesRange(ctx context.Context, idRange
 }
 
 func (f *EntityFetcher[TEntity]) FetchEntitiesRangeSequentially(ctx context.Context, idRange ClosedRange) ([]TEntity, error) {
+	progressLogTicker := time.NewTicker(30 * time.Second)
+	defer progressLogTicker.Stop()
+
 	entities := make([]TEntity, 0, idRange.Len())
 	for id := idRange.Start; id <= idRange.End; id++ {
+		select {
+		case <-progressLogTicker.C:
+			f.logger.Info(
+				heimdallLogPrefix(f.name+" fetch entities sequentially periodic progress"),
+				"current", id,
+				"start", idRange.Start,
+				"end", idRange.End,
+			)
+		default:
+			// carry-on
+		}
+
 		entity, err := f.fetchEntity(ctx, int64(id))
 		if err != nil {
 			// return fetched entities up to this point in case of transient errors
@@ -134,8 +149,8 @@ func (f *EntityFetcher[TEntity]) FetchAllEntities(ctx context.Context) ([]TEntit
 
 		select {
 		case <-progressLogTicker.C:
-			f.logger.Debug(
-				heimdallLogPrefix(f.name+" progress"),
+			f.logger.Info(
+				heimdallLogPrefix(f.name+" fetch all entities periodic progress"),
 				"page", page,
 				"len", len(entities),
 			)
@@ -155,7 +170,7 @@ func (f *EntityFetcher[TEntity]) FetchAllEntities(ctx context.Context) ([]TEntit
 	}
 
 	f.logger.Debug(
-		heimdallLogPrefix(f.name+" done"),
+		heimdallLogPrefix(f.name+" fetch all entities done"),
 		"len", len(entities),
 		"duration", time.Since(fetchStartTime),
 	)

polygon/heimdall/scraper.go

@@ -31,6 +31,7 @@ import (
 )
 
 type Scraper[TEntity Entity] struct {
+	name            string
 	store           EntityStore[TEntity]
 	fetcher         entityFetcher[TEntity]
 	pollDelay       time.Duration
@@ -41,13 +42,15 @@ type Scraper[TEntity Entity] struct {
 }
 
 func NewScraper[TEntity Entity](
+	name string,
 	store EntityStore[TEntity],
 	fetcher entityFetcher[TEntity],
 	pollDelay time.Duration,
 	transientErrors []error,
 	logger log.Logger,
 ) *Scraper[TEntity] {
 	return &Scraper[TEntity]{
+		name:            name,
 		store:           store,
 		fetcher:         fetcher,
 		pollDelay:       pollDelay,
@@ -59,11 +62,16 @@ func NewScraper[TEntity Entity](
 }
 
 func (s *Scraper[TEntity]) Run(ctx context.Context) error {
+	s.logger.Info(heimdallLogPrefix("running scraper component"), "name", s.name)
+
 	defer s.store.Close()
 	if err := s.store.Prepare(ctx); err != nil {
 		return err
 	}
 
+	progressLogTicker := time.NewTicker(30 * time.Second)
+	defer progressLogTicker.Stop()
+
 	for ctx.Err() == nil {
 		lastKnownId, hasLastKnownId, err := s.store.LastEntityId(ctx)
 		if err != nil {
@@ -116,8 +124,25 @@ func (s *Scraper[TEntity]) Run(ctx context.Context) error {
 			}
 
 			s.observers.NotifySync(entities) // NotifySync preserves order of events
+
+			select {
+			case <-progressLogTicker.C:
+				if len(entities) > 0 {
+					s.logger.Info(
+						heimdallLogPrefix("scraper periodic progress"),
+						"name", s.name,
+						"rangeStart", idRange.Start,
+						"rangeEnd", idRange.End,
+						"priorLastKnownId", lastKnownId,
+						"newLast", entities[len(entities)-1].RawId(),
+					)
+				}
+			default:
+				// carry on
+			}
 		}
 	}
+
 	return ctx.Err()
 }
 

polygon/heimdall/scraper_test.go

@@ -72,7 +72,7 @@ func TestScrapper_Run_TransientErr(t *testing.T) {
 	)
 
 	transientErrs := []error{ErrNotInMilestoneList, ErrBadGateway}
-	scrapper := NewScraper[*Milestone](store, fetcher, time.Millisecond, transientErrs, logger)
+	scrapper := NewScraper[*Milestone]("test", store, fetcher, time.Millisecond, transientErrs, logger)
 
 	eg, ctx := errgroup.WithContext(ctx)
 	eg.Go(func() error {

polygon/heimdall/service.go

@@ -60,6 +60,7 @@ func NewService(config ServiceConfig) *Service {
 	spanFetcher := NewSpanFetcher(client, logger)
 
 	checkpointScraper := NewScraper(
+		"checkpoints",
 		store.Checkpoints(),
 		checkpointFetcher,
 		1*time.Second,
@@ -74,6 +75,7 @@ func NewService(config ServiceConfig) *Service {
 	milestoneScraperTransientErrors := []error{ErrNotInMilestoneList}
 	milestoneScraperTransientErrors = append(milestoneScraperTransientErrors, TransientErrors...)
 	milestoneScraper := NewScraper(
+		"milestones",
 		store.Milestones(),
 		milestoneFetcher,
 		1*time.Second,
@@ -82,6 +84,7 @@ func NewService(config ServiceConfig) *Service {
 	)
 
 	spanScraper := NewScraper(
+		"spans",
 		store.Spans(),
 		spanFetcher,
 		1*time.Second,
@@ -297,8 +300,9 @@ func (s *Service) Ready(ctx context.Context) <-chan error {
 }
 
 func (s *Service) Run(ctx context.Context) error {
-	defer s.store.Close()
+	s.logger.Info(heimdallLogPrefix("running heimdall service component"))
 
+	defer s.store.Close()
 	if err := s.store.Prepare(ctx); err != nil {
 		return nil
 	}

polygon/heimdall/snapshot_store.go

@@ -102,12 +102,15 @@ func (s *SpanSnapshotStore) LastFrozenEntityId() uint64 {
 	if len(segments) == 0 {
 		return 0
 	}
-	// find the last segment which has a built index
+	// find the last segment which has a built non-empty index
 	var lastSegment *snapshotsync.VisibleSegment
 	for i := len(segments) - 1; i >= 0; i-- {
 		if segments[i].Src().Index() != nil {
-			lastSegment = segments[i]
-			break
+			gg := segments[i].Src().MakeGetter()
+			if gg.HasNext() {
+				lastSegment = segments[i]
+				break
+			}
 		}
 	}
 	if lastSegment == nil {
@@ -221,12 +224,15 @@ func (r *milestoneSnapshotStore) LastFrozenEntityId() uint64 {
 	if len(segments) == 0 {
 		return 0
 	}
-	// find the last segment which has a built index
+	// find the last segment which has a built non-empty index
 	var lastSegment *snapshotsync.VisibleSegment
 	for i := len(segments) - 1; i >= 0; i-- {
 		if segments[i].Src().Index() != nil {
-			lastSegment = segments[i]
-			break
+			gg := segments[i].Src().MakeGetter()
+			if gg.HasNext() {
+				lastSegment = segments[i]
+				break
+			}
 		}
 	}
 	if lastSegment == nil {
@@ -380,12 +386,15 @@ func (r *checkpointSnapshotStore) LastFrozenEntityId() uint64 {
 	if len(segments) == 0 {
 		return 0
 	}
-	// find the last segment which has a built index
+	// find the last segment which has a built non-empty index
 	var lastSegment *snapshotsync.VisibleSegment
 	for i := len(segments) - 1; i >= 0; i-- {
 		if segments[i].Src().Index() != nil {
-			lastSegment = segments[i]
-			break
+			gg := segments[i].Src().MakeGetter()
+			if gg.HasNext() {
+				lastSegment = segments[i]
+				break
+			}
 		}
 	}
 

polygon/heimdall/span_block_producers_tracker.go

@@ -61,8 +61,9 @@ type spanBlockProducersTracker struct {
 }
 
 func (t *spanBlockProducersTracker) Run(ctx context.Context) error {
-	defer close(t.idleSignal)
+	t.logger.Info(heimdallLogPrefix("running span block producers tracker component"))
 
+	defer close(t.idleSignal)
 	for {
 		select {
 		case <-ctx.Done():

polygon/p2p/message_listener.go

@@ -73,7 +73,7 @@ type MessageListener struct {
 }
 
 func (ml *MessageListener) Run(ctx context.Context) error {
-	ml.logger.Debug(messageListenerLogPrefix("running p2p message listener component"))
+	ml.logger.Info(messageListenerLogPrefix("running p2p message listener component"))
 
 	backgroundLoops := []func(ctx context.Context){
 		ml.listenInboundMessages,

polygon/p2p/peer_tracker.go

@@ -63,7 +63,7 @@ type PeerTracker struct {
 }
 
 func (pt *PeerTracker) Run(ctx context.Context) error {
-	pt.logger.Debug(peerTrackerLogPrefix("running peer tracker component"))
+	pt.logger.Info(peerTrackerLogPrefix("running peer tracker component"))
 
 	var peerEventUnreg polygoncommon.UnregisterFunc
 	defer func() { peerEventUnreg() }()

polygon/p2p/publisher.go

@@ -73,7 +73,8 @@ func (p Publisher) PublishNewBlockHashes(block *types.Block) {
 }
 
 func (p Publisher) Run(ctx context.Context) error {
-	p.logger.Debug("[p2p-publisher] running publisher")
+	p.logger.Info("[p2p-publisher] running publisher component")
+
 	for {
 		select {
 		case <-ctx.Done():

polygon/p2p/service.go

@@ -43,6 +43,7 @@ func NewService(logger log.Logger, maxPeers int, sc sentryproto.SentryClient, sd
 	fetcher = NewTrackingFetcher(fetcher, peerTracker)
 	publisher := NewPublisher(logger, messageSender, peerTracker)
 	return &Service{
+		logger:          logger,
 		fetcher:         fetcher,
 		messageListener: messageListener,
 		peerPenalizer:   peerPenalizer,
@@ -53,6 +54,7 @@ func NewService(logger log.Logger, maxPeers int, sc sentryproto.SentryClient, sd
 }
 
 type Service struct {
+	logger          log.Logger
 	fetcher         Fetcher
 	messageListener *MessageListener
 	peerPenalizer   *PeerPenalizer
@@ -62,6 +64,8 @@ type Service struct {
 }
 
 func (s *Service) Run(ctx context.Context) error {
+	s.logger.Info("[p2p] running p2p service component")
+
 	eg, ctx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
 		if err := s.messageListener.Run(ctx); err != nil {

polygon/sync/block_downloader.go

@@ -138,7 +138,7 @@ func (d *BlockDownloader) downloadBlocksUsingWaypoints(
 
 	waypoints = d.limitWaypoints(waypoints)
 
-	d.logger.Debug(
+	d.logger.Info(
 		syncLogPrefix("downloading blocks using waypoints"),
 		"waypointsLen", len(waypoints),
 		"start", waypoints[0].StartBlock().Uint64(),
@@ -283,17 +283,22 @@ func (d *BlockDownloader) downloadBlocksUsingWaypoints(
 			continue
 		}
 
-		d.logger.Debug(syncLogPrefix("fetched blocks"), "start", blocks[0].NumberU64(), "end", blocks[len(blocks)-1].NumberU64(),
+		d.logger.Debug(
+			syncLogPrefix("fetched blocks"),
+			"start", blocks[0].NumberU64(),
+			"end", blocks[len(blocks)-1].NumberU64(),
 			"blocks", len(blocks),
 			"duration", time.Since(batchFetchStartTime),
-			"blks/sec", float64(len(blocks))/math.Max(time.Since(batchFetchStartTime).Seconds(), 0.0001))
+			"blks/sec", float64(len(blocks))/math.Max(time.Since(batchFetchStartTime).Seconds(), 0.0001),
+		)
 
 		batchFetchStartTime = time.Now() // reset for next time
 
 		d.logger.Info(
-			syncLogPrefix(fmt.Sprintf("inserting %d fetched blocks", len(blocks))),
+			syncLogPrefix("inserting fetched blocks"),
 			"start", blocks[0].NumberU64(),
 			"end", blocks[len(blocks)-1].NumberU64(),
+			"blocks", len(blocks),
 		)
 		if err := d.store.InsertBlocks(ctx, blocks); err != nil {
 			return nil, err