Merge pull request #3471 from OffchainLabs/mel-uses-logs

Message extraction function works with logs instead of receipts

Author: rauljordan

arbnode/db-schema/schema.go

@@ -15,7 +15,8 @@ var (
 	SequencerBatchMetaPrefix            []byte = []byte("s") // maps a batch sequence number to BatchMetadata
 	DelayedSequencedPrefix              []byte = []byte("a") // maps a delayed message count to the first sequencer batch sequence number with this delayed count
 	MelStatePrefix                      []byte = []byte("l") // maps a parent chain block number to its computed MEL state
-	MelDelayedMessagePrefix             []byte = []byte("y") // maps a delayed sequence number to an accumulator and an RLP encoded message [Note: might need to replace or be replaced by RlpDelayedMessagePrefix]
+	MelDelayedMessagePrefix             []byte = []byte("y") // maps a delayed sequence number to an accumulator and an RLP encoded message [TODO: might need to replace or be replaced by RlpDelayedMessagePrefix]
+	MelSequencerBatchMetaPrefix         []byte = []byte("q") // maps a batch sequence number to BatchMetadata [TODO: might need to replace or be replaced by SequencerBatchMetaPrefix
 
 	MessageCountKey             []byte = []byte("_messageCount")                // contains the current message count
 	LastPrunedMessageKey        []byte = []byte("_lastPrunedMessageKey")        // contains the last pruned message key

arbnode/mel/delayed_message_backlog.go

@@ -24,6 +24,7 @@ type DelayedMessageBacklog struct {
 	ctx                          context.Context
 	capacity                     int
 	entries                      []*DelayedMessageBacklogEntry
+	dirtiesStartPos              int // represents the starting point of dirties in the entries list, items added while processing a state
 	initMessage                  *DelayedInboxMessage
 	finalizedAndReadIndexFetcher func(context.Context) (uint64, error)
 }
@@ -80,7 +81,8 @@ func (d *DelayedMessageBacklog) Get(index uint64) (*DelayedMessageBacklogEntry,
 	return entry, nil
 }
 
-func (d *DelayedMessageBacklog) Len() int                            { return len(d.entries) } // Used for testing InitializeDelayedMessageBacklog function in melrunner
+func (d *DelayedMessageBacklog) CommitDirties()                      { d.dirtiesStartPos = len(d.entries) } // Add dirties to the entries by moving dirtiesStartPos to the end
+func (d *DelayedMessageBacklog) Len() int                            { return len(d.entries) }              // Used for testing InitializeDelayedMessageBacklog function in melrunner
 func (d *DelayedMessageBacklog) GetInitMsg() *DelayedInboxMessage    { return d.initMessage }
 func (d *DelayedMessageBacklog) setInitMsg(msg *DelayedInboxMessage) { d.initMessage = msg }
 
@@ -89,22 +91,29 @@ func (d *DelayedMessageBacklog) clear() error {
 	if len(d.entries) <= d.capacity {
 		return nil
 	}
-	if d.finalizedAndReadIndexFetcher != nil {
+	if d.finalizedAndReadIndexFetcher != nil && d.dirtiesStartPos > 0 { // if all entries are currently dirty we dont trim the finalized ones
 		finalizedDelayedMessagesRead, err := d.finalizedAndReadIndexFetcher(d.ctx)
 		if err != nil {
 			log.Error("Unable to trim finalized and read delayed messages from DelayedMessageBacklog, will be retried later", "err", err)
 			return nil // we should not interrupt delayed messages accumulation if we cannot trim the backlog, since its not high priority
 		}
 		if finalizedDelayedMessagesRead > d.entries[0].Index {
 			leftTrimPos := min(finalizedDelayedMessagesRead-d.entries[0].Index, uint64(len(d.entries)))
+			// #nosec G115
+			leftTrimPos = min(leftTrimPos, uint64(d.dirtiesStartPos)) // cannot clear dirties yet, they will be cleared out in the next attempt
 			d.entries = d.entries[leftTrimPos:]
+			// #nosec G115
+			d.dirtiesStartPos -= int(leftTrimPos) // adjust start position of dirties
 		}
 	}
 	return nil
 }
 
 // Reorg removes from backlog the entries that corresponded to the reorged out parent chain blocks
 func (d *DelayedMessageBacklog) reorg(newDelayedMessagedSeen uint64) error {
+	if d.dirtiesStartPos != len(d.entries) {
+		return fmt.Errorf("delayedMessageBacklog dirties is non-empty when reorg was called, size of dirties:%d", len(d.entries)-d.dirtiesStartPos)
+	}
 	if len(d.entries) == 0 {
 		return nil
 	}
@@ -117,19 +126,13 @@ func (d *DelayedMessageBacklog) reorg(newDelayedMessagedSeen uint64) error {
 	} else {
 		d.entries = make([]*DelayedMessageBacklogEntry, 0)
 	}
+	d.dirtiesStartPos = len(d.entries)
 	return nil
 }
 
+// clone is a shallow clone of DelayedMessageBacklog
 func (d *DelayedMessageBacklog) clone() *DelayedMessageBacklog {
-	var deque []*DelayedMessageBacklogEntry
-	for _, item := range d.entries {
-		msgHash := common.Hash{}
-		copy(msgHash[:], item.MsgHash[:])
-		deque = append(deque, &DelayedMessageBacklogEntry{
-			Index:                       item.Index,
-			MsgHash:                     msgHash,
-			MelStateParentChainBlockNum: item.MelStateParentChainBlockNum,
-		})
-	}
-	return &DelayedMessageBacklog{d.ctx, d.capacity, deque, nil, d.finalizedAndReadIndexFetcher} // Init msg should only be read once, no need to persist it
+	// Remove dirties from entries
+	d.entries = d.entries[:d.dirtiesStartPos]
+	return d
 }

arbnode/mel/delayed_message_backlog_test.go

@@ -15,10 +15,26 @@ func TestDelayedMessageBacklog(t *testing.T) {
 
 	backlog, err := NewDelayedMessageBacklog(ctx, 1, func(ctx context.Context) (uint64, error) { return 0, nil }, WithUnboundedCapacity)
 	require.NoError(t, err)
+
+	// Verify handling of dirties
+	for i := uint64(0); i < 2; i++ {
+		require.NoError(t, backlog.Add(&DelayedMessageBacklogEntry{Index: i}))
+	}
+	backlog.CommitDirties()
+	require.True(t, backlog.dirtiesStartPos == 2)
+	// Add dirties and verify that calling a clone would remove them
+	for i := uint64(2); i < 5; i++ {
+		require.NoError(t, backlog.Add(&DelayedMessageBacklogEntry{Index: i}))
+	}
+	backlog.clone() // should remove all the dirties from entries list
+	require.True(t, len(backlog.entries) == 2)
 	numEntries := uint64(25)
-	for i := uint64(0); i < numEntries; i++ {
+	for i := uint64(2); i < numEntries; i++ {
 		require.NoError(t, backlog.Add(&DelayedMessageBacklogEntry{Index: i}))
 	}
+	backlog.CommitDirties()
+	// #nosec G115
+	require.True(t, uint64(backlog.dirtiesStartPos) == numEntries)
 
 	// Test that clone works
 	cloned := backlog.clone()

arbnode/mel/extraction/abis.go

@@ -7,11 +7,11 @@ import (
 	"github.com/offchainlabs/nitro/solgen/go/bridgegen"
 )
 
-var batchDeliveredID common.Hash
-var inboxMessageDeliveredID common.Hash
-var inboxMessageFromOriginID common.Hash
-var seqInboxABI *abi.ABI
-var iBridgeABI *abi.ABI
+var BatchDeliveredID common.Hash
+var InboxMessageDeliveredID common.Hash
+var InboxMessageFromOriginID common.Hash
+var SeqInboxABI *abi.ABI
+var IBridgeABI *abi.ABI
 var iInboxABI *abi.ABI
 var iDelayedMessageProviderABI *abi.ABI
 
@@ -21,20 +21,20 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
-	batchDeliveredID = sequencerBridgeABI.Events["SequencerBatchDelivered"].ID
+	BatchDeliveredID = sequencerBridgeABI.Events["SequencerBatchDelivered"].ID
 	parsedIBridgeABI, err := bridgegen.IBridgeMetaData.GetAbi()
 	if err != nil {
 		panic(err)
 	}
-	iBridgeABI = parsedIBridgeABI
+	IBridgeABI = parsedIBridgeABI
 	parsedIMessageProviderABI, err := bridgegen.IDelayedMessageProviderMetaData.GetAbi()
 	if err != nil {
 		panic(err)
 	}
 	iDelayedMessageProviderABI = parsedIMessageProviderABI
-	inboxMessageDeliveredID = parsedIMessageProviderABI.Events["InboxMessageDelivered"].ID
-	inboxMessageFromOriginID = parsedIMessageProviderABI.Events["InboxMessageDeliveredFromOrigin"].ID
-	seqInboxABI, err = bridgegen.SequencerInboxMetaData.GetAbi()
+	InboxMessageDeliveredID = parsedIMessageProviderABI.Events["InboxMessageDelivered"].ID
+	InboxMessageFromOriginID = parsedIMessageProviderABI.Events["InboxMessageDeliveredFromOrigin"].ID
+	SeqInboxABI, err = bridgegen.SequencerInboxMetaData.GetAbi()
 	if err != nil {
 		panic(err)
 	}
@@ -43,5 +43,5 @@ func init() {
 		panic(err)
 	}
 	iInboxABI = parsedIInboxABI
-	batchDeliveredID = sequencerBridgeABI.Events["SequencerBatchDelivered"].ID
+	BatchDeliveredID = sequencerBridgeABI.Events["SequencerBatchDelivered"].ID
 }

arbnode/mel/extraction/batch_lookup.go

@@ -20,82 +20,60 @@ func parseBatchesFromBlock(
 	ctx context.Context,
 	melState *mel.State,
 	parentChainHeader *types.Header,
-	txsFetcher TransactionsFetcher,
-	receiptFetcher ReceiptFetcher,
+	txFetcher TransactionFetcher,
+	logsFetcher LogsFetcher,
 	eventUnpacker eventUnpacker,
-) ([]*mel.SequencerInboxBatch, []*types.Transaction, []uint, error) {
-	allBatches := make([]*mel.SequencerInboxBatch, 0)
-	allBatchTxs := make([]*types.Transaction, 0)
-	allBatchTxIndices := make([]uint, 0)
-	parentChainBlockTxs, err := txsFetcher.TransactionsByHeader(ctx, parentChainHeader.Hash())
+) ([]*mel.SequencerInboxBatch, []*types.Transaction, error) {
+	batches := make([]*mel.SequencerInboxBatch, 0)
+	batchTxs := make([]*types.Transaction, 0)
+	logs, err := logsFetcher.LogsForBlockHash(ctx, parentChainHeader.Hash())
 	if err != nil {
-		return nil, nil, nil, fmt.Errorf("failed to fetch transactions for parent chain block %v: %w", parentChainHeader.Hash(), err)
+		return nil, nil, fmt.Errorf("failed to fetch logs from parent chain block %v: %w", parentChainHeader.Hash(), err)
 	}
-	for i, tx := range parentChainBlockTxs {
-		// TODO: remove this temporary work around for handling init message, i.e skipping the check when msgCount==0
-		if melState.MsgCount != 0 {
-			if tx.To() == nil {
-				continue
-			}
-			if *tx.To() != melState.BatchPostingTargetAddress {
-				continue
-			}
+	var lastSeqNum *uint64
+	for _, log := range logs {
+		if log == nil || log.Topics[0] != BatchDeliveredID {
+			continue
 		}
-		// Fetch the receipts for the transaction to get the logs.
-		txIndex := uint(i) // #nosec G115
-		receipt, err := receiptFetcher.ReceiptForTransactionIndex(ctx, txIndex)
-		if err != nil {
-			return nil, nil, nil, err
+		event := new(bridgegen.SequencerInboxSequencerBatchDelivered)
+		if err := eventUnpacker.unpackLogTo(event, SeqInboxABI, "SequencerBatchDelivered", *log); err != nil {
+			return nil, nil, err
 		}
-		if len(receipt.Logs) == 0 {
-			continue
+		if !event.BatchSequenceNumber.IsUint64() {
+			return nil, nil, errors.New("sequencer inbox event has non-uint64 sequence number")
+		}
+		if !event.AfterDelayedMessagesRead.IsUint64() {
+			return nil, nil, errors.New("sequencer inbox event has non-uint64 delayed messages read")
 		}
-		batches := make([]*mel.SequencerInboxBatch, 0, len(receipt.Logs))
-		txs := make([]*types.Transaction, 0, len(receipt.Logs))
-		txIndices := make([]uint, 0, len(receipt.Logs))
-		var lastSeqNum *uint64
-		for _, log := range receipt.Logs {
-			if log == nil || log.Topics[0] != batchDeliveredID {
-				continue
-			}
-			event := new(bridgegen.SequencerInboxSequencerBatchDelivered)
-			if err := eventUnpacker.unpackLogTo(event, seqInboxABI, "SequencerBatchDelivered", *log); err != nil {
-				return nil, nil, nil, err
-			}
-			if !event.BatchSequenceNumber.IsUint64() {
-				return nil, nil, nil, errors.New("sequencer inbox event has non-uint64 sequence number")
-			}
-			if !event.AfterDelayedMessagesRead.IsUint64() {
-				return nil, nil, nil, errors.New("sequencer inbox event has non-uint64 delayed messages read")
-			}
 
-			seqNum := event.BatchSequenceNumber.Uint64()
-			if lastSeqNum != nil {
-				if seqNum != *lastSeqNum+1 {
-					return nil, nil, nil, fmt.Errorf("sequencer batches out of order; after batch %v got batch %v", *lastSeqNum, seqNum)
-				}
+		seqNum := event.BatchSequenceNumber.Uint64()
+		if lastSeqNum != nil {
+			if seqNum != *lastSeqNum+1 {
+				return nil, nil, fmt.Errorf("sequencer batches out of order; after batch %v got batch %v", *lastSeqNum, seqNum)
 			}
-			lastSeqNum = &seqNum
-			batch := &mel.SequencerInboxBatch{
-				BlockHash:              log.BlockHash,
-				ParentChainBlockNumber: log.BlockNumber,
-				SequenceNumber:         seqNum,
-				BeforeInboxAcc:         event.BeforeAcc,
-				AfterInboxAcc:          event.AfterAcc,
-				AfterDelayedAcc:        event.DelayedAcc,
-				AfterDelayedCount:      event.AfterDelayedMessagesRead.Uint64(),
-				RawLog:                 *log,
-				TimeBounds:             event.TimeBounds,
-				DataLocation:           mel.BatchDataLocation(event.DataLocation),
-				BridgeAddress:          log.Address,
-			}
-			batches = append(batches, batch)
-			txs = append(txs, tx)
-			txIndices = append(txIndices, uint(i)) // #nosec G115
 		}
-		allBatches = append(allBatches, batches...)
-		allBatchTxs = append(allBatchTxs, txs...)
-		allBatchTxIndices = append(allBatchTxIndices, txIndices...)
+		lastSeqNum = &seqNum
+
+		tx, err := txFetcher.TransactionByLog(ctx, log)
+		if err != nil {
+			return nil, nil, fmt.Errorf("error fetching tx by hash: %v in parseBatchesFromBlock: %w ", log.TxHash, err)
+		}
+
+		batch := &mel.SequencerInboxBatch{
+			BlockHash:              log.BlockHash,
+			ParentChainBlockNumber: log.BlockNumber,
+			SequenceNumber:         seqNum,
+			BeforeInboxAcc:         event.BeforeAcc,
+			AfterInboxAcc:          event.AfterAcc,
+			AfterDelayedAcc:        event.DelayedAcc,
+			AfterDelayedCount:      event.AfterDelayedMessagesRead.Uint64(),
+			RawLog:                 *log,
+			TimeBounds:             event.TimeBounds,
+			DataLocation:           mel.BatchDataLocation(event.DataLocation),
+			BridgeAddress:          log.Address,
+		}
+		batches = append(batches, batch)
+		batchTxs = append(batchTxs, tx)
 	}
-	return allBatches, allBatchTxs, allBatchTxIndices, nil
+	return batches, batchTxs, nil
 }