Fetch and skip sync events

cmd/devnet/services/polygon/heimdall.go

@@ -238,6 +238,10 @@ func (h *Heimdall) FetchStateSyncEvents(ctx context.Context, fromID uint64, to t
 	return nil, fmt.Errorf("TODO")
 }
 
+func (h *Heimdall) FetchStateSyncEvent(ctx context.Context, id uint64) (*heimdall.EventRecordWithTime, error) {
+	return nil, fmt.Errorf("TODO")
+}
+
 func (h *Heimdall) Close() {
 	h.unsubscribe()
 }

eth/stagedsync/bor_heimdall_shared.go

@@ -191,7 +191,7 @@ func fetchAndWriteHeimdallStateSyncEvents(
 
 	from = lastStateSyncEventID + 1
 
-	logger.Debug(
+	logger.Trace(
 		fmt.Sprintf("[%s] Fetching state updates from Heimdall", logPrefix),
 		"fromID", from,
 		"to", to.Format(time.RFC3339),

eth/stagedsync/stage_bor_heimdall.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/binary"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sort"
 	"time"
@@ -14,10 +15,11 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/ledgerwatch/erigon-lib/chain"
-	"github.com/ledgerwatch/erigon-lib/common"
+	"github.com/ledgerwatch/erigon-lib/chain/networkname"
 	libcommon "github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/erigon-lib/kv"
 	"github.com/ledgerwatch/erigon/accounts/abi"
+	"github.com/ledgerwatch/erigon/common/math"
 	"github.com/ledgerwatch/erigon/consensus"
 	"github.com/ledgerwatch/erigon/core/types"
 	"github.com/ledgerwatch/erigon/dataflow"
@@ -90,6 +92,8 @@ func StageBorHeimdallCfg(
 	}
 }
 
+var lastMumbaiEventRecord *heimdall.EventRecordWithTime
+
 func BorHeimdallForward(
 	s *StageState,
 	u Unwinder,
@@ -168,6 +172,8 @@ func BorHeimdallForward(
 	var fetchTime time.Duration
 	var snapTime time.Duration
 	var snapInitTime time.Duration
+	var syncEventTime time.Duration
+
 	var eventRecords int
 
 	lastSpanID, err := fetchRequiredHeimdallSpansIfNeeded(ctx, headNumber, tx, cfg, s.LogPrefix(), logger)
@@ -185,6 +191,9 @@ func BorHeimdallForward(
 	defer logTimer.Stop()
 
 	logger.Info(fmt.Sprintf("[%s] Processing sync events...", s.LogPrefix()), "from", lastBlockNum+1, "to", headNumber)
+
+	var nextEventRecord *heimdall.EventRecordWithTime
+
 	for blockNum = lastBlockNum + 1; blockNum <= headNumber; blockNum++ {
 		select {
 		default:
@@ -195,7 +204,8 @@ func BorHeimdallForward(
 				"lastSpanID", lastSpanID,
 				"lastStateSyncEventID", lastStateSyncEventID,
 				"total records", eventRecords,
-				"fetch time", fetchTime,
+				"sync-events", syncEventTime,
+				"sync-event-fetch", fetchTime,
 				"snaps", snapTime,
 				"snap-init", snapInitTime,
 				"process time", time.Since(processStart),
@@ -291,23 +301,70 @@ func BorHeimdallForward(
 			return err
 		}
 
+		syncEventStart := time.Now()
+
 		var callTime time.Duration
-		var records int
-		lastStateSyncEventID, records, callTime, err = fetchRequiredHeimdallStateSyncEventsIfNeeded(
-			ctx,
-			header,
-			tx,
-			cfg,
-			s.LogPrefix(),
-			logger,
-			lastStateSyncEventID,
-		)
-		if err != nil {
-			return err
+
+		var endStateSyncEventId uint64
+
+		// mumbai event records have stopped being produced as of march 2024
+		// as part of the goerli decom - so there is no point trying to
+		// fetch them
+		if cfg.chainConfig.ChainName == networkname.MumbaiChainName {
+			if nextEventRecord == nil {
+				nextEventRecord = lastMumbaiEventRecord
+			}
+		}
+
+		if nextEventRecord == nil || header.Time > uint64(nextEventRecord.Time.Unix()) {
+			var records int
+
+			if lastStateSyncEventID == 0 || lastStateSyncEventID != endStateSyncEventId {
+				lastStateSyncEventID, records, callTime, err = fetchRequiredHeimdallStateSyncEventsIfNeeded(
+					ctx,
+					header,
+					tx,
+					cfg,
+					s.LogPrefix(),
+					logger,
+					lastStateSyncEventID,
+				)
+
+				if err != nil {
+					return err
+				}
+			}
+
+			if records != 0 {
+				nextEventRecord = nil
+				eventRecords += records
+			} else {
+				if nextEventRecord == nil || nextEventRecord.ID <= lastStateSyncEventID {
+					if eventRecord, err := cfg.heimdallClient.FetchStateSyncEvent(ctx, lastStateSyncEventID+1); err == nil {
+						nextEventRecord = eventRecord
+						endStateSyncEventId = 0
+					} else {
+						if !errors.Is(err, heimdall.ErrEventRecordNotFound) {
+							return err
+						}
+
+						if cfg.chainConfig.ChainName == networkname.MumbaiChainName && lastStateSyncEventID == 276850 {
+							lastMumbaiEventRecord = &heimdall.EventRecordWithTime{
+								EventRecord: heimdall.EventRecord{
+									ID: 276851,
+								},
+								Time: time.Unix(math.MaxInt64, 0),
+							}
+						}
+
+						endStateSyncEventId = lastStateSyncEventID
+					}
+				}
+			}
 		}
 
-		eventRecords += records
 		fetchTime += callTime
+		syncEventTime = syncEventTime + time.Since(syncEventStart)
 
 		if cfg.loopBreakCheck != nil && cfg.loopBreakCheck(int(blockNum-lastBlockNum)) {
 			headNumber = blockNum
@@ -331,6 +388,7 @@ func BorHeimdallForward(
 		"lastSpanID", lastSpanID,
 		"lastStateSyncEventID", lastStateSyncEventID,
 		"total records", eventRecords,
+		"sync event time", syncEventTime,
 		"fetch time", fetchTime,
 		"snap time", snapTime,
 		"process time", time.Since(processStart),
@@ -455,7 +513,7 @@ func persistValidatorSets(
 		var err error
 		if snap, err = snap.Apply(parent, headers, logger); err != nil {
 			if snap != nil {
-				var badHash common.Hash
+				var badHash libcommon.Hash
 				for _, header := range headers {
 					if header.Number.Uint64() == snap.Number+1 {
 						badHash = header.Hash()

eth/stagedsync/stagedsynctest/harness.go

@@ -651,6 +651,13 @@ func (h *Harness) mockHeimdallClient() {
 			return []*heimdall.EventRecordWithTime{&newEvent}, nil
 		}).
 		AnyTimes()
+	h.heimdallClient.
+		EXPECT().
+		FetchStateSyncEvent(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ uint64) (*heimdall.EventRecordWithTime, error) {
+			return nil, heimdall.ErrEventRecordNotFound
+		}).
+		AnyTimes()
 }
 
 func (h *Harness) runSyncStageForwardWithErrorIs(

polygon/bor/bor_test.go

@@ -56,6 +56,10 @@ func (h test_heimdall) FetchStateSyncEvents(ctx context.Context, fromID uint64,
 	return nil, nil
 }
 
+func (h *test_heimdall) FetchStateSyncEvent(ctx context.Context, id uint64) (*heimdall.EventRecordWithTime, error) {
+	return nil, nil
+}
+
 func (h *test_heimdall) FetchSpan(ctx context.Context, spanID uint64) (*heimdall.Span, error) {
 
 	if span, ok := h.spans[heimdall.SpanId(spanID)]; ok {

polygon/heimdall/client.go

@@ -40,6 +40,7 @@ const (
 //go:generate mockgen -destination=./client_mock.go -package=heimdall . HeimdallClient
 type HeimdallClient interface {
 	FetchStateSyncEvents(ctx context.Context, fromId uint64, to time.Time, limit int) ([]*EventRecordWithTime, error)
+	FetchStateSyncEvent(ctx context.Context, id uint64) (*EventRecordWithTime, error)
 
 	FetchLatestSpan(ctx context.Context) (*Span, error)
 	FetchSpan(ctx context.Context, spanID uint64) (*Span, error)
@@ -106,6 +107,7 @@ func newHeimdallClient(urlString string, httpClient HttpClient, retryBackOff tim
 const (
 	fetchStateSyncEventsFormat = "from-id=%d&to-time=%d&limit=%d"
 	fetchStateSyncEventsPath   = "clerk/event-record/list"
+	fetchStateSyncEvent        = "clerk/event-record/%s"
 
 	fetchCheckpoint                = "/checkpoints/%s"
 	fetchCheckpointCount           = "/checkpoints/count"
@@ -130,12 +132,12 @@ func (c *Client) FetchStateSyncEvents(ctx context.Context, fromID uint64, to tim
 	eventRecords := make([]*EventRecordWithTime, 0)
 
 	for {
-		url, err := stateSyncURL(c.urlString, fromID, to.Unix())
+		url, err := stateSyncListURL(c.urlString, fromID, to.Unix())
 		if err != nil {
 			return nil, err
 		}
 
-		c.logger.Debug(heimdallLogPrefix("Fetching state sync events"), "queryParams", url.RawQuery)
+		c.logger.Trace(heimdallLogPrefix("Fetching state sync events"), "queryParams", url.RawQuery)
 
 		ctx = withRequestType(ctx, stateSyncRequest)
 
@@ -173,6 +175,32 @@ func (c *Client) FetchStateSyncEvents(ctx context.Context, fromID uint64, to tim
 	return eventRecords, nil
 }
 
+func (c *Client) FetchStateSyncEvent(ctx context.Context, id uint64) (*EventRecordWithTime, error) {
+	url, err := stateSyncURL(c.urlString, id)
+
+	if err != nil {
+		return nil, err
+	}
+
+	ctx = withRequestType(ctx, stateSyncRequest)
+
+	isRecoverableError := func(err error) bool {
+		return !strings.Contains(err.Error(), "could not get state record; No record found")
+	}
+
+	response, err := FetchWithRetryEx[StateSyncEventResponse](ctx, c, url, isRecoverableError, c.logger)
+
+	if err != nil {
+		if strings.Contains(err.Error(), "could not get state record; No record found") {
+			return nil, ErrEventRecordNotFound
+		}
+
+		return nil, err
+	}
+
+	return &response.Result, nil
+}
+
 func (c *Client) FetchLatestSpan(ctx context.Context) (*Span, error) {
 	url, err := latestSpanURL(c.urlString)
 	if err != nil {
@@ -457,12 +485,16 @@ func latestSpanURL(urlString string) (*url.URL, error) {
 	return makeURL(urlString, fetchSpanLatest, "")
 }
 
-func stateSyncURL(urlString string, fromID uint64, to int64) (*url.URL, error) {
+func stateSyncListURL(urlString string, fromID uint64, to int64) (*url.URL, error) {
 	queryParams := fmt.Sprintf(fetchStateSyncEventsFormat, fromID, to, stateFetchLimit)
 
 	return makeURL(urlString, fetchStateSyncEventsPath, queryParams)
 }
 
+func stateSyncURL(urlString string, id uint64) (*url.URL, error) {
+	return makeURL(urlString, fmt.Sprintf(fetchStateSyncEvent, fmt.Sprint(id)), "")
+}
+
 func checkpointURL(urlString string, number int64) (*url.URL, error) {
 	url := ""
 	if number == -1 {

polygon/heimdall/event_record.go

@@ -28,6 +28,8 @@ type EventRecordWithTime struct {
 	Time time.Time `json:"record_time" yaml:"record_time"`
 }
 
+var ErrEventRecordNotFound = fmt.Errorf("event record not found")
+
 // String returns the string representatin of a state record
 func (e *EventRecordWithTime) String() string {
 	return fmt.Sprintf(
@@ -78,3 +80,8 @@ type StateSyncEventsResponse struct {
 	Height string                 `json:"height"`
 	Result []*EventRecordWithTime `json:"result"`
 }
+
+type StateSyncEventResponse struct {
+	Height string              `json:"height"`
+	Result EventRecordWithTime `json:"result"`
+}