Implement new batch fetching scheme for automatic L1 gas pricing

.github/workflows/docker.yml

@@ -35,8 +35,8 @@ jobs:
         uses: actions/cache@v2
         with:
           path: /tmp/.buildx-cache
-          key: ${{ runner.os }}-buildx-a-${{ github.sha }}
-          restore-keys: ${{ runner.os }}-buildx-a-
+          key: ${{ runner.os }}-buildx-b-${{ github.sha }}
+          restore-keys: ${{ runner.os }}-buildx-b-
 
       - name: Build nitro-node docker
         uses: docker/build-push-action@v2

arbnode/api.go

@@ -40,10 +40,7 @@ func (a *BlockValidatorAPI) RevalidateBlock(ctx context.Context, blockNum rpc.Bl
 		}
 		moduleRoot = moduleRoots[0]
 	}
-	batchFetcher := func(batchSeqNum uint64, _ common.Hash) ([]byte, error) {
-		panic("not yet implemented") // BUGBUG
-	}
-	return a.val.ValidateBlock(ctx, header, moduleRoot, batchFetcher)
+	return a.val.ValidateBlock(ctx, header, moduleRoot)
 }
 
 func (a *BlockValidatorAPI) LatestValidatedBlock(ctx context.Context) (uint64, error) {

arbnode/inbox_reader.go

@@ -7,7 +7,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/offchainlabs/nitro/util/headerreader"
 	"math/big"
 	"strings"
 	"sync"
@@ -18,6 +17,7 @@ import (
 
 	"github.com/offchainlabs/nitro/arbutil"
 	"github.com/offchainlabs/nitro/util/arbmath"
+	"github.com/offchainlabs/nitro/util/headerreader"
 	"github.com/offchainlabs/nitro/util/stopwaiter"
 )
 

arbnode/node.go

@@ -7,17 +7,18 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/ethereum/go-ethereum/accounts"
-	"github.com/ethereum/go-ethereum/accounts/keystore"
-	"github.com/offchainlabs/nitro/cmd/genericconf"
-	"golang.org/x/term"
 	"math/big"
 	"os"
 	"path/filepath"
 	"strings"
 	"syscall"
 	"time"
 
+	"github.com/ethereum/go-ethereum/accounts"
+	"github.com/ethereum/go-ethereum/accounts/keystore"
+	"github.com/offchainlabs/nitro/cmd/genericconf"
+	"golang.org/x/term"
+
 	"github.com/offchainlabs/nitro/util/headerreader"
 
 	"github.com/ethereum/go-ethereum/rpc"
@@ -670,18 +671,7 @@ func createNodeImpl(
 	if err != nil {
 		return nil, err
 	}
-
-	arbos.BatchFetcher = func(batchNum uint64, batchDataHash common.Hash) (batchData []byte, err error) {
-		bnum := new(big.Int).SetUint64(batchNum)
-		batches, err := sequencerInbox.LookupBatchesInRange(context.Background(), bnum, bnum)
-		if err != nil {
-			return nil, err
-		}
-		if len(batches) <= 1 {
-			return nil, errors.New("expected sequencer batch not found")
-		}
-		return batches[0].GetData(ctx, l1client)
-	}
+	txStreamer.SetInboxReader(inboxReader)
 
 	nitroMachineConfig := validator.DefaultNitroMachineConfig
 	if config.Wasm.RootPath != "" {

arbnode/transaction_streamer.go

@@ -54,6 +54,7 @@ type TransactionStreamer struct {
 	coordinator     *SeqCoordinator
 	broadcastServer *broadcaster.Broadcaster
 	validator       *validator.BlockValidator
+	inboxReader     *InboxReader
 }
 
 func NewTransactionStreamer(db ethdb.Database, bc *core.BlockChain, broadcastServer *broadcaster.Broadcaster) (*TransactionStreamer, error) {
@@ -96,6 +97,16 @@ func (s *TransactionStreamer) SetSeqCoordinator(coordinator *SeqCoordinator) {
 	s.coordinator = coordinator
 }
 
+func (s *TransactionStreamer) SetInboxReader(inboxReader *InboxReader) {
+	if s.Started() {
+		panic("trying to set inbox reader after start")
+	}
+	if s.inboxReader != nil {
+		panic("trying to set inbox reader when already set")
+	}
+	s.inboxReader = inboxReader
+}
+
 func (s *TransactionStreamer) cleanupInconsistentState() error {
 	// If it doesn't exist yet, set the message count to 0
 	hasMessageCount, err := s.db.Has(messageCountKey)
@@ -696,6 +707,10 @@ func (s *TransactionStreamer) createBlocks(ctx context.Context) error {
 		}
 	}()
 
+	batchFetcher := func(batchNum uint64) ([]byte, error) {
+		return s.inboxReader.GetSequencerMessageBytes(ctx, batchNum)
+	}
+
 	for pos < msgCount {
 
 		statedb, err = s.bc.StateAt(lastBlockHeader.Root)
@@ -720,14 +735,18 @@ func (s *TransactionStreamer) createBlocks(ctx context.Context) error {
 			return err
 		}
 
-		block, receipts := arbos.ProduceBlock(
+		block, receipts, err := arbos.ProduceBlock(
 			msg.Message,
 			msg.DelayedMessagesRead,
 			lastBlockHeader,
 			statedb,
 			s.bc,
 			s.bc.Config(),
+			batchFetcher,
 		)
+		if err != nil {
+			return err
+		}
 
 		// ProduceBlock advances one message
 		pos++

arbos/block_processor.go

@@ -6,11 +6,12 @@ package arbos
 import (
 	"encoding/binary"
 	"fmt"
-	"github.com/offchainlabs/nitro/arbos/l1pricing"
 	"math"
 	"math/big"
 	"strconv"
 
+	"github.com/offchainlabs/nitro/arbos/l1pricing"
+
 	"github.com/offchainlabs/nitro/arbos/arbosState"
 	"github.com/offchainlabs/nitro/arbos/l2pricing"
 	"github.com/offchainlabs/nitro/arbos/util"
@@ -99,24 +100,39 @@ func noopSequencingHooks() *SequencingHooks {
 	}
 }
 
+type FallibleBatchFetcher func(batchNum uint64) ([]byte, error)
+
 func ProduceBlock(
 	message *L1IncomingMessage,
 	delayedMessagesRead uint64,
 	lastBlockHeader *types.Header,
 	statedb *state.StateDB,
 	chainContext core.ChainContext,
 	chainConfig *params.ChainConfig,
-) (*types.Block, types.Receipts) {
-	txes, err := message.ParseL2Transactions(chainConfig.ChainID)
+	batchFetcher FallibleBatchFetcher,
+) (*types.Block, types.Receipts, error) {
+	var batchFetchErr error
+	txes, err := message.ParseL2Transactions(chainConfig.ChainID, func(batchNum uint64) []byte {
+		data, err := batchFetcher(batchNum)
+		if err != nil {
+			batchFetchErr = err
+			return nil
+		}
+		return data
+	})
+	if batchFetchErr != nil {
+		return nil, nil, batchFetchErr
+	}
 	if err != nil {
 		log.Warn("error parsing incoming message", "err", err)
 		txes = types.Transactions{}
 	}
 
 	hooks := noopSequencingHooks()
-	return ProduceBlockAdvanced(
+	block, receipts := ProduceBlockAdvanced(
 		message.Header, txes, delayedMessagesRead, lastBlockHeader, statedb, chainContext, chainConfig, hooks,
 	)
+	return block, receipts, nil
 }
 
 // A bit more flexible than ProduceBlock for use in the sequencer.

arbos/incomingmessage.go

@@ -16,6 +16,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/params"
 
 	"github.com/offchainlabs/nitro/arbos/util"
 )
@@ -170,7 +171,9 @@ func ParseIncomingL1Message(rd io.Reader) (*L1IncomingMessage, error) {
 	}, nil
 }
 
-func (msg *L1IncomingMessage) ParseL2Transactions(chainId *big.Int) (types.Transactions, error) {
+type InfallibleBatchFetcher func(batchNum uint64) []byte
+
+func (msg *L1IncomingMessage) ParseL2Transactions(chainId *big.Int, batchFetcher InfallibleBatchFetcher) (types.Transactions, error) {
 	if len(msg.L2msg) > MaxL2MessageSize {
 		// ignore the message if l2msg is too large
 		return nil, errors.New("message too large")
@@ -222,7 +225,7 @@ func (msg *L1IncomingMessage) ParseL2Transactions(chainId *big.Int) (types.Trans
 		log.Debug("ignoring rollup event message")
 		return types.Transactions{}, nil
 	case L1MessageType_BatchPostingReport:
-		tx, err := parseBatchPostingReportMessage(bytes.NewReader(msg.L2msg), chainId)
+		tx, err := parseBatchPostingReportMessage(bytes.NewReader(msg.L2msg), chainId, batchFetcher)
 		if err != nil {
 			return nil, err
 		}
@@ -503,7 +506,7 @@ func parseSubmitRetryableMessage(rd io.Reader, header *L1IncomingMessageHeader,
 	return types.NewTx(tx), err
 }
 
-func parseBatchPostingReportMessage(rd io.Reader, chainId *big.Int) (*types.Transaction, error) {
+func parseBatchPostingReportMessage(rd io.Reader, chainId *big.Int, batchFetcher InfallibleBatchFetcher) (*types.Transaction, error) {
 	batchTimestamp, err := util.HashFromReader(rd)
 	if err != nil {
 		return nil, err
@@ -512,16 +515,26 @@ func parseBatchPostingReportMessage(rd io.Reader, chainId *big.Int) (*types.Tran
 	if err != nil {
 		return nil, err
 	}
-	batchNum, err := util.HashFromReader(rd)
+	batchNumHash, err := util.HashFromReader(rd)
 	if err != nil {
 		return nil, err
 	}
+	batchNum := batchNumHash.Big().Uint64()
 
 	l1BaseFee, err := util.HashFromReader(rd)
 	if err != nil {
 		return nil, err
 	}
-	data, err := util.PackInternalTxDataBatchPostingReport(batchTimestamp, batchPosterAddr, batchNum, l1BaseFee)
+	batchData := batchFetcher(batchNum)
+	var batchDataGas uint64
+	for _, b := range batchData {
+		if b == 0 {
+			batchDataGas += params.TxDataZeroGas
+		} else {
+			batchDataGas += params.TxDataNonZeroGasEIP2028
+		}
+	}
+	data, err := util.PackInternalTxDataBatchPostingReport(batchTimestamp, batchPosterAddr, batchNum, batchDataGas, l1BaseFee.Big())
 	if err != nil {
 		return nil, err
 	}

arbos/incomingmessage_test.go

@@ -34,7 +34,7 @@ func TestSerializeAndParseL1Message(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-	txes, err := newMsg.ParseL2Transactions(chainId)
+	txes, err := newMsg.ParseL2Transactions(chainId, nil)
 	if err != nil {
 		t.Error(err)
 	}

arbos/internal_tx.go

@@ -5,10 +5,10 @@ package arbos
 
 import (
 	"fmt"
-	"github.com/ethereum/go-ethereum/log"
-	"github.com/ethereum/go-ethereum/params"
 	"math/big"
 
+	"github.com/ethereum/go-ethereum/log"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/core/vm"
@@ -48,9 +48,7 @@ func InternalTxStartBlock(
 	}
 }
 
-type BatchFetcherFunc func(batchNum uint64, batchDataHash common.Hash) (batchData []byte, err error)
-
-func ApplyInternalTxUpdate(tx *types.ArbitrumInternalTx, state *arbosState.ArbosState, evm *vm.EVM, fetcher BatchFetcherFunc) {
+func ApplyInternalTxUpdate(tx *types.ArbitrumInternalTx, state *arbosState.ArbosState, evm *vm.EVM) {
 	switch tx.SubType {
 	case arbInternalTxStartBlock:
 		inputs, err := util.UnpackInternalTxDataStartBlock(tx.Data)
@@ -102,23 +100,11 @@ func ApplyInternalTxUpdate(tx *types.ArbitrumInternalTx, state *arbosState.Arbos
 			panic(err)
 		}
 		batchTimestamp, _ := inputs[0].(*big.Int)
-		// ignore input[1], batchPosterAddress, which exists because we might need it in the future
-		batchNumberBig, _ := inputs[2].(*big.Int)
-		batchNumber := batchNumberBig.Uint64()
-		batchDataHash, _ := inputs[3].(common.Hash)
+		// ignore input[1], batchPosterAddress, and input[2], batchNumber, which exist because we might need them in the future
+		batchDataGas, _ := inputs[3].(uint64)
 		l1BaseFeeWei, _ := inputs[4].(*big.Int)
 
-		batchData, err := fetcher(batchNumber, batchDataHash)
-		if err != nil {
-			panic(err)
-		}
-		dataGas := params.TxDataNonZeroGasEIP2028 * uint64(len(batchData))
-		for _, b := range batchData {
-			if b == 0 {
-				dataGas -= params.TxDataNonZeroGasEIP2028 - params.TxDataZeroGas
-			}
-		}
-		weiSpent := new(big.Int).Mul(l1BaseFeeWei, new(big.Int).SetUint64(dataGas))
+		weiSpent := new(big.Int).Mul(l1BaseFeeWei, new(big.Int).SetUint64(batchDataGas))
 		err = state.L1PricingState().UpdateForSequencerSpending(evm.StateDB, batchTimestamp.Uint64(), evm.Context.Time.Uint64(), weiSpent)
 		if err != nil {
 			log.Warn("L1Pricing UpdateForSequencerSpending failed", "err", err)

arbos/tx_processor.go

@@ -28,8 +28,6 @@ import (
 
 var arbosAddress = types.ArbosAddress
 
-var BatchFetcher BatchFetcherFunc
-
 // A TxProcessor is created and freed for every L2 transaction.
 // It tracks state for ArbOS, allowing it infuence in Geth's tx processing.
 // Public fields are accessible in precompiles.
@@ -57,7 +55,6 @@ func NewTxProcessor(evm *vm.EVM, msg core.Message) *TxProcessor {
 		TopTxType:        nil,
 		evm:              evm,
 		CurrentRetryable: nil,
-		// TODO: initialize batchFetcher
 	}
 }
 
@@ -122,7 +119,7 @@ func (p *TxProcessor) StartTxHook() (endTxNow bool, gasUsed uint64, err error, r
 		if p.msg.From() != arbosAddress {
 			return false, 0, errors.New("internal tx not from arbAddress"), nil
 		}
-		ApplyInternalTxUpdate(tx, p.state, evm, BatchFetcher)
+		ApplyInternalTxUpdate(tx, p.state, evm)
 		return true, 0, nil, nil
 	case *types.ArbitrumSubmitRetryableTx:
 		defer (startTracer())()

arbstate/geth_test.go

@@ -116,7 +116,7 @@ func RunMessagesThroughAPI(t *testing.T, msgs [][]byte, statedb *state.StateDB)
 		if err != nil {
 			t.Error(err)
 		}
-		txes, err := msg.ParseL2Transactions(chainId)
+		txes, err := msg.ParseL2Transactions(chainId, nil)
 		if err != nil {
 			t.Error(err)
 		}

cmd/replay/main.go

@@ -152,12 +152,14 @@ func main() {
 
 		message := readMessage(chainConfig.ArbitrumChainParams.DataAvailabilityCommittee)
 
-		arbos.BatchFetcher = func(batchSeqNum uint64, batchHash common.Hash) ([]byte, error) {
-			return wavmio.ResolvePreImage(batchHash), nil
-		}
-
 		chainContext := WavmChainContext{}
-		newBlock, _ = arbos.ProduceBlock(message.Message, message.DelayedMessagesRead, lastBlockHeader, statedb, chainContext, chainConfig)
+		batchFetcher := func(batchNum uint64) ([]byte, error) {
+			return wavmio.ReadInboxMessage(batchNum), nil
+		}
+		newBlock, _, err = arbos.ProduceBlock(message.Message, message.DelayedMessagesRead, lastBlockHeader, statedb, chainContext, chainConfig, batchFetcher)
+		if err != nil {
+			panic(err)
+		}
 
 	} else {
 		// Initialize ArbOS with this init message and create the genesis block.

contracts/src/precompiles/ArbosActs.sol

@@ -41,8 +41,8 @@ interface ArbosActs {
     function batchPostingReport(
         uint256 batchTimestamp,
         address batchPosterAddress,
-        uint256 batchNumber,
-        bytes32 batchDataHash,
+        uint64 batchNumber,
+        uint64 batchDataGas,
         uint256 l1BaseFeeWei
     ) external;
 

precompiles/ArbosActs.go

@@ -15,6 +15,6 @@ func (con ArbosActs) StartBlock(c ctx, evm mech, l1BaseFee, l2BaseFeeLastBlock h
 	return con.CallerNotArbOSError()
 }
 
-func (con ArbosActs) BatchPostingReport(c ctx, evm mech, batchTimestamp huge, batchPosterAddress addr, batchNumber huge, batchDataHash hash, l1BaseFeeWei huge) error {
+func (con ArbosActs) BatchPostingReport(c ctx, evm mech, batchTimestamp huge, batchPosterAddress addr, batchNumber uint64, batchDataGas uint64, l1BaseFeeWei huge) error {
 	return con.CallerNotArbOSError()
 }