address PR comments

Author: ganeshvanahalli

arbnode/db-schema/schema.go

@@ -15,8 +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 [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
+	MelDelayedMessagePrefix             []byte = []byte("y") // maps a delayed sequence number to an accumulator and an RLP encoded message [TODO(NIT-4209): might need to replace or be replaced by RlpDelayedMessagePrefix]
+	MelSequencerBatchMetaPrefix         []byte = []byte("q") // maps a batch sequence number to BatchMetadata [TODO(NIT-4209): 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

@@ -1,7 +1,6 @@
 package mel
 
 import (
-	"context"
 	"errors"
 	"fmt"
 
@@ -21,40 +20,30 @@ type DelayedMessageBacklogEntry struct {
 // also contains compact witnesses of a Merkle tree representing all seen delayed messages. To prove that a delayed message is part of
 // this Merkle tree, this data structure can be used to verify Merkle proofs against the MEL state.
 type DelayedMessageBacklog struct {
-	ctx                          context.Context
-	capacity                     int
+	targetBufferSize             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)
+	finalizedAndReadIndexFetcher func() (uint64, error)
 }
 
-func NewDelayedMessageBacklog(ctx context.Context, capacity int, finalizedAndReadIndexFetcher func(context.Context) (uint64, error), opts ...func(*DelayedMessageBacklog)) (*DelayedMessageBacklog, error) {
-	if capacity == 0 {
-		return nil, fmt.Errorf("capacity of DelayedMessageBacklog cannot be zero")
+func NewDelayedMessageBacklog(targetBufferSize int, finalizedAndReadIndexFetcher func() (uint64, error)) (*DelayedMessageBacklog, error) {
+	if targetBufferSize == 0 {
+		return nil, fmt.Errorf("targetBufferSize of DelayedMessageBacklog cannot be zero")
 	}
 	if finalizedAndReadIndexFetcher == nil {
 		return nil, fmt.Errorf("finalizedAndReadIndexFetcher of DelayedMessageBacklog cannot be nil")
 	}
 	backlog := &DelayedMessageBacklog{
-		ctx:                          ctx,
-		capacity:                     capacity,
+		targetBufferSize:             targetBufferSize,
 		entries:                      make([]*DelayedMessageBacklogEntry, 0),
 		initMessage:                  nil,
 		finalizedAndReadIndexFetcher: finalizedAndReadIndexFetcher,
 	}
-	for _, opt := range opts {
-		opt(backlog)
-	}
 	return backlog, nil
 }
 
-func WithUnboundedCapacity(d *DelayedMessageBacklog) {
-	d.capacity = 0
-	d.finalizedAndReadIndexFetcher = nil
-}
-
-// Add takes values of a DelayedMessageBacklogEntry and adds it to the backlog given the entry succeeds validation. It also attempts trimming of backlog if capacity is reached
+// Add takes values of a DelayedMessageBacklogEntry and adds it to the backlog given the entry succeeds validation. It also attempts trimming of backlog if targetBufferSize is reached
 func (d *DelayedMessageBacklog) Add(entry *DelayedMessageBacklogEntry) error {
 	if len(d.entries) > 0 {
 		expectedIndex := d.entries[0].Index + uint64(len(d.entries))
@@ -63,7 +52,11 @@ func (d *DelayedMessageBacklog) Add(entry *DelayedMessageBacklogEntry) error {
 		}
 	}
 	d.entries = append(d.entries, entry)
-	return d.clear()
+	if len(d.entries) <= d.targetBufferSize {
+		return nil
+	}
+	d.trimFinalizedAndReadEntries()
+	return nil
 }
 
 func (d *DelayedMessageBacklog) Get(index uint64) (*DelayedMessageBacklogEntry, error) {
@@ -86,27 +79,24 @@ func (d *DelayedMessageBacklog) Len() int                            { return le
 func (d *DelayedMessageBacklog) GetInitMsg() *DelayedInboxMessage    { return d.initMessage }
 func (d *DelayedMessageBacklog) setInitMsg(msg *DelayedInboxMessage) { d.initMessage = msg }
 
-// clear removes from backlog (if exceeds capacity) the entries that correspond to the delayed messages that are both READ and belong to finalized parent chain blocks
-func (d *DelayedMessageBacklog) clear() error {
-	if len(d.entries) <= d.capacity {
-		return nil
-	}
+// trimFinalizedAndReadEntries removes from backlog (if exceeds targetBufferSize) the entries that correspond to the delayed messages that are both READ
+// and belong to finalized parent chain blocks. We should not interrupt delayed messages accumulation if we cannot trim the backlog, since its not high priority
+func (d *DelayedMessageBacklog) trimFinalizedAndReadEntries() {
 	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)
+		finalizedDelayedMessagesRead, err := d.finalizedAndReadIndexFetcher()
 		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
+			return
 		}
 		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
+			leftTrimPos = min(leftTrimPos, uint64(d.dirtiesStartPos)) // cannot trim dirties yet, they will be trimmed 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
@@ -120,7 +110,7 @@ func (d *DelayedMessageBacklog) reorg(newDelayedMessagedSeen uint64) error {
 	if newDelayedMessagedSeen >= d.entries[0].Index {
 		rightTrimPos := newDelayedMessagedSeen - d.entries[0].Index
 		if rightTrimPos > uint64(len(d.entries)) {
-			return fmt.Errorf("newDelayedMessagedSeen: %d durign a reorg is greater (by more than 1) than the greatest delayed message index stored in backlog: %d", newDelayedMessagedSeen, d.entries[len(d.entries)-1].Index)
+			return fmt.Errorf("newDelayedMessagedSeen: %d during a reorg is greater (by more than 1) than the greatest delayed message index stored in backlog: %d", newDelayedMessagedSeen, d.entries[len(d.entries)-1].Index)
 		}
 		d.entries = d.entries[:rightTrimPos]
 	} else {

arbnode/mel/delayed_message_backlog_test.go

@@ -1,7 +1,6 @@
 package mel
 
 import (
-	"context"
 	"reflect"
 	"strings"
 	"testing"
@@ -10,10 +9,7 @@ import (
 )
 
 func TestDelayedMessageBacklog(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	backlog, err := NewDelayedMessageBacklog(ctx, 1, func(ctx context.Context) (uint64, error) { return 0, nil }, WithUnboundedCapacity)
+	backlog, err := NewDelayedMessageBacklog(100, func() (uint64, error) { return 0, nil })
 	require.NoError(t, err)
 
 	// Verify handling of dirties
@@ -64,8 +60,8 @@ func TestDelayedMessageBacklog(t *testing.T) {
 
 	// Verify that advancing the finalizedAndRead will trim the delayedMessageBacklogEntry while keeping the unread ones
 	finalizedAndRead := uint64(7)
-	backlog.finalizedAndReadIndexFetcher = func(context.Context) (uint64, error) { return finalizedAndRead, nil }
-	require.NoError(t, backlog.clear())
+	backlog.finalizedAndReadIndexFetcher = func() (uint64, error) { return finalizedAndRead, nil }
+	backlog.trimFinalizedAndReadEntries()
 	require.True(t, len(backlog.entries) == int(numEntries-finalizedAndRead)) // #nosec G115
 	require.True(t, backlog.entries[0].Index == finalizedAndRead)
 

arbnode/mel/runner/backlog.go

@@ -9,7 +9,7 @@ import (
 
 // InitializeDelayedMessageBacklog is to be only called by the Start fsm step of MEL. This function fills the backlog based on the seen and read count from the given mel state
 func InitializeDelayedMessageBacklog(ctx context.Context, d *mel.DelayedMessageBacklog, db *Database, state *mel.State, finalizedAndReadIndexFetcher func(context.Context) (uint64, error)) error {
-	if state.DelayedMessagesSeen == 0 && state.DelayedMessagesRead == 0 { // this is the first mel state so no need to initialize backlog even if the state isnt finalized yet
+	if state.DelayedMessagesSeen == 0 && state.DelayedMessagesRead == 0 { // this is the first mel state so no need to initialize backlog even if the state isn't finalized yet
 		return nil
 	}
 	finalizedDelayedMessagesRead := state.DelayedMessagesRead // Assume to be finalized, then update if needed
@@ -31,7 +31,7 @@ func InitializeDelayedMessageBacklog(ctx context.Context, d *mel.DelayedMessageB
 		return err
 	}
 	if uint64(len(delayedMsgIndexToParentChainBlockNum)) < state.DelayedMessagesSeen-targetDelayedMessagesRead {
-		return fmt.Errorf("number of mappings from index to ParentChainBlockNum: %d are insufficient, needed atleast: %d", uint64(len(delayedMsgIndexToParentChainBlockNum)), state.DelayedMessagesSeen-targetDelayedMessagesRead)
+		return fmt.Errorf("number of mappings from index to ParentChainBlockNum: %d are insufficient, needed atleast: %d", len(delayedMsgIndexToParentChainBlockNum), state.DelayedMessagesSeen-targetDelayedMessagesRead)
 	}
 
 	// Create DelayedMessageBacklogEntry for all the delayed messages that are seen but not read

arbnode/mel/runner/backlog_test.go

@@ -73,7 +73,7 @@ func TestDelayedMessageBacklogInitialization(t *testing.T) {
 
 	require.True(t, state.DelayedMessagesSeen == uint64(numMelStates)*5+1) // #nosec G115
 	require.True(t, state.DelayedMessagesRead == 1)
-	delayedMessageBacklog, err := mel.NewDelayedMessageBacklog(ctx, 1, func(ctx context.Context) (uint64, error) { return 0, nil }, mel.WithUnboundedCapacity)
+	delayedMessageBacklog, err := mel.NewDelayedMessageBacklog(100, func() (uint64, error) { return 0, nil })
 	require.NoError(t, err)
 	require.NoError(t, InitializeDelayedMessageBacklog(ctx, delayedMessageBacklog, melDb, state, nil))
 	require.True(t, delayedMessageBacklog.Len() == 25)
@@ -108,7 +108,7 @@ func TestDelayedMessageBacklogInitialization(t *testing.T) {
 	// delayedMessageBacklogEntry for indexes below the DelayedMessagesRead as those have not been finalized yet!
 	newState, err := melDb.GetHeadMelState(ctx)
 	require.NoError(t, err)
-	newDelayedMessageBacklog, err := mel.NewDelayedMessageBacklog(ctx, 1, func(ctx context.Context) (uint64, error) { return 0, nil }, mel.WithUnboundedCapacity)
+	newDelayedMessageBacklog, err := mel.NewDelayedMessageBacklog(100, func() (uint64, error) { return 0, nil })
 	require.NoError(t, err)
 	require.NoError(t, InitializeDelayedMessageBacklog(ctx, newDelayedMessageBacklog, melDb, newState, func(context.Context) (uint64, error) { return 7, nil }))
 	// Notice that instead of having seenUnread list from delayed index 13 to 25 inclusive we will have it from 7 to 25 as only till block=7 the chain has finalized and that block has DelayedMessagesRead=7