Use espresso streamer in batcher (EspressoSystems#708)

* Use espresso streamer in batcher

* Lint

* Move helper functions to a new file

* Fix msgCount

* Fix msg count

Author: ImJeremyHe

arbnode/batch_poster.go

@@ -0,0 +1,95 @@
+package arbnode
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/offchainlabs/nitro/arbutil"
+	"github.com/offchainlabs/nitro/espressotee"
+	"github.com/offchainlabs/nitro/solgen/go/espressogen"
+)
+
+// Reads state from external sources and resets the espresso streamer to start producing
+// messages from hotshot based on the source of truth on the parent chain
+func (b *BatchPoster) resetStreamerToParentChainOrConfigHotshotBlock(messageCount arbutil.MessageIndex, ctx context.Context) {
+	hotshotBlock := b.fetchHotshotBlockFromLastCheckpoint(ctx)
+	if hotshotBlock == 0 {
+		// if there hasn't been a batch posted, or we encountered an error, start reading from the configured hotshot block number.
+		hotshotBlock = b.config().HotShotBlock
+	}
+	b.espressoStreamer.Reset(uint64(messageCount), hotshotBlock)
+}
+
+// fetchHotshotBlockFromLastCheckpoint:
+// This function uses the sequencer inbox bridgegen contract to filter for logs related to the TEESignatureVerified events
+// If any of these events are encountered, it checks the log iterator for the data about this event.
+// Return:
+// returns the Hotshot height of the last event in the iterator returned from FilterTEESignatureVerified()
+// representing the most recently emitted hotshotblock height. Any errors encountered will result in 0 being returned.
+func (b *BatchPoster) fetchHotshotBlockFromLastCheckpoint(ctx context.Context) uint64 {
+	pollingStep := b.config().EspressoEventPollingStep
+	header, err := b.l1Reader.LastHeader(ctx)
+	if err != nil {
+		log.Error("Failed to fetch last header from parent chain", "err", err)
+		return 0
+	}
+
+	var lastHotshotHeight uint64 = 0
+	// Prevent unsigned integer underflow: in Go, subtracting a larger value
+	// from a smaller uint64 will wrap around to a very large number.
+	for i := header.Number.Uint64(); i >= b.config().HotShotFirstPostingBlock; i -= min(i, pollingStep) {
+		start := i - min(i, pollingStep)
+		if start < b.config().HotShotFirstPostingBlock {
+			start = b.config().HotShotFirstPostingBlock
+		}
+		filterOpts := bind.FilterOpts{
+			Start:   start,
+			End:     &i,
+			Context: ctx,
+		}
+
+		logIterator, err := b.seqInbox.FilterTEESignatureVerified(&filterOpts, []*big.Int{}, []*big.Int{})
+		if err != nil {
+			log.Error("Failed to obtain iterator for logs for block", "blockNumber", i, "err", err)
+			continue
+		}
+
+		if logIterator == nil {
+			continue
+		}
+
+		for logIterator.Next() {
+			lastHotshotHeight = logIterator.Event.HotshotHeight.Uint64()
+		}
+
+		if lastHotshotHeight > 0 {
+			return lastHotshotHeight
+		}
+	}
+
+	log.Warn("No logs found for Hotshot block")
+	return 0
+}
+
+func setupNitroVerifier(teeVerifier *espressogen.IEspressoTEEVerifier, l1Client *ethclient.Client) (espressotee.EspressoNitroTEEVerifierInterface, error) {
+	// Setup nitro contract interface
+	nitroAddr, err := teeVerifier.EspressoNitroTEEVerifier(&bind.CallOpts{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to get nitro tee verifier address from caller: %w", err)
+	}
+	log.Info("successfully retrieved nitro contract verifier address", "address", nitroAddr)
+
+	nitroVerifierBindings, err := espressogen.NewIEspressoNitroTEEVerifier(
+		nitroAddr,
+		l1Client)
+	if err != nil {
+		return nil, err
+	}
+	nitroVerifier := espressotee.NewEspressoNitroTEEVerifier(nitroVerifierBindings, l1Client, nitroAddr)
+	return nitroVerifier, nil
+}

arbnode/espresso_batch_poster_helpers.go

@@ -86,14 +86,14 @@ func NewBatcherAddrMonitor(
 	l1Reader *headerreader.HeaderReader,
 	seqInboxAddr common.Address,
 	deployAt uint64,
-	fromBlock uint64,
+	fromParentBlock uint64,
 ) *BatcherAddrMonitor {
 	seqInboxInterface, err := bridgegen.NewSequencerInbox(seqInboxAddr, l1Reader.Client())
 	if err != nil {
 		panic(err)
 	}
-	if fromBlock < deployAt+1 {
-		fromBlock = deployAt + 1
+	if fromParentBlock < deployAt+1 {
+		fromParentBlock = deployAt + 1
 	}
 	return &BatcherAddrMonitor{
 		initAddresses:             initAddresses,
@@ -102,7 +102,7 @@ func NewBatcherAddrMonitor(
 		seqInboxAddr:              seqInboxAddr,
 		seqInboxInterface:         seqInboxInterface,
 		deployAt:                  deployAt,
-		lastProcessedParentHeight: fromBlock - 1,
+		lastProcessedParentHeight: fromParentBlock - 1,
 	}
 }
 

arbnode/espresso_batcher_addr_monitor.go

@@ -200,7 +200,6 @@ func ConfigureEspressoFields(
 		TxnsResubmissionInterval:              DefaultBatchPosterConfig.EspressoTxnsResubmissionInterval,
 		MaxTransactionSize:                    DefaultBatchPosterConfig.EspressoTxSizeLimit,
 		ResubmitEspressoTxDeadline:            DefaultBatchPosterConfig.ResubmitEspressoTxDeadline,
-		UseEscapeHatch:                        DefaultBatchPosterConfig.UseEscapeHatch,
 
 		SubmitterCreator: submitter.NewPollingEspressoSubmitter,
 	}
@@ -220,10 +219,7 @@ func ConfigureEspressoFields(
 		submitter.WithTxnsSendingInterval(config.TxnsSendingInterval),
 		submitter.WithTxnsResubmissionInterval(config.TxnsResubmissionInterval),
 		submitter.WithResubmitEspressoTxDeadline(config.ResubmitEspressoTxDeadline),
-		submitter.WithUseEscapeHatch(config.UseEscapeHatch),
-		submitter.WithEscapeHatchEnabled(config.EscapeHatchEnabled),
 		submitter.WithMaxTransactionSize(config.MaxTransactionSize),
-		submitter.WithMaxBlockLagBeforeEscapeHatch(config.MaxBlockLagBeforeEscapeHatch),
 		submitter.WithInitialFinalizedSequencerMessageCount(config.InitialFinalizedSequencerMessageCount),
 		submitter.WithMultipleOptions(config.SubmitterConfiguration...),
 	)

arbnode/espresso_transaction_streamer_helpers.go

@@ -2,6 +2,11 @@ package arbnode
 
 import (
 	"context"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+
+	"github.com/offchainlabs/nitro/util/signature"
 )
 
 var (
@@ -34,3 +39,21 @@ func binarySearchForBlockNumber(
 	}
 	return start, nil
 }
+
+// We should be able to get the address as soon as we have the signer.
+// We don't want to change a lot of code to make this work since we are working on a forked repo.
+// This function is not costly and it should be called only once.
+func recoverAddressFromSigner(signer signature.DataSignerFunc) (common.Address, error) {
+	message := make([]byte, 32)
+	signature, err := signer(message)
+	if err != nil {
+		return common.Address{}, err
+	}
+
+	publicKey, err := crypto.SigToPub(message, signature)
+	if err != nil {
+		return common.Address{}, err
+	}
+
+	return crypto.PubkeyToAddress(*publicKey), nil
+}

arbnode/espresso_utils.go

@@ -0,0 +1,74 @@
+package arbnode
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/crypto"
+
+	"github.com/offchainlabs/nitro/util/signature"
+)
+
+func TestRecoverAddressFromSigner(t *testing.T) {
+	privateKey, err := crypto.GenerateKey()
+	if err != nil {
+		t.Fatal(err)
+	}
+	address, err := recoverAddressFromSigner(signature.DataSignerFromPrivateKey(privateKey))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if address != crypto.PubkeyToAddress(privateKey.PublicKey) {
+		t.Fatalf("expected address %v, got %v", crypto.PubkeyToAddress(privateKey.PublicKey), address)
+	}
+}
+
+func TestBinarySearchForBlockNumber(t *testing.T) {
+	target := uint64(64)
+	count := 0
+	ctx := context.Background()
+	start := uint64(0)
+	end := uint64(100)
+	f := func(ctx context.Context, blockNumber uint64) (int, error) {
+		count++
+		if blockNumber < target {
+			return -1, nil
+		} else if blockNumber > target {
+			return 1, nil
+		}
+		return 0, nil
+	}
+	result, err := binarySearchForBlockNumber(ctx, start, end, f)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result != target {
+		t.Fatalf("expected result %d, got %d", target, result)
+	}
+	if count > 7 {
+		t.Fatalf("expected count less than %d, got %d", 7, count)
+	}
+
+	targetRangeStart := uint64(60)
+	targetRangeEnd := uint64(70)
+	count = 0
+	f = func(ctx context.Context, blockNumber uint64) (int, error) {
+		count++
+		if blockNumber < targetRangeStart {
+			return -1, nil
+		} else if blockNumber > targetRangeEnd {
+			return 1, nil
+		}
+		return 0, nil
+	}
+	result, err = binarySearchForBlockNumber(ctx, start, end, f)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result != targetRangeStart {
+		t.Fatalf("expected result %d, got %d", targetRangeStart, result)
+	}
+	if count > 7 {
+		t.Fatalf("expected count less than %d, got %d", 7, count)
+	}
+}

arbnode/espresso_utils_test.go

@@ -916,6 +916,7 @@ func getBatchPoster(
 	dapReaders []daprovider.Reader,
 	stakerAddr common.Address,
 	dataSigner signature.DataSignerFunc,
+	l2ChainId uint64,
 ) (*BatchPoster, error) {
 	var batchPoster *BatchPoster
 	if config.BatchPoster.Enable {
@@ -931,6 +932,7 @@ func getBatchPoster(
 		}
 		var err error
 		batchPoster, err = NewBatchPoster(ctx, &BatchPosterOpts{
+			ChainID:       l2ChainId,
 			DataPosterDB:  rawdb.NewTable(arbDb, storage.BatchPosterPrefix),
 			L1Reader:      l1Reader,
 			Inbox:         inboxTracker,
@@ -1201,7 +1203,7 @@ func createNodeImpl(
 		return nil, err
 	}
 
-	batchPoster, err := getBatchPoster(ctx, config, configFetcher, txOptsBatchPoster, dapWriter, l1Reader, inboxTracker, txStreamer, executionBatchPoster, arbDb, syncMonitor, deployInfo, parentChainID, dapReaders, stakerAddr, dataSigner)
+	batchPoster, err := getBatchPoster(ctx, config, configFetcher, txOptsBatchPoster, dapWriter, l1Reader, inboxTracker, txStreamer, executionBatchPoster, arbDb, syncMonitor, deployInfo, parentChainID, dapReaders, stakerAddr, dataSigner, l2Config.ChainID.Uint64())
 	if err != nil {
 		return nil, err
 	}