Merge pull request #728 from onflow/mpeter/poc-index-finalized-block-results

Implement the `RPCBlockHeaderSubscriber` for indexing finalized results

Author: m-Peter

Makefile

@@ -80,10 +80,8 @@ e2e-test:
 
 .PHONY: check-tidy
 check-tidy:
-	go mod tidy
-	git diff --exit-code
-	cd tests
-	go mod tidy
+	go mod tidy -v
+	cd tests; go mod tidy -v
 	git diff --exit-code
 
 .PHONY: build

api/api.go

@@ -206,7 +206,7 @@ func (b *BlockChainAPI) GetBalance(
 		return nil, err
 	}
 
-	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks, b.logger)
+	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks)
 	if err != nil {
 		return handleError[*hexutil.Big](err, l, b.collector)
 	}
@@ -436,7 +436,7 @@ func (b *BlockChainAPI) GetBlockReceipts(
 		return nil, err
 	}
 
-	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks, b.logger)
+	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks)
 	if err != nil {
 		return handleError[[]map[string]any](err, l, b.collector)
 	}
@@ -567,7 +567,7 @@ func (b *BlockChainAPI) Call(
 		return handleError[hexutil.Bytes](err, l, b.collector)
 	}
 
-	height, err := resolveBlockTag(blockNumberOrHash, b.blocks, b.logger)
+	height, err := resolveBlockTag(blockNumberOrHash, b.blocks)
 	if err != nil {
 		return handleError[hexutil.Bytes](err, l, b.collector)
 	}
@@ -689,7 +689,7 @@ func (b *BlockChainAPI) GetTransactionCount(
 		return nil, err
 	}
 
-	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks, b.logger)
+	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks)
 	if err != nil {
 		return handleError[*hexutil.Uint64](err, l, b.collector)
 	}
@@ -756,7 +756,7 @@ func (b *BlockChainAPI) EstimateGas(
 		from = *args.From
 	}
 
-	height, err := resolveBlockTag(blockNumberOrHash, b.blocks, b.logger)
+	height, err := resolveBlockTag(blockNumberOrHash, b.blocks)
 	if err != nil {
 		return handleError[hexutil.Uint64](err, l, b.collector)
 	}
@@ -791,7 +791,7 @@ func (b *BlockChainAPI) GetCode(
 		return nil, err
 	}
 
-	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks, b.logger)
+	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks)
 	if err != nil {
 		return handleError[hexutil.Bytes](err, l, b.collector)
 	}
@@ -912,7 +912,7 @@ func (b *BlockChainAPI) GetStorageAt(
 		)
 	}
 
-	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks, b.logger)
+	height, err := resolveBlockTag(&blockNumberOrHash, b.blocks)
 	if err != nil {
 		return handleError[hexutil.Bytes](err, l, b.collector)
 	}

api/debug.go

@@ -154,7 +154,7 @@ func (d *DebugAPI) TraceCall(
 		return nil, err
 	}
 
-	height, err := resolveBlockTag(&blockNrOrHash, d.blocks, d.logger)
+	height, err := resolveBlockTag(&blockNrOrHash, d.blocks)
 	if err != nil {
 		return nil, err
 	}
@@ -258,7 +258,7 @@ func (d *DebugAPI) FlowHeightByBlock(
 		return 0, err
 	}
 
-	height, err := resolveBlockTag(&blockNrOrHash, d.blocks, d.logger)
+	height, err := resolveBlockTag(&blockNrOrHash, d.blocks)
 	if err != nil {
 		return 0, err
 	}

api/utils.go

@@ -18,7 +18,6 @@ import (
 func resolveBlockTag(
 	blockNumberOrHash *rpc.BlockNumberOrHash,
 	blocksDB storage.BlockIndexer,
-	logger zerolog.Logger,
 ) (uint64, error) {
 	if blockNumberOrHash == nil {
 		return 0, fmt.Errorf(
@@ -29,9 +28,6 @@ func resolveBlockTag(
 	if number, ok := blockNumberOrHash.Number(); ok {
 		height, err := resolveBlockNumber(number, blocksDB)
 		if err != nil {
-			logger.Error().Err(err).
-				Stringer("block_number", number).
-				Msg("failed to resolve block by number")
 			return 0, err
 		}
 		return height, nil
@@ -40,9 +36,6 @@ func resolveBlockTag(
 	if hash, ok := blockNumberOrHash.Hash(); ok {
 		height, err := blocksDB.GetHeightByID(hash)
 		if err != nil {
-			logger.Error().Err(err).
-				Stringer("block_hash", hash).
-				Msg("failed to resolve block by hash")
 			return 0, err
 		}
 		return height, nil

bootstrap/bootstrap.go

@@ -57,6 +57,7 @@ type Storages struct {
 	Transactions storage.TransactionIndexer
 	Receipts     storage.ReceiptIndexer
 	Traces       storage.TraceIndexer
+	EventsHash   *pebble.EventsHash
 }
 
 type Publishers struct {
@@ -159,14 +160,37 @@ func (b *Bootstrap) StartEventIngestion(ctx context.Context) error {
 		nextCadenceHeight -= 1
 	}
 
-	// create EVM event subscriber
-	subscriber := ingestion.NewRPCEventSubscriber(
-		b.logger,
-		b.client,
-		chainID,
-		b.keystore,
-		nextCadenceHeight,
-	)
+	// create event subscriber
+	var subscriber ingestion.EventSubscriber
+	if b.config.ExperimentalSoftFinalityEnabled {
+		var verifier *ingestion.SealingVerifier
+		if b.config.ExperimentalSealingVerificationEnabled {
+			verifier = ingestion.NewSealingVerifier(
+				b.logger,
+				b.client,
+				chainID,
+				b.storages.EventsHash,
+				nextCadenceHeight,
+			)
+		}
+
+		subscriber = ingestion.NewRPCBlockTrackingSubscriber(
+			b.logger,
+			b.client,
+			chainID,
+			b.keystore,
+			nextCadenceHeight,
+			verifier,
+		)
+	} else {
+		subscriber = ingestion.NewRPCEventSubscriber(
+			b.logger,
+			b.client,
+			chainID,
+			b.keystore,
+			nextCadenceHeight,
+		)
+	}
 
 	blocksProvider := replayer.NewBlocksProvider(
 		b.storages.Blocks,
@@ -603,6 +627,7 @@ func setupStorage(
 	blocks := pebble.NewBlocks(store, config.FlowNetworkID)
 	storageAddress := evm.StorageAccountAddress(config.FlowNetworkID)
 	registerStore := pebble.NewRegisterStorage(store, storageAddress)
+	eventsHash := pebble.NewEventsHash(store)
 
 	batch := store.NewBatch()
 	defer func() {
@@ -617,7 +642,20 @@ func setupStorage(
 	if config.ForceStartCadenceHeight != 0 {
 		logger.Warn().Uint64("height", config.ForceStartCadenceHeight).Msg("force setting starting Cadence height!!!")
 		if err := blocks.SetLatestCadenceHeight(config.ForceStartCadenceHeight, batch); err != nil {
-			return nil, nil, err
+			return nil, nil, fmt.Errorf("failed to set latest cadence height: %w", err)
+		}
+
+		verifiedHeight, err := eventsHash.ProcessedSealedHeight()
+		if err != nil && !errors.Is(err, errs.ErrStorageNotInitialized) {
+			return nil, nil, fmt.Errorf("failed to get latest verified sealed height: %w", err)
+		}
+		if verifiedHeight > config.ForceStartCadenceHeight {
+			if err := eventsHash.BatchSetProcessedSealedHeight(config.ForceStartCadenceHeight, batch); err != nil {
+				return nil, nil, fmt.Errorf("failed to set latest verified sealed height: %w", err)
+			}
+			if err := eventsHash.BatchRemoveAboveHeight(config.ForceStartCadenceHeight, batch); err != nil {
+				return nil, nil, fmt.Errorf("failed to reset events hash above height: %w", err)
+			}
 		}
 	}
 
@@ -682,6 +720,7 @@ func setupStorage(
 		Transactions: pebble.NewTransactions(store),
 		Receipts:     pebble.NewReceipts(store),
 		Traces:       pebble.NewTraces(store),
+		EventsHash:   eventsHash,
 	}, nil
 }
 

cmd/run/cmd.go

@@ -254,6 +254,10 @@ func parseConfigFromFlags() error {
 		}
 	}
 
+	if !cfg.ExperimentalSoftFinalityEnabled && cfg.ExperimentalSealingVerificationEnabled {
+		return fmt.Errorf("experimental-sealing-verification-enabled should be enabled only when experimental-soft-finality-enabled=true")
+	}
+
 	return nil
 }
 
@@ -318,6 +322,8 @@ func init() {
 	Cmd.Flags().BoolVar(&cfg.TxBatchMode, "tx-batch-mode", false, "Enable batch transaction submission, to avoid nonce mismatch issues for high-volume EOAs.")
 	Cmd.Flags().DurationVar(&cfg.TxBatchInterval, "tx-batch-interval", time.Millisecond*1200, "Time interval upon which to submit the transaction batches to the Flow network.")
 	Cmd.Flags().DurationVar(&cfg.EOAActivityCacheTTL, "eoa-activity-cache-ttl", time.Second*10, "Time interval used to track EOA activity. Tx send more frequently than this interval will be batched. Useful only when batch transaction submission is enabled.")
+	Cmd.Flags().BoolVar(&cfg.ExperimentalSoftFinalityEnabled, "experimental-soft-finality-enabled", false, "Sets whether the gateway should use the experimental soft finality feature. This results in faster indexing time, because EVM state is fetched from finalized, instead of sealed Flow blocks.")
+	Cmd.Flags().BoolVar(&cfg.ExperimentalSealingVerificationEnabled, "experimental-sealing-verification-enabled", false, "Sets whether the gateway should use the experimental soft finality sealing verification feature. This is an extra safety check for --experimental-soft-finality-enabled=true, which verifies that all finalized Flow blocks that were indexed, have eventually been sealed. The ingestion will halt, even if a single Flow block was not found to be sealed.")
 	Cmd.Flags().BoolVar(&cfg.TxMemPoolMode, "tx-mempool-mode", false, "Enable the transaction mempool: expected-nonce transactions are submitted immediately, out-of-order transactions are held until their nonce gap fills. Mutually exclusive with --tx-batch-mode and requires --tx-state-validation=local-index.")
 	Cmd.Flags().DurationVar(&cfg.TxCollectionWindow, "tx-collection-window", 300*time.Millisecond, "Per-EOA sliding collection window for the transaction mempool. Resets on each arrival from the same EOA.")
 	Cmd.Flags().DurationVar(&cfg.TxSubmissionSpacing, "tx-submission-spacing", 1200*time.Millisecond, "Minimum gap between consecutive Cadence submissions for the same EOA in the transaction mempool; also serves as the flush deadline for a continuously-fed collection window. Recommended ~1.5x the block production rate.")

config/config.go

@@ -124,6 +124,15 @@ type Config struct {
 	// frequently than this interval will be batched.
 	// Useful only when batch transaction submission is enabled.
 	EOAActivityCacheTTL time.Duration
+	// ExperimentalSoftFinalityEnabled enables the experimental soft finality feature which syncs
+	// EVM block and transaction data from the upstream Access node before the block is sealed.
+	// CAUTION: This feature is experimental and may return incorrect data in certain circumstances.
+	ExperimentalSoftFinalityEnabled bool
+	// ExperimentalSealingVerificationEnabled enables the experimental sealing verification feature
+	// which verifies the hash of the EVM events ingested by the requester engine match the hash
+	// of the events from the sealed block in the Flow network.
+	// CAUTION: This feature is experimental and will cause the node to halt if the events don't match.
+	ExperimentalSealingVerificationEnabled bool
 	// TxMemPoolMode configures the gateway to use the transaction mempool:
 	// transactions carrying the expected next nonce (with nothing in flight)
 	// are submitted immediately, out-of-order transactions are held until their

models/events.go

@@ -155,6 +155,11 @@ func decodeCadenceEvents(events flow.BlockEvents) (*CadenceEvents, error) {
 	return e, nil
 }
 
+// BlockEvents returns the Flow block events.
+func (c *CadenceEvents) BlockEvents() flow.BlockEvents {
+	return c.events
+}
+
 // Block evm block. If event doesn't contain EVM block the return value is nil.
 func (c *CadenceEvents) Block() *Block {
 	return c.block