Merge pull request #3454 from OffchainLabs/sequencetxs-using-hooks

Tx sequencing using hooks

Author: ganeshvanahalli

arbos/block_processor.go

@@ -115,6 +115,8 @@ func createNewHeader(prevHeader *types.Header, l1info *L1Info, state *arbosState
 type ConditionalOptionsForTx []*arbitrum_types.ConditionalOptions
 
 type SequencingHooks struct {
+	NextTxToSequence        func() (*types.Transaction, error)                                                                                                                                      // Must be set
+	SequencedTx             func(int) (*types.Transaction, error)                                                                                                                                   // Must be set
 	TxErrors                []error                                                                                                                                                                 // This can be unset
 	DiscardInvalidTxsEarly  bool                                                                                                                                                                    // This can be unset
 	PreTxFilter             func(*params.ChainConfig, *types.Header, *state.StateDB, *arbosState.ArbosState, *types.Transaction, *arbitrum_types.ConditionalOptions, common.Address, *L1Info) error // This has to be set. Writes to *state.StateDB object should be avoided to prevent invalid state from permeating
@@ -123,8 +125,43 @@ type SequencingHooks struct {
 	ConditionalOptionsForTx []*arbitrum_types.ConditionalOptions                                                                                                                                    // This can be unset
 }
 
-func NoopSequencingHooks() *SequencingHooks {
+type noopTxScheduler struct {
+	txs               types.Transactions
+	scheduledTxsCount int
+}
+
+func (s *noopTxScheduler) GetNextTx() (*types.Transaction, error) {
+	// This is not supposed to happen, if so we have a bug
+	if s.scheduledTxsCount > len(s.txs) {
+		return nil, errors.New("noopTxScheduler: requested too many transactions")
+	}
+	if s.scheduledTxsCount == len(s.txs) {
+		return nil, nil
+	}
+	s.scheduledTxsCount += 1
+	return s.txs[s.scheduledTxsCount-1], nil
+}
+
+func (s *noopTxScheduler) GetScheduledTx(txId int) (*types.Transaction, error) {
+	// This is not supposed to happen, if so we have a bug
+	if txId > len(s.txs) {
+		return nil, errors.New("transaction queried for does not exist in the noopTxScheduler")
+	}
+	// This is not supposed to happen, if so we have a bug
+	if txId > s.scheduledTxsCount {
+		return nil, errors.New("transaction queried for was not scheduled by the noopTxScheduler")
+	}
+	return s.txs[txId], nil
+}
+
+func NoopSequencingHooks(txes types.Transactions) *SequencingHooks {
+	scheduler := &noopTxScheduler{
+		txes,
+		0,
+	}
 	return &SequencingHooks{
+		scheduler.GetNextTx,
+		scheduler.GetScheduledTx,
 		[]error{},
 		false,
 		func(*params.ChainConfig, *types.Header, *state.StateDB, *arbosState.ArbosState, *types.Transaction, *arbitrum_types.ConditionalOptions, common.Address, *L1Info) error {
@@ -153,17 +190,16 @@ func ProduceBlock(
 		log.Warn("error parsing incoming message", "err", err)
 		txes = types.Transactions{}
 	}
+	hooks := NoopSequencingHooks(txes)
 
-	hooks := NoopSequencingHooks()
 	return ProduceBlockAdvanced(
-		message.Header, txes, delayedMessagesRead, lastBlockHeader, statedb, chainContext, hooks, isMsgForPrefetch, runCtx,
+		message.Header, delayedMessagesRead, lastBlockHeader, statedb, chainContext, hooks, isMsgForPrefetch, runCtx,
 	)
 }
 
 // A bit more flexible than ProduceBlock for use in the sequencer.
 func ProduceBlockAdvanced(
 	l1Header *arbostypes.L1IncomingMessageHeader,
-	txes types.Transactions,
 	delayedMessagesRead uint64,
 	lastBlockHeader *types.Header,
 	statedb *state.StateDB,
@@ -200,7 +236,6 @@ func ProduceBlockAdvanced(
 
 	// Prepend a tx before all others to touch up the state (update the L1 block num, pricing pools, etc)
 	startTx := InternalTxStartBlock(chainConfig.ChainID, l1Header.L1BaseFee, l1BlockNum, header, lastBlockHeader)
-	txes = append(types.Transactions{types.NewTx(startTx)}, txes...)
 
 	complete := types.Transactions{}
 	receipts := types.Receipts{}
@@ -213,14 +248,19 @@ func ProduceBlockAdvanced(
 	// We'll check that the block can fit each message, so this pool is set to not run out
 	gethGas := core.GasPool(l2pricing.GethBlockGasLimit)
 
-	for len(txes) > 0 || len(redeems) > 0 {
+	firstTx := types.NewTx(startTx)
+
+	for {
 		// repeatedly process the next tx, doing redeems created along the way in FIFO order
 
 		var tx *types.Transaction
 		var options *arbitrum_types.ConditionalOptions
-		hooks := NoopSequencingHooks()
+		hooks := NoopSequencingHooks(nil) // TODO: NIT-3678
 		isUserTx := false
-		if len(redeems) > 0 {
+		if firstTx != nil {
+			tx = firstTx
+			firstTx = nil
+		} else if len(redeems) > 0 {
 			tx = redeems[0]
 			redeems = redeems[1:]
 
@@ -234,8 +274,13 @@ func ProduceBlockAdvanced(
 				continue
 			}
 		} else {
-			tx = txes[0]
-			txes = txes[1:]
+			tx, err = sequencingHooks.NextTxToSequence()
+			if err != nil {
+				return nil, nil, fmt.Errorf("error fetching next transaction to sequence, userTxsProcessed: %d, hookTxErrors: %d, err: %w", userTxsProcessed, len(sequencingHooks.TxErrors), err)
+			}
+			if tx == nil {
+				break
+			}
 			if tx.Type() != types.ArbitrumInternalTxType {
 				hooks = sequencingHooks // the sequencer has the ability to drop this tx
 				isUserTx = true

execution/gethexec/executionengine.go

@@ -372,10 +372,14 @@ func (s *ExecutionEngine) NextDelayedMessageNumber() (uint64, error) {
 	return currentHeader.Nonce.Uint64(), nil
 }
 
-func MessageFromTxes(header *arbostypes.L1IncomingMessageHeader, txes types.Transactions, txErrors []error) (*arbostypes.L1IncomingMessage, error) {
+func MessageFromTxes(header *arbostypes.L1IncomingMessageHeader, hooks *arbos.SequencingHooks) (*arbostypes.L1IncomingMessage, error) {
 	var l2Message []byte
-	if len(txes) == 1 && txErrors[0] == nil {
-		txBytes, err := txes[0].MarshalBinary()
+	if len(hooks.TxErrors) == 1 && hooks.TxErrors[0] == nil {
+		tx, err := hooks.SequencedTx(0)
+		if err != nil {
+			return nil, err
+		}
+		txBytes, err := tx.MarshalBinary()
 		if err != nil {
 			return nil, err
 		}
@@ -384,10 +388,14 @@ func MessageFromTxes(header *arbostypes.L1IncomingMessageHeader, txes types.Tran
 	} else {
 		l2Message = append(l2Message, arbos.L2MessageKind_Batch)
 		sizeBuf := make([]byte, 8)
-		for i, tx := range txes {
-			if txErrors[i] != nil {
+		for i := 0; i < len(hooks.TxErrors); i++ {
+			if hooks.TxErrors[i] != nil {
 				continue
 			}
+			tx, err := hooks.SequencedTx(i)
+			if err != nil {
+				return nil, err
+			}
 			txBytes, err := tx.MarshalBinary()
 			if err != nil {
 				return nil, err
@@ -451,9 +459,9 @@ func (s *ExecutionEngine) resequenceReorgedMessages(messages []*arbostypes.Messa
 			log.Warn("failed to parse sequencer message found from reorg", "err", err)
 			continue
 		}
-		hooks := arbos.NoopSequencingHooks()
+		hooks := arbos.NoopSequencingHooks(txes)
 		hooks.DiscardInvalidTxsEarly = true
-		_, err = s.sequenceTransactionsWithBlockMutex(msg.Message.Header, txes, hooks, nil)
+		_, err = s.sequenceTransactionsWithBlockMutex(msg.Message.Header, hooks, nil)
 		if err != nil {
 			log.Error("failed to re-sequence old user message removed by reorg", "err", err)
 			return
@@ -490,17 +498,17 @@ func (s *ExecutionEngine) sequencerWrapper(sequencerFunc func() (*types.Block, e
 	}
 }
 
-func (s *ExecutionEngine) SequenceTransactions(header *arbostypes.L1IncomingMessageHeader, txes types.Transactions, hooks *arbos.SequencingHooks, timeboostedTxs map[common.Hash]struct{}) (*types.Block, error) {
+func (s *ExecutionEngine) SequenceTransactions(header *arbostypes.L1IncomingMessageHeader, hooks *arbos.SequencingHooks, timeboostedTxs map[common.Hash]struct{}) (*types.Block, error) {
 	return s.sequencerWrapper(func() (*types.Block, error) {
 		hooks.TxErrors = nil
-		return s.sequenceTransactionsWithBlockMutex(header, txes, hooks, timeboostedTxs)
+		return s.sequenceTransactionsWithBlockMutex(header, hooks, timeboostedTxs)
 	})
 }
 
 // SequenceTransactionsWithProfiling runs SequenceTransactions with tracing and
 // CPU profiling enabled. If the block creation takes longer than 2 seconds, it
 // keeps both and prints out filenames in an error log line.
-func (s *ExecutionEngine) SequenceTransactionsWithProfiling(header *arbostypes.L1IncomingMessageHeader, txes types.Transactions, hooks *arbos.SequencingHooks, timeboostedTxs map[common.Hash]struct{}) (*types.Block, error) {
+func (s *ExecutionEngine) SequenceTransactionsWithProfiling(header *arbostypes.L1IncomingMessageHeader, hooks *arbos.SequencingHooks, timeboostedTxs map[common.Hash]struct{}) (*types.Block, error) {
 	pprofBuf, traceBuf := bytes.NewBuffer(nil), bytes.NewBuffer(nil)
 	if err := pprof.StartCPUProfile(pprofBuf); err != nil {
 		log.Error("Starting CPU profiling", "error", err)
@@ -509,7 +517,7 @@ func (s *ExecutionEngine) SequenceTransactionsWithProfiling(header *arbostypes.L
 		log.Error("Starting tracing", "error", err)
 	}
 	start := time.Now()
-	res, err := s.SequenceTransactions(header, txes, hooks, timeboostedTxs)
+	res, err := s.SequenceTransactions(header, hooks, timeboostedTxs)
 	elapsed := time.Since(start)
 	pprof.StopCPUProfile()
 	trace.Stop()
@@ -535,7 +543,7 @@ func writeAndLog(pprof, trace *bytes.Buffer) {
 	log.Info("Transactions sequencing took longer than 2 seconds, created pprof and trace files", "pprof", pprofFile, "traceFile", traceFile)
 }
 
-func (s *ExecutionEngine) sequenceTransactionsWithBlockMutex(header *arbostypes.L1IncomingMessageHeader, txes types.Transactions, hooks *arbos.SequencingHooks, timeboostedTxs map[common.Hash]struct{}) (*types.Block, error) {
+func (s *ExecutionEngine) sequenceTransactionsWithBlockMutex(header *arbostypes.L1IncomingMessageHeader, hooks *arbos.SequencingHooks, timeboostedTxs map[common.Hash]struct{}) (*types.Block, error) {
 	lastBlockHeader, err := s.getCurrentHeader()
 	if err != nil {
 		return nil, err
@@ -563,7 +571,6 @@ func (s *ExecutionEngine) sequenceTransactionsWithBlockMutex(header *arbostypes.
 	startTime := time.Now()
 	block, receipts, err := arbos.ProduceBlockAdvanced(
 		header,
-		txes,
 		delayedMessagesRead,
 		lastBlockHeader,
 		statedb,
@@ -577,9 +584,6 @@ func (s *ExecutionEngine) sequenceTransactionsWithBlockMutex(header *arbostypes.
 	}
 	blockCalcTime := time.Since(startTime)
 	blockExecutionTimer.Update(blockCalcTime.Nanoseconds())
-	if len(hooks.TxErrors) != len(txes) {
-		return nil, fmt.Errorf("unexpected number of error results: %v vs number of txes %v", len(hooks.TxErrors), len(txes))
-	}
 
 	if len(receipts) == 0 {
 		return nil, nil
@@ -596,7 +600,7 @@ func (s *ExecutionEngine) sequenceTransactionsWithBlockMutex(header *arbostypes.
 		return nil, nil
 	}
 
-	msg, err := MessageFromTxes(header, txes, hooks.TxErrors)
+	msg, err := MessageFromTxes(header, hooks)
 	if err != nil {
 		return nil, err
 	}