Implement the RPCBlockHeaderSubscriber for indexing finalized results

Makefile

@@ -78,10 +78,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

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 {
@@ -153,14 +154,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,
+		)
+	}
 
 	callTracerCollector, err := replayer.NewCallTracerCollector(
 		b.config.EVMNetworkID,
@@ -580,6 +604,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() {
@@ -594,7 +619,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)
 		}
 	}
 
@@ -659,6 +697,7 @@ func setupStorage(
 		Transactions: pebble.NewTransactions(store),
 		Receipts:     pebble.NewReceipts(store),
 		Traces:       pebble.NewTraces(store),
+		EventsHash:   eventsHash,
 	}, nil
 }
 

cmd/run/cmd.go

@@ -227,6 +227,9 @@ func parseConfigFromFlags() error {
 		return fmt.Errorf("tx-batch-mode should be enabled with tx-state-validation=local-index")
 	}
 
+	cfg.ExperimentalSoftFinalityEnabled = experimentalSoftFinalityEnabled
+	cfg.ExperimentalSealingVerificationEnabled = experimentalSealingVerificationEnabled
+
 	return nil
 }
 
@@ -250,6 +253,9 @@ var (
 	txStateValidation string
 	initHeight,
 	forceStartHeight uint64
+
+	experimentalSoftFinalityEnabled,
+	experimentalSealingVerificationEnabled bool
 )
 
 func init() {
@@ -290,6 +296,8 @@ func init() {
 	Cmd.Flags().DurationVar(&cfg.TxRequestLimitDuration, "tx-request-limit-duration", time.Second*3, "Time interval upon which to enforce transaction submission rate limiting.")
 	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().BoolVar(&experimentalSoftFinalityEnabled, "experimental-soft-finality-enabled", false, "Sets whether the gateway should use the experimental soft finality feature. WARNING: This may result in incorrect results being returned in certain circumstances. Use only if you know what you are doing.")
+	Cmd.Flags().BoolVar(&experimentalSealingVerificationEnabled, "experimental-sealing-verification-enabled", true, "Sets whether the gateway should use the experimental soft finality sealing verification feature. WARNING: This may result in indexing halts if events do not match. Use only if you know what you are doing.")
 	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.")
 
 	err := Cmd.Flags().MarkDeprecated("init-cadence-height", "This flag is no longer necessary and will be removed in future version. The initial Cadence height is known for testnet/mainnet and this was only required for fresh deployments of EVM Gateway. Once the DB has been initialized, the latest index Cadence height will be used upon start-up.")

config/config.go

@@ -117,6 +117,15 @@ type Config struct {
 	// TxBatchInterval is the time interval upon which to submit the transaction batches to the
 	// Flow network.
 	TxBatchInterval 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
 	// EOAActivityCacheTTL is the 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.

models/events.go

@@ -152,6 +152,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