polygon/bridge: fix bridge integration in stage mode

cmd/rpcdaemon/rpcservices/eth_backend.go

@@ -331,6 +331,10 @@ func (back *RemoteBackend) LastFrozenEventId() uint64 {
 	panic("not implemented")
 }
 
+func (back *RemoteBackend) LastFrozenEventBlockNum() uint64 {
+	panic("not implemented")
+}
+
 func (back *RemoteBackend) Span(ctx context.Context, tx kv.Getter, spanId uint64) ([]byte, error) {
 	return back.blockReader.Span(ctx, tx, spanId)
 }

erigon-lib/kv/tables.go

@@ -331,18 +331,19 @@ const (
 	PendingEpoch = "DevPendingEpoch" // block_num_u64+block_hash->transition_proof
 
 	// BOR
-	BorReceipts           = "BorReceipt"
-	BorFinality           = "BorFinality"
-	BorTxLookup           = "BlockBorTransactionLookup" // transaction_hash -> block_num_u64
-	BorSeparate           = "BorSeparate"               // persisted snapshots of the Validator Sets, with their proposer priorities
-	BorEvents             = "BorEvents"                 // event_id -> event_payload
-	BorEventNums          = "BorEventNums"              // block_num -> event_id (last event_id in that block)
-	BorSpans              = "BorSpans"                  // span_id -> span (in JSON encoding)
-	BorMilestones         = "BorMilestones"             // milestone_id -> milestone (in JSON encoding)
-	BorMilestoneEnds      = "BorMilestoneEnds"          // start block_num -> milestone_id (first block of milestone)
-	BorCheckpoints        = "BorCheckpoints"            // checkpoint_id -> checkpoint (in JSON encoding)
-	BorCheckpointEnds     = "BorCheckpointEnds"         // start block_num -> checkpoint_id (first block of checkpoint)
-	BorProducerSelections = "BorProducerSelections"     // span_id -> span selection with accumulated proposer priorities (in JSON encoding)
+	BorReceipts             = "BorReceipt"
+	BorFinality             = "BorFinality"
+	BorTxLookup             = "BlockBorTransactionLookup" // transaction_hash -> block_num_u64
+	BorSeparate             = "BorSeparate"               // persisted snapshots of the Validator Sets, with their proposer priorities
+	BorEvents               = "BorEvents"                 // event_id -> event_payload
+	BorEventNums            = "BorEventNums"              // block_num -> event_id (last event_id in that block)
+	BorEventProcessedBlocks = "BorEventProcessedBlocks"   // block_num -> block_time, tracks processed blocks in the bridge, used for unwinds and restarts, gets pruned
+	BorSpans                = "BorSpans"                  // span_id -> span (in JSON encoding)
+	BorMilestones           = "BorMilestones"             // milestone_id -> milestone (in JSON encoding)
+	BorMilestoneEnds        = "BorMilestoneEnds"          // start block_num -> milestone_id (first block of milestone)
+	BorCheckpoints          = "BorCheckpoints"            // checkpoint_id -> checkpoint (in JSON encoding)
+	BorCheckpointEnds       = "BorCheckpointEnds"         // start block_num -> checkpoint_id (first block of checkpoint)
+	BorProducerSelections   = "BorProducerSelections"     // span_id -> span selection with accumulated proposer priorities (in JSON encoding)
 
 	// Downloader
 	BittorrentCompletion = "BittorrentCompletion"
@@ -540,6 +541,7 @@ var ChaindataTables = []string{
 	BorSeparate,
 	BorEvents,
 	BorEventNums,
+	BorEventProcessedBlocks,
 	BorSpans,
 	BorMilestones,
 	BorMilestoneEnds,
@@ -743,17 +745,18 @@ var ChaindataTablesCfg = TableCfg{
 }
 
 var BorTablesCfg = TableCfg{
-	BorReceipts:           {Flags: DupSort},
-	BorFinality:           {Flags: DupSort},
-	BorTxLookup:           {Flags: DupSort},
-	BorEvents:             {Flags: DupSort},
-	BorEventNums:          {Flags: DupSort},
-	BorSpans:              {Flags: DupSort},
-	BorCheckpoints:        {Flags: DupSort},
-	BorCheckpointEnds:     {Flags: DupSort},
-	BorMilestones:         {Flags: DupSort},
-	BorMilestoneEnds:      {Flags: DupSort},
-	BorProducerSelections: {Flags: DupSort},
+	BorReceipts:             {Flags: DupSort},
+	BorFinality:             {Flags: DupSort},
+	BorTxLookup:             {Flags: DupSort},
+	BorEvents:               {Flags: DupSort},
+	BorEventNums:            {Flags: DupSort},
+	BorEventProcessedBlocks: {Flags: DupSort},
+	BorSpans:                {Flags: DupSort},
+	BorCheckpoints:          {Flags: DupSort},
+	BorCheckpointEnds:       {Flags: DupSort},
+	BorMilestones:           {Flags: DupSort},
+	BorMilestoneEnds:        {Flags: DupSort},
+	BorProducerSelections:   {Flags: DupSort},
 }
 
 var TxpoolTablesCfg = TableCfg{}

eth/stagedsync/stage_polygon_sync.go

@@ -92,15 +92,18 @@ func NewPolygonSyncStageCfg(
 			txActionStream: txActionStream,
 		},
 	}
+	bridgeStore := &polygonSyncStageBridgeStore{
+		eventReader:    blockReader,
+		txActionStream: txActionStream,
+	}
 	borConfig := chainConfig.Bor.(*borcfg.BorConfig)
 	heimdallService := heimdall.NewService(borConfig, heimdallClient, heimdallStore, logger)
+	bridgeService := bridge.NewBridge(bridgeStore, logger, borConfig, heimdallClient, nil)
 	p2pService := p2p.NewService(maxPeers, logger, sentry, statusDataProvider.GetStatusData)
 	checkpointVerifier := polygonsync.VerifyCheckpointHeaders
 	milestoneVerifier := polygonsync.VerifyMilestoneHeaders
 	blocksVerifier := polygonsync.VerifyBlocks
-	syncStore := &polygonSyncStageSyncStore{
-		executionEngine: executionEngine,
-	}
+	syncStore := polygonsync.NewStore(logger, executionEngine, bridgeService)
 	blockDownloader := polygonsync.NewBlockDownloader(
 		logger,
 		p2pService,
@@ -112,11 +115,6 @@ func NewPolygonSyncStageCfg(
 		blockLimit,
 	)
 	events := polygonsync.NewTipEvents(logger, p2pService, heimdallService)
-	bridgeStore := &polygonSyncStageBridgeStore{
-		eventReader:    blockReader,
-		txActionStream: txActionStream,
-	}
-	bridgeService := bridge.NewBridge(bridgeStore, logger, borConfig, heimdallClient, nil)
 	sync := polygonsync.NewSync(
 		syncStore,
 		executionEngine,
@@ -133,6 +131,7 @@ func NewPolygonSyncStageCfg(
 	syncService := &polygonSyncStageService{
 		logger:          logger,
 		sync:            sync,
+		syncStore:       syncStore,
 		events:          events,
 		p2p:             p2pService,
 		executionEngine: executionEngine,
@@ -318,6 +317,18 @@ func UnwindEvents(tx kv.RwTx, unwindPoint uint64) error {
 			return err
 		}
 	}
+
+	epbCursor, err := tx.RwCursor(kv.BorEventProcessedBlocks)
+	if err != nil {
+		return err
+	}
+
+	defer epbCursor.Close()
+	for k, _, err = epbCursor.Seek(blockNumBuf[:]); err == nil && k != nil; k, _, err = epbCursor.Next() {
+		if err = epbCursor.DeleteCurrent(); err != nil {
+			return err
+		}
+	}
 	return err
 }
 
@@ -399,6 +410,7 @@ type polygonSyncStageTxAction struct {
 type polygonSyncStageService struct {
 	logger          log.Logger
 	sync            *polygonsync.Sync
+	syncStore       polygonsync.Store
 	events          *polygonsync.TipEvents
 	p2p             p2p.Service
 	executionEngine *polygonSyncStageExecutionEngine
@@ -493,6 +505,10 @@ func (s *polygonSyncStageService) runBgComponentsOnce(ctx context.Context) {
 			}
 		})
 
+		eg.Go(func() error {
+			return s.syncStore.Run(ctx)
+		})
+
 		eg.Go(func() error {
 			return s.sync.Run(ctx)
 		})
@@ -503,22 +519,6 @@ func (s *polygonSyncStageService) runBgComponentsOnce(ctx context.Context) {
 	}()
 }
 
-type polygonSyncStageSyncStore struct {
-	executionEngine *polygonSyncStageExecutionEngine
-}
-
-func (s *polygonSyncStageSyncStore) InsertBlocks(ctx context.Context, blocks []*types.Block) error {
-	return s.executionEngine.InsertBlocks(ctx, blocks)
-}
-
-func (s *polygonSyncStageSyncStore) Flush(context.Context) error {
-	return nil
-}
-
-func (s *polygonSyncStageSyncStore) Run(context.Context) error {
-	return nil
-}
-
 type polygonSyncStageHeimdallStore struct {
 	checkpoints                 *polygonSyncStageCheckpointStore
 	milestones                  *polygonSyncStageMilestoneStore
@@ -1133,6 +1133,45 @@ func (s polygonSyncStageBridgeStore) LastProcessedEventID(ctx context.Context) (
 	return r.id, nil
 }
 
+func (s polygonSyncStageBridgeStore) LastProcessedBlockInfo(ctx context.Context) (bridge.ProcessedBlockInfo, bool, error) {
+	type response struct {
+		info bridge.ProcessedBlockInfo
+		ok   bool
+		err  error
+	}
+
+	r, err := awaitTxAction(ctx, s.txActionStream, func(tx kv.RwTx, responseStream chan<- response) error {
+		info, ok, err := bridge.LastProcessedBlockInfo(tx)
+		responseStream <- response{info: info, ok: ok, err: err}
+		return nil
+	})
+	if err != nil {
+		return bridge.ProcessedBlockInfo{}, false, err
+	}
+
+	return r.info, r.ok, r.err
+}
+
+func (s polygonSyncStageBridgeStore) PutProcessedBlockInfo(ctx context.Context, info bridge.ProcessedBlockInfo) error {
+	type response struct {
+		err error
+	}
+
+	r, err := awaitTxAction(ctx, s.txActionStream, func(tx kv.RwTx, responseStream chan<- response) error {
+		responseStream <- response{err: bridge.PutProcessedBlockInfo(tx, info)}
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	return r.err
+}
+
+func (s polygonSyncStageBridgeStore) LastFrozenEventBlockNum() uint64 {
+	return s.eventReader.LastFrozenEventBlockNum()
+}
+
 func (s polygonSyncStageBridgeStore) LastEventIDWithinWindow(ctx context.Context, fromID uint64, toTime time.Time) (uint64, error) {
 	type response struct {
 		id  uint64
@@ -1154,13 +1193,13 @@ func (s polygonSyncStageBridgeStore) LastEventIDWithinWindow(ctx context.Context
 	return r.id, nil
 }
 
-func (s polygonSyncStageBridgeStore) PutEventIDs(ctx context.Context, eventMap map[uint64]uint64) error {
+func (s polygonSyncStageBridgeStore) PutBlockNumToEventID(ctx context.Context, blockNumToEventId map[uint64]uint64) error {
 	type response struct {
 		err error
 	}
 
 	r, err := awaitTxAction(ctx, s.txActionStream, func(tx kv.RwTx, responseStream chan<- response) error {
-		responseStream <- response{err: bridge.PutEventsIDs(tx, eventMap)}
+		responseStream <- response{err: bridge.PutBlockNumToEventID(tx, blockNumToEventId)}
 		return nil
 	})
 	if err != nil {
@@ -1178,12 +1217,12 @@ func (s polygonSyncStageBridgeStore) Events(context.Context, uint64, uint64) ([]
 	panic("polygonSyncStageBridgeStore.Events not supported")
 }
 
-func (s polygonSyncStageBridgeStore) EventIDRange(context.Context, uint64) (uint64, uint64, error) {
+func (s polygonSyncStageBridgeStore) BlockEventIDsRange(context.Context, uint64) (uint64, uint64, error) {
 	// used for accessing events in execution
 	// astrid stage integration intends to use the bridge only for scrapping
 	// not for reading which remains the same in execution (via BlockReader)
 	// astrid standalone mode introduces its own reader
-	panic("polygonSyncStageBridgeStore.EventIDRange not supported")
+	panic("polygonSyncStageBridgeStore.BlockEventIDsRange not supported")
 }
 
 func (s polygonSyncStageBridgeStore) EventTxnToBlockNum(context.Context, common.Hash) (uint64, bool, error) {
@@ -1225,6 +1264,24 @@ type polygonSyncStageExecutionEngine struct {
 	cachedForkChoice *types.Header
 }
 
+func (e *polygonSyncStageExecutionEngine) GetHeader(ctx context.Context, blockNum uint64) (*types.Header, error) {
+	type response struct {
+		header *types.Header
+		err    error
+	}
+
+	r, err := awaitTxAction(ctx, e.txActionStream, func(tx kv.RwTx, responseStream chan<- response) error {
+		header, err := e.blockReader.HeaderByNumber(ctx, tx, blockNum)
+		responseStream <- response{header: header, err: err}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return r.header, r.err
+}
+
 func (e *polygonSyncStageExecutionEngine) InsertBlocks(ctx context.Context, blocks []*types.Block) error {
 	type response struct {
 		err error

polygon/bor/bordb/prune.go

@@ -30,6 +30,7 @@ import (
 // doesn't change sequences of kv.EthTx
 // doesn't delete Receipts, Senders, Canonical markers, TotalDifficulty
 func PruneBorBlocks(tx kv.RwTx, blockTo uint64, blocksDeleteLimit int, SpanIdAt func(number uint64) uint64) (deleted int, err error) {
+	// events
 	c, err := tx.Cursor(kv.BorEventNums)
 	if err != nil {
 		return deleted, err
@@ -70,14 +71,40 @@ func PruneBorBlocks(tx kv.RwTx, blockTo uint64, blocksDeleteLimit int, SpanIdAt
 	if err != nil {
 		return deleted, err
 	}
+
+	epbCursor, err := tx.RwCursor(kv.BorEventProcessedBlocks)
+	if err != nil {
+		return deleted, err
+	}
+
+	defer epbCursor.Close()
+	counter = blocksDeleteLimit
+	for k, _, err = epbCursor.First(); err == nil && k != nil && counter > 0; k, _, err = epbCursor.Next() {
+		blockNum := binary.BigEndian.Uint64(k)
+		if blockNum >= blockTo {
+			break
+		}
+
+		if err = epbCursor.DeleteCurrent(); err != nil {
+			return deleted, err
+		}
+
+		deleted++
+		counter--
+	}
+	if err != nil {
+		return deleted, err
+	}
+
+	// spans
 	firstSpanToKeep := SpanIdAt(blockTo)
 	c2, err := tx.RwCursor(kv.BorSpans)
 	if err != nil {
 		return deleted, err
 	}
 	defer c2.Close()
 	counter = blocksDeleteLimit
-	for k, _, err := c2.First(); err == nil && k != nil && counter > 0; k, _, err = c2.Next() {
+	for k, _, err = c2.First(); err == nil && k != nil && counter > 0; k, _, err = c2.Next() {
 		spanId := binary.BigEndian.Uint64(k)
 		if spanId >= firstSpanToKeep {
 			break
@@ -88,6 +115,9 @@ func PruneBorBlocks(tx kv.RwTx, blockTo uint64, blocksDeleteLimit int, SpanIdAt
 		deleted++
 		counter--
 	}
+	if err != nil {
+		return deleted, err
+	}
 
 	if snaptype.CheckpointsEnabled() {
 		checkpointCursor, err := tx.RwCursor(kv.BorCheckpoints)