Track and index the latest Flow fees surge factor

Author: m-Peter

Makefile

@@ -99,8 +99,8 @@ generate:
 	mockery --dir=storage --name=BlockIndexer --output=storage/mocks
 	mockery --dir=storage --name=ReceiptIndexer --output=storage/mocks
 	mockery --dir=storage --name=TransactionIndexer --output=storage/mocks
-	mockery --dir=storage --name=AccountIndexer --output=storage/mocks
 	mockery --dir=storage --name=TraceIndexer --output=storage/mocks
+	mockery --dir=storage --name=FeeParametersIndexer --output=storage/mocks
 	mockery --all --dir=services/traces --output=services/traces/mocks
 	mockery --all --dir=services/ingestion --output=services/ingestion/mocks
 	mockery --dir=models --name=Engine --output=models/mocks

bootstrap/bootstrap.go

@@ -51,12 +51,13 @@ const (
 )
 
 type Storages struct {
-	Storage      *pebble.Storage
-	Registers    *pebble.RegisterStorage
-	Blocks       storage.BlockIndexer
-	Transactions storage.TransactionIndexer
-	Receipts     storage.ReceiptIndexer
-	Traces       storage.TraceIndexer
+	Storage       *pebble.Storage
+	Registers     *pebble.RegisterStorage
+	Blocks        storage.BlockIndexer
+	Transactions  storage.TransactionIndexer
+	Receipts      storage.ReceiptIndexer
+	Traces        storage.TraceIndexer
+	FeeParameters storage.FeeParametersIndexer
 }
 
 type Publishers struct {
@@ -191,6 +192,7 @@ func (b *Bootstrap) StartEventIngestion(ctx context.Context) error {
 		b.storages.Receipts,
 		b.storages.Transactions,
 		b.storages.Traces,
+		b.storages.FeeParameters,
 		b.publishers.Block,
 		b.publishers.Logs,
 		b.logger,
@@ -645,6 +647,13 @@ func setupStorage(
 	//	// TODO(JanezP): verify storage account owner is correct
 	// }
 
+	feeParameters := pebble.NewFeeParameters(store)
+	if _, err = feeParameters.Get(); errors.Is(err, errs.ErrEntityNotFound) {
+		if err := feeParameters.Store(models.DefaultFeeParameters, batch); err != nil {
+			return nil, nil, fmt.Errorf("failed to bootstrap fee parameters: %w", err)
+		}
+	}
+
 	if batch.Count() > 0 {
 		err = batch.Commit(pebbleDB.Sync)
 		if err != nil {
@@ -653,12 +662,13 @@ func setupStorage(
 	}
 
 	return db, &Storages{
-		Storage:      store,
-		Blocks:       blocks,
-		Registers:    registerStore,
-		Transactions: pebble.NewTransactions(store),
-		Receipts:     pebble.NewReceipts(store),
-		Traces:       pebble.NewTraces(store),
+		Storage:       store,
+		Blocks:        blocks,
+		Registers:     registerStore,
+		Transactions:  pebble.NewTransactions(store),
+		Receipts:      pebble.NewReceipts(store),
+		Traces:        pebble.NewTraces(store),
+		FeeParameters: feeParameters,
 	}, nil
 }
 

models/events.go

@@ -13,8 +13,9 @@ import (
 )
 
 const (
-	BlockExecutedQualifiedIdentifier       = string(events.EventTypeBlockExecuted)
-	TransactionExecutedQualifiedIdentifier = string(events.EventTypeTransactionExecuted)
+	BlockExecutedQualifiedIdentifier        = string(events.EventTypeBlockExecuted)
+	TransactionExecutedQualifiedIdentifier  = string(events.EventTypeTransactionExecuted)
+	FeeParametersChangedQualifiedIdentifier = "FlowFees.FeeParametersChanged"
 )
 
 // isBlockExecutedEvent checks whether the given event contains block executed data.
@@ -33,6 +34,15 @@ func isTransactionExecutedEvent(event cadence.Event) bool {
 	return event.EventType.QualifiedIdentifier == TransactionExecutedQualifiedIdentifier
 }
 
+// isFeeParametersChangedEvent checks whether the given event contains updates
+// to Flow fees parameters.
+func isFeeParametersChangedEvent(event cadence.Event) bool {
+	if event.EventType == nil {
+		return false
+	}
+	return event.EventType.QualifiedIdentifier == FeeParametersChangedQualifiedIdentifier
+}
+
 // CadenceEvents contains Flow emitted events containing one or zero evm block executed event,
 // and multiple or zero evm transaction events.
 type CadenceEvents struct {
@@ -42,6 +52,7 @@ 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.
@@ -124,6 +135,15 @@ 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
@@ -182,6 +202,11 @@ 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 {

models/fee_parameters.go

@@ -0,0 +1,46 @@
+package models
+
+import (
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/rlp"
+	"github.com/onflow/cadence"
+)
+
+var DefaultFeeParameters = &FeeParameters{
+	SurgeFactor:         cadence.UFix64(100_000_000),
+	InclusionEffortCost: cadence.UFix64(100_000_000),
+	ExecutionEffortCost: cadence.UFix64(100_000_000),
+}
+
+type FeeParameters struct {
+	SurgeFactor         cadence.UFix64 `cadence:"surgeFactor"`
+	InclusionEffortCost cadence.UFix64 `cadence:"inclusionEffortCost"`
+	ExecutionEffortCost cadence.UFix64 `cadence:"executionEffortCost"`
+}
+
+func (f *FeeParameters) ToBytes() ([]byte, error) {
+	return rlp.EncodeToBytes(f)
+}
+
+func NewFeeParametersFromBytes(data []byte) (*FeeParameters, error) {
+	feeParameters := &FeeParameters{}
+	if err := rlp.DecodeBytes(data, feeParameters); err != nil {
+		return nil, err
+	}
+
+	return feeParameters, nil
+}
+
+func decodeFeeParametersChangedEvent(event cadence.Event) (*FeeParameters, error) {
+	feeParameters := &FeeParameters{}
+	if err := cadence.DecodeFields(event, feeParameters); err != nil {
+		return nil, fmt.Errorf(
+			"failed to Cadence-decode FlowFees.FeeParametersChanged event [%s]: %w",
+			event.String(),
+			err,
+		)
+	}
+
+	return feeParameters, nil
+}

services/ingestion/engine.go

@@ -48,6 +48,7 @@ type Engine struct {
 	receipts        storage.ReceiptIndexer
 	transactions    storage.TransactionIndexer
 	traces          storage.TraceIndexer
+	feeParameters   storage.FeeParametersIndexer
 	log             zerolog.Logger
 	evmLastHeight   *models.SequentialHeight
 	blocksPublisher *models.Publisher[*models.Block]
@@ -65,6 +66,7 @@ func NewEventIngestionEngine(
 	receipts storage.ReceiptIndexer,
 	transactions storage.TransactionIndexer,
 	traces storage.TraceIndexer,
+	feeParameters storage.FeeParametersIndexer,
 	blocksPublisher *models.Publisher[*models.Block],
 	logsPublisher *models.Publisher[[]*gethTypes.Log],
 	log zerolog.Logger,
@@ -84,6 +86,7 @@ func NewEventIngestionEngine(
 		receipts:        receipts,
 		transactions:    transactions,
 		traces:          traces,
+		feeParameters:   feeParameters,
 		log:             log,
 		blocksPublisher: blocksPublisher,
 		logsPublisher:   logsPublisher,
@@ -217,6 +220,16 @@ 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

@@ -54,6 +54,7 @@ func TestSerialBlockIngestion(t *testing.T) {
 			Once() // make sure this isn't called multiple times
 
 		traces := &storageMock.TraceIndexer{}
+		feeParams := &storageMock.FeeParametersIndexer{}
 
 		eventsChan := make(chan models.BlockEvents)
 
@@ -73,6 +74,7 @@ func TestSerialBlockIngestion(t *testing.T) {
 			receipts,
 			transactions,
 			traces,
+			feeParams,
 			models.NewPublisher[*models.Block](),
 			models.NewPublisher[[]*gethTypes.Log](),
 			zerolog.Nop(),
@@ -134,6 +136,7 @@ func TestSerialBlockIngestion(t *testing.T) {
 			Once() // make sure this isn't called multiple times
 
 		traces := &storageMock.TraceIndexer{}
+		feeParams := &storageMock.FeeParametersIndexer{}
 
 		eventsChan := make(chan models.BlockEvents)
 		subscriber := &mocks.EventSubscriber{}
@@ -152,6 +155,7 @@ func TestSerialBlockIngestion(t *testing.T) {
 			receipts,
 			transactions,
 			traces,
+			feeParams,
 			models.NewPublisher[*models.Block](),
 			models.NewPublisher[[]*gethTypes.Log](),
 			zerolog.Nop(),
@@ -264,6 +268,8 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 				return nil
 			})
 
+		feeParams := &storageMock.FeeParametersIndexer{}
+
 		engine := NewEventIngestionEngine(
 			subscriber,
 			replayer.NewBlocksProvider(blocks, flowGo.Emulator, nil),
@@ -273,6 +279,7 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 			receipts,
 			transactions,
 			traces,
+			feeParams,
 			models.NewPublisher[*models.Block](),
 			models.NewPublisher[[]*gethTypes.Log](),
 			zerolog.Nop(),
@@ -372,6 +379,8 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 				return nil
 			})
 
+		feeParams := &storageMock.FeeParametersIndexer{}
+
 		engine := NewEventIngestionEngine(
 			subscriber,
 			replayer.NewBlocksProvider(blocks, flowGo.Emulator, nil),
@@ -381,6 +390,7 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 			receipts,
 			transactions,
 			traces,
+			feeParams,
 			models.NewPublisher[*models.Block](),
 			models.NewPublisher[[]*gethTypes.Log](),
 			zerolog.Nop(),
@@ -456,6 +466,7 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 			Once() // make sure this isn't called multiple times
 
 		traces := &storageMock.TraceIndexer{}
+		feeParams := &storageMock.FeeParametersIndexer{}
 
 		eventsChan := make(chan models.BlockEvents)
 		subscriber := &mocks.EventSubscriber{}
@@ -475,6 +486,7 @@ func TestBlockAndTransactionIngestion(t *testing.T) {
 			receipts,
 			transactions,
 			traces,
+			feeParams,
 			models.NewPublisher[*models.Block](),
 			models.NewPublisher[[]*gethTypes.Log](),
 			zerolog.Nop(),

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,
-			blocksFilter(r.chain),
+			blocksEventFilter(r.chain),
 			access.WithHeartbeatInterval(1),
 		)
 
@@ -471,7 +471,7 @@ func (r *RPCEventSubscriber) fetchMissingData(
 	// remove existing events
 	blockEvents.Events = nil
 
-	for _, eventType := range blocksFilter(r.chain).EventTypes {
+	for _, eventType := range evmEventFilter(r.chain).EventTypes {
 		recoveredEvents, err := r.client.GetEventsForHeightRange(
 			ctx,
 			eventType,
@@ -552,11 +552,37 @@ func (r *RPCEventSubscriber) recover(
 	return models.NewBlockEventsError(err)
 }
 
-// blockFilter define events we subscribe to:
-// A.{evm}.EVM.BlockExecuted and A.{evm}.EVM.TransactionExecuted,
-// where {evm} is EVM deployed contract address, which depends on the chain ID we configure.
-func blocksFilter(chainId flowGo.ChainID) flow.EventFilter {
-	evmAddress := common.Address(systemcontracts.SystemContractsForChain(chainId).EVMContract.Address)
+// 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,
+// where {evm} is the EVM deployed contract address, which depends on the
+// configured chain ID.
+func evmEventFilter(chainID flowGo.ChainID) flow.EventFilter {
+	contracts := systemcontracts.SystemContractsForChain(chainID)
+	evmAddress := common.Address(contracts.EVMContract.Address)
 
 	blockExecutedEvent := common.NewAddressLocation(
 		nil,

storage/index.go

@@ -102,3 +102,9 @@ type TraceIndexer interface {
 	// GetTransaction will retrieve transaction trace by the transaction ID.
 	GetTransaction(ID common.Hash) (json.RawMessage, error)
 }
+
+type FeeParametersIndexer interface {
+	Store(feeParameters *models.FeeParameters, batch *pebble.Batch) error
+
+	Get() (*models.FeeParameters, error)
+}