Track Flow fees params updates in its own subscriber service

Author: m-Peter

bootstrap/bootstrap.go

@@ -182,9 +182,17 @@ func (b *Bootstrap) StartEventIngestion(ctx context.Context) error {
 		ValidateResults:     true,
 	}
 
+	feeParamsSubscriber := ingestion.NewFeeParamsEventSubscriber(
+		b.logger,
+		b.client,
+		chainID,
+		nextCadenceHeight,
+	)
+
 	// initialize event ingestion engine
 	b.events = ingestion.NewEventIngestionEngine(
 		subscriber,
+		feeParamsSubscriber,
 		blocksProvider,
 		b.storages.Storage,
 		b.storages.Registers,

models/events.go

@@ -52,7 +52,6 @@ type CadenceEvents struct {
 	transactions      []Transaction                    // transactions in the EVM block
 	txEventPayloads   []events.TransactionEventPayload // EVM.TransactionExecuted event payloads
 	receipts          []*Receipt                       // receipts for transactions
-	feeParameters     *FeeParameters                   // updates to Flow fees parameters
 }
 
 // NewCadenceEvents decodes the events into evm types.
@@ -135,15 +134,6 @@ func decodeCadenceEvents(events flow.BlockEvents) (*CadenceEvents, error) {
 			e.txEventPayloads = append(e.txEventPayloads, *txEventPayload)
 			e.receipts = append(e.receipts, receipt)
 		}
-
-		if isFeeParametersChangedEvent(val) {
-			feeParameters, err := decodeFeeParametersChangedEvent(val)
-			if err != nil {
-				return nil, err
-			}
-
-			e.feeParameters = feeParameters
-		}
 	}
 
 	// safety check, we have a missing block in the events
@@ -202,11 +192,6 @@ func (c *CadenceEvents) Receipts() []*Receipt {
 	return c.receipts
 }
 
-// FeeParameters returns any updates to the Flow fees parameters.
-func (c *CadenceEvents) FeeParameters() *FeeParameters {
-	return c.feeParameters
-}
-
 // Empty checks if there is an EVM block included in the events.
 // If there are no evm block or transactions events this is a heartbeat event.
 func (c *CadenceEvents) Empty() bool {
@@ -279,3 +264,34 @@ func NewBlockEventsError(err error) BlockEvents {
 		Err: err,
 	}
 }
+
+type FeeParamsEvents struct {
+	FeeParameters *FeeParameters // updates to Flow fees parameters
+	Err           error
+}
+
+func NewFeeParamsEvents(events flow.BlockEvents) *FeeParamsEvents {
+	for _, event := range events.Events {
+		val := event.Value
+		if isFeeParametersChangedEvent(val) {
+			feeParameters, err := decodeFeeParametersChangedEvent(val)
+			return &FeeParamsEvents{
+				FeeParameters: feeParameters,
+				Err:           err,
+			}
+		}
+	}
+
+	return &FeeParamsEvents{
+		Err: fmt.Errorf(
+			"could not find any %s events",
+			FeeParametersChangedQualifiedIdentifier,
+		),
+	}
+}
+
+func NewFeeParamsEventsError(err error) *FeeParamsEvents {
+	return &FeeParamsEvents{
+		Err: err,
+	}
+}

services/ingestion/engine.go

@@ -40,25 +40,27 @@ var _ models.Engine = &Engine{}
 type Engine struct {
 	*models.EngineStatus
 
-	subscriber      EventSubscriber
-	blocksProvider  *replayer.BlocksProvider
-	store           *pebble.Storage
-	registerStore   *pebble.RegisterStorage
-	blocks          storage.BlockIndexer
-	receipts        storage.ReceiptIndexer
-	transactions    storage.TransactionIndexer
-	traces          storage.TraceIndexer
-	feeParameters   storage.FeeParametersIndexer
-	log             zerolog.Logger
-	evmLastHeight   *models.SequentialHeight
-	blocksPublisher *models.Publisher[*models.Block]
-	logsPublisher   *models.Publisher[[]*gethTypes.Log]
-	collector       metrics.Collector
-	replayerConfig  replayer.Config
+	subscriber          EventSubscriber
+	feeParamsSubscriber FeeParamsSubscriber
+	blocksProvider      *replayer.BlocksProvider
+	store               *pebble.Storage
+	registerStore       *pebble.RegisterStorage
+	blocks              storage.BlockIndexer
+	receipts            storage.ReceiptIndexer
+	transactions        storage.TransactionIndexer
+	traces              storage.TraceIndexer
+	feeParameters       storage.FeeParametersIndexer
+	log                 zerolog.Logger
+	evmLastHeight       *models.SequentialHeight
+	blocksPublisher     *models.Publisher[*models.Block]
+	logsPublisher       *models.Publisher[[]*gethTypes.Log]
+	collector           metrics.Collector
+	replayerConfig      replayer.Config
 }
 
 func NewEventIngestionEngine(
 	subscriber EventSubscriber,
+	feeParamsSubscriber FeeParamsSubscriber,
 	blocksProvider *replayer.BlocksProvider,
 	store *pebble.Storage,
 	registerStore *pebble.RegisterStorage,
@@ -78,20 +80,21 @@ func NewEventIngestionEngine(
 	return &Engine{
 		EngineStatus: models.NewEngineStatus(),
 
-		subscriber:      subscriber,
-		blocksProvider:  blocksProvider,
-		store:           store,
-		registerStore:   registerStore,
-		blocks:          blocks,
-		receipts:        receipts,
-		transactions:    transactions,
-		traces:          traces,
-		feeParameters:   feeParameters,
-		log:             log,
-		blocksPublisher: blocksPublisher,
-		logsPublisher:   logsPublisher,
-		collector:       collector,
-		replayerConfig:  replayerConfig,
+		subscriber:          subscriber,
+		feeParamsSubscriber: feeParamsSubscriber,
+		blocksProvider:      blocksProvider,
+		store:               store,
+		registerStore:       registerStore,
+		blocks:              blocks,
+		receipts:            receipts,
+		transactions:        transactions,
+		traces:              traces,
+		feeParameters:       feeParameters,
+		log:                 log,
+		blocksPublisher:     blocksPublisher,
+		logsPublisher:       logsPublisher,
+		collector:           collector,
+		replayerConfig:      replayerConfig,
 	}
 }
 
@@ -123,6 +126,7 @@ func (e *Engine) Run(ctx context.Context) error {
 	defer e.MarkStopped()
 
 	events := e.subscriber.Subscribe(ctx)
+	feeParamsEvents := e.feeParamsSubscriber.Subscribe(ctx)
 
 	for {
 		select {
@@ -145,6 +149,20 @@ func (e *Engine) Run(ctx context.Context) error {
 				e.log.Error().Err(err).Msg("failed to process EVM events")
 				return err
 			}
+		case feeParams, ok := <-feeParamsEvents:
+			if !ok {
+				return nil
+			}
+			if feeParams.Err != nil {
+				return fmt.Errorf(
+					"failure in FeeParametersChanged event subscription with: %w",
+					feeParams.Err,
+				)
+			}
+			if err := e.processFeeParamsEvents(feeParams); err != nil {
+				e.log.Error().Err(err).Msg("failed to process FeeParametersChanged events")
+				return err
+			}
 		}
 	}
 }
@@ -172,6 +190,25 @@ func (e *Engine) withBatch(f func(batch *pebbleDB.Batch) error) error {
 	return nil
 }
 
+func (e *Engine) processFeeParamsEvents(events *models.FeeParamsEvents) error {
+	if events == nil || events.FeeParameters == nil {
+		return nil
+	}
+
+	err := e.withBatch(
+		func(batch *pebbleDB.Batch) error {
+			return e.feeParameters.Store(events.FeeParameters, batch)
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("failed to update fee parameters during events ingestion: %w", err)
+	}
+
+	e.log.Info().Msg("updated fee parameters")
+
+	return nil
+}
+
 // processEvents converts the events to block and transactions and indexes them.
 //
 // BlockEvents are received by the access node API and contain Cadence height (always a single Flow block),
@@ -220,16 +257,6 @@ func (e *Engine) processEvents(events *models.CadenceEvents) error {
 
 // indexEvents will replay the evm transactions using the block events and index all results.
 func (e *Engine) indexEvents(events *models.CadenceEvents, batch *pebbleDB.Batch) error {
-	if events.FeeParameters() != nil {
-		if err := e.feeParameters.Store(events.FeeParameters(), batch); err != nil {
-			return fmt.Errorf(
-				"failed to update fee parameters for height: %d, during events ingestion: %w",
-				events.CadenceHeight(),
-				err,
-			)
-		}
-	}
-
 	// if heartbeat interval with no data still update the cadence height
 	if events.Empty() {
 		if err := e.blocks.SetLatestCadenceHeight(events.CadenceHeight(), batch); err != nil {

services/ingestion/engine_test.go

@@ -65,8 +65,15 @@ func TestSerialBlockIngestion(t *testing.T) {
 				return eventsChan
 			})
 
+		feeParamsSubscriber := &mocks.FeeParamsSubscriber{}
+		feeParamsSubscriber.On("Subscribe", mock.Anything).
+			Return(func(ctx context.Context) <-chan *models.FeeParamsEvents {
+				return make(chan *models.FeeParamsEvents)
+			})
+
 		engine := NewEventIngestionEngine(
 			subscriber,
+			feeParamsSubscriber,
 			replayer.NewBlocksProvider(blocks, flowGo.Emulator, nil),
 			store,
 			registerStore,
@@ -146,8 +153,15 @@ func TestSerialBlockIngestion(t *testing.T) {
 				return eventsChan
 			})
 
+		feeParamsSubscriber := &mocks.FeeParamsSubscriber{}
+		feeParamsSubscriber.On("Subscribe", mock.Anything).
+			Return(func(ctx context.Context) <-chan *models.FeeParamsEvents {
+				return make(chan *models.FeeParamsEvents)
+			})
+
 		engine := NewEventIngestionEngine(
 			subscriber,
+			feeParamsSubscriber,
 			replayer.NewBlocksProvider(blocks, flowGo.Emulator, nil),
 			store,
 			registerStore,
@@ -270,8 +284,15 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 
 		feeParams := &storageMock.FeeParametersIndexer{}
 
+		feeParamsSubscriber := &mocks.FeeParamsSubscriber{}
+		feeParamsSubscriber.On("Subscribe", mock.Anything).
+			Return(func(ctx context.Context) <-chan *models.FeeParamsEvents {
+				return make(chan *models.FeeParamsEvents)
+			})
+
 		engine := NewEventIngestionEngine(
 			subscriber,
+			feeParamsSubscriber,
 			replayer.NewBlocksProvider(blocks, flowGo.Emulator, nil),
 			store,
 			registerStore,
@@ -366,6 +387,12 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 				return eventsChan
 			})
 
+		feeParamsSubscriber := &mocks.FeeParamsSubscriber{}
+		feeParamsSubscriber.On("Subscribe", mock.Anything).
+			Return(func(ctx context.Context) <-chan *models.FeeParamsEvents {
+				return make(chan *models.FeeParamsEvents)
+			})
+
 		txCdc, txEvent, transaction, res, err := newTransaction(nextHeight)
 		require.NoError(t, err)
 		blockCdc, _, blockEvent, err := newBlock(nextHeight, []gethCommon.Hash{res.TxHash})
@@ -383,6 +410,7 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 
 		engine := NewEventIngestionEngine(
 			subscriber,
+			feeParamsSubscriber,
 			replayer.NewBlocksProvider(blocks, flowGo.Emulator, nil),
 			store,
 			registerStore,
@@ -477,8 +505,15 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 			}).
 			Once()
 
+		feeParamsSubscriber := &mocks.FeeParamsSubscriber{}
+		feeParamsSubscriber.On("Subscribe", mock.Anything).
+			Return(func(ctx context.Context) <-chan *models.FeeParamsEvents {
+				return make(chan *models.FeeParamsEvents)
+			})
+
 		engine := NewEventIngestionEngine(
 			subscriber,
+			feeParamsSubscriber,
 			replayer.NewBlocksProvider(blocks, flowGo.Emulator, nil),
 			store,
 			registerStore,

services/ingestion/event_subscriber.go

@@ -146,7 +146,7 @@ func (r *RPCEventSubscriber) subscribe(ctx context.Context, height uint64) <-cha
 		blockEventsStream, errChan, err = r.client.SubscribeEventsByBlockHeight(
 			ctx,
 			height,
-			blocksEventFilter(r.chain),
+			evmEventFilter(r.chain),
 			access.WithHeartbeatInterval(1),
 		)
 
@@ -552,29 +552,6 @@ func (r *RPCEventSubscriber) recover(
 	return models.NewBlockEventsError(err)
 }
 
-// blocksEventFilter defines the full set of events we subscribe to:
-// - A.{evm}.EVM.BlockExecuted
-// - A.{evm}.EVM.TransactionExecuted,
-// - A.{flow_fees}.FlowFees.FeeParametersChanged,
-// where {evm} is the EVM deployed contract address, which depends on the
-// configured chain ID and {flow_fees} is the FlowFees deployed contract
-// address for the configured chain ID.
-func blocksEventFilter(chainID flowGo.ChainID) flow.EventFilter {
-	contracts := systemcontracts.SystemContractsForChain(chainID)
-	flowFeesAddress := common.Address(contracts.FlowFees.Address)
-	eventFilter := evmEventFilter(chainID)
-
-	feeParametersChangedEvent := common.NewAddressLocation(
-		nil,
-		flowFeesAddress,
-		models.FeeParametersChangedQualifiedIdentifier,
-	).ID()
-
-	eventFilter.EventTypes = append(eventFilter.EventTypes, feeParametersChangedEvent)
-
-	return eventFilter
-}
-
 // evmEventFilter defines the EVM-related events we subscribe to:
 // - A.{evm}.EVM.BlockExecuted,
 // - A.{evm}.EVM.TransactionExecuted,