Merge branch 'master' into fix-historical-feed-flood

Author: Tristan-Wilson

arbnode/batch_poster.go

@@ -310,6 +310,7 @@ var TestBatchPosterConfig = BatchPosterConfig{
 	L1BlockBound:                   "",
 	L1BlockBoundBypass:             time.Hour,
 	UseAccessLists:                 true,
+	RedisLock:                      redislock.TestCfg,
 	GasEstimateBaseFeeMultipleBips: arbmath.OneInUBips * 3 / 2,
 	CheckBatchCorrectness:          true,
 	DelayBufferThresholdMargin:     0,
@@ -1647,12 +1648,14 @@ func (b *BatchPoster) MaybePostSequencerBatch(ctx context.Context) (bool, error)
 	}
 
 	sequencerMsg, err := b.building.segments.CloseAndGetBytes()
+	defer func() {
+		b.building = nil // a closed batchSegments can't be reused
+	}()
 	if err != nil {
 		return false, err
 	}
 	if sequencerMsg == nil {
 		log.Debug("BatchPoster: batch nil", "sequence nr.", batchPosition.NextSeqNum, "from", batchPosition.MessageCount, "prev delayed", batchPosition.DelayedMessageCount)
-		b.building = nil // a closed batchSegments can't be reused
 		return false, nil
 	}
 
@@ -1803,17 +1806,19 @@ func (b *BatchPoster) MaybePostSequencerBatch(ctx context.Context) (bool, error)
 				return false, fmt.Errorf("error getting message from simulated inbox multiplexer (Pop) when testing correctness of batch: %w", err)
 			}
 			if msg.DelayedMessagesRead != b.building.muxBackend.allMsgs[i].DelayedMessagesRead {
-				b.building = nil
 				return false, fmt.Errorf("simulated inbox multiplexer failed to produce correct delayedMessagesRead field for msg with seqNum: %d. Got: %d, Want: %d", i, msg.DelayedMessagesRead, b.building.muxBackend.allMsgs[i].DelayedMessagesRead)
 			}
 			if !msg.Message.Equals(b.building.muxBackend.allMsgs[i].Message) {
-				b.building = nil
 				return false, fmt.Errorf("simulated inbox multiplexer failed to produce correct message field for msg with seqNum: %d", i)
 			}
 		}
 		log.Debug("Successfully checked that the batch produces correct messages when ran through inbox multiplexer", "sequenceNumber", batchPosition.NextSeqNum)
 	}
 
+	if !b.redisLock.AttemptLock(ctx) {
+		return false, errAttemptLockFailed
+	}
+
 	tx, err := b.dataPoster.PostTransaction(ctx,
 		firstUsefulMsgTime,
 		nonce,
@@ -1888,7 +1893,6 @@ func (b *BatchPoster) MaybePostSequencerBatch(ctx context.Context) (bool, error)
 		backlog = 0
 	}
 	b.backlog.Store(backlog)
-	b.building = nil
 
 	// If we aren't queueing up transactions, wait for the receipt before moving on to the next batch.
 	if config.DataPoster.UseNoOpStorage {

arbnode/dataposter/data_poster.go

@@ -1213,7 +1213,6 @@ func (p *DataPoster) Start(ctxIn context.Context) {
 			} else {
 				log.Error("Failed to fetch latest confirmed tx from queue", "confirmedNonce", confirmedNonce, "err", err, "confirmedMeta", confirmedMeta)
 			}
-
 		}
 
 		for _, tx := range queueContents {
@@ -1226,9 +1225,14 @@ func (p *DataPoster) Start(ctxIn context.Context) {
 				err := p.sendTx(ctx, tx, tx)
 				p.maybeLogError(err, tx, "failed to re-send transaction")
 			}
-			tx, err = p.queue.Get(ctx, tx.FullTx.Nonce())
+			nonce := tx.FullTx.Nonce()
+			tx, err = p.queue.Get(ctx, nonce)
 			if err != nil {
-				log.Error("Failed to fetch tx from queue to check updated status", "nonce", tx.FullTx.Nonce(), "err", err)
+				log.Error("Failed to fetch tx from queue to check updated status", "nonce", nonce, "err", err)
+				return minWait
+			}
+			if tx == nil {
+				log.Error("Failed to fetch tx from queue to check updated status, got tx == nil", "nonce", nonce)
 				return minWait
 			}
 			if nextCheck.After(tx.NextReplacement) {

arbnode/dataposter/dbstorage/storage.go

@@ -18,7 +18,7 @@ import (
 
 // Storage implements db based storage for batch poster.
 type Storage struct {
-	db     ethdb.Database
+	db     ethdb.KeyValueStore
 	encDec storage.EncoderDecoderF
 }
 
@@ -31,7 +31,7 @@ var (
 	countKey       = []byte(".count_key")
 )
 
-func New(db ethdb.Database, enc storage.EncoderDecoderF) *Storage {
+func New(db ethdb.KeyValueStore, enc storage.EncoderDecoderF) *Storage {
 	return &Storage{db: db, encDec: enc}
 }
 

arbnode/dataposter/storage_test.go

@@ -47,7 +47,7 @@ func newLevelDBStorage(t *testing.T, encF storage.EncoderDecoderF) *dbstorage.St
 
 func newPebbleDBStorage(t *testing.T, encF storage.EncoderDecoderF) *dbstorage.Storage {
 	t.Helper()
-	db, err := node.NewPebbleDBDatabase(path.Join(t.TempDir(), "pebble.db"), 0, 0, "default", false, true, conf.PersistentConfigDefault.Pebble.ExtraOptions("pebble"))
+	db, err := node.NewPebbleDBDatabase(path.Join(t.TempDir(), "pebble.db"), 0, 0, "default", false, conf.PersistentConfigDefault.Pebble.ExtraOptions("pebble"))
 	if err != nil {
 		t.Fatalf("NewPebbleDBDatabase() unexpected error: %v", err)
 	}

arbnode/inbox_test.go

@@ -135,8 +135,8 @@ func NewTransactionStreamerForTest(t *testing.T, ctx context.Context, ownerAddre
 	arbDb := rawdb.NewMemoryDatabase()
 	initReader := statetransfer.NewMemoryInitDataReader(&initData)
 
-	cacheConfig := core.DefaultCacheConfigWithScheme(env.GetTestStateScheme())
-	bc, err := gethexec.WriteOrTestBlockChain(chainDb, cacheConfig, initReader, chainConfig, nil, nil, arbostypes.TestInitMessage, &gethexec.ConfigDefault.TxIndexer, 0)
+	options := core.DefaultConfig().WithStateScheme(env.GetTestStateScheme())
+	bc, err := gethexec.WriteOrTestBlockChain(chainDb, options, initReader, chainConfig, nil, nil, arbostypes.TestInitMessage, &gethexec.ConfigDefault.TxIndexer, 0)
 
 	if err != nil {
 		Fail(t, err)

arbnode/redislock/redis.go

@@ -71,6 +71,14 @@ var DefaultCfg = SimpleCfg{
 	BackgroundLock:  false,
 }
 
+var TestCfg = SimpleCfg{
+	Enable:          true,
+	LockoutDuration: time.Second,
+	RefreshDuration: time.Second / 6,
+	Key:             "",
+	BackgroundLock:  false,
+}
+
 func (l *Simple) attemptLock(ctx context.Context) (bool, error) {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()

arbos/arbosState/initialization_test.go

@@ -65,10 +65,10 @@ func tryMarshalUnmarshal(input *statetransfer.ArbosInitializationInfo, t *testin
 	initReader := statetransfer.NewMemoryInitDataReader(&initData)
 	chainConfig := chaininfo.ArbitrumDevTestChainConfig()
 
-	cacheConfig := core.DefaultCacheConfigWithScheme(env.GetTestStateScheme())
-	stateroot, err := InitializeArbosInDatabase(raw, cacheConfig, initReader, chainConfig, nil, arbostypes.TestInitMessage, 0, 0)
+	options := core.DefaultConfig().WithStateScheme(env.GetTestStateScheme())
+	stateroot, err := InitializeArbosInDatabase(raw, options, initReader, chainConfig, nil, arbostypes.TestInitMessage, 0, 0)
 	Require(t, err)
-	triedbConfig := cacheConfig.TriedbConfig()
+	triedbConfig := options.TriedbConfig()
 	stateDb, err := state.New(stateroot, state.NewDatabase(triedb.NewDatabase(raw, triedbConfig), nil))
 	Require(t, err)
 

arbos/arbosState/initialize.go

@@ -58,7 +58,7 @@ func MakeGenesisBlock(parentHash common.Hash, blockNumber uint64, timestamp uint
 	return types.NewBlock(head, nil, nil, trie.NewStackTrie(nil))
 }
 
-func InitializeArbosInDatabase(db ethdb.Database, cacheConfig *core.CacheConfig, initData statetransfer.InitDataReader, chainConfig *params.ChainConfig, genesisArbOSInit *params.ArbOSInit, initMessage *arbostypes.ParsedInitMessage, timestamp uint64, accountsPerSync uint) (root common.Hash, err error) {
+func InitializeArbosInDatabase(db ethdb.Database, cacheConfig *core.BlockChainConfig, initData statetransfer.InitDataReader, chainConfig *params.ChainConfig, genesisArbOSInit *params.ArbOSInit, initMessage *arbostypes.ParsedInitMessage, timestamp uint64, accountsPerSync uint) (root common.Hash, err error) {
 	triedbConfig := cacheConfig.TriedbConfig()
 	triedbConfig.Preimages = false
 	stateDatabase := state.NewDatabase(triedb.NewDatabase(db, triedbConfig), nil)

broadcaster/backlog/backlog.go

@@ -87,6 +87,29 @@ func (b *backlog) backlogSizeInBytes() (uint64, error) {
 	return size, nil
 }
 
+// deepCopyMessageForDebug creates a selective copy of the message for memory debugging.
+// It only deep copies the L2msg field which is typically the largest memory consumer.
+// This helps pprof show allocations at the Append point rather than at message creation.
+func deepCopyMessageForDebug(msg *m.BroadcastFeedMessage) *m.BroadcastFeedMessage {
+	if msg == nil || msg.Message.Message == nil || msg.Message.Message.L2msg == nil {
+		return msg
+	}
+
+	copied := *msg
+
+	l2msgCopy := make([]byte, len(msg.Message.Message.L2msg))
+	copy(l2msgCopy, msg.Message.Message.L2msg)
+
+	l1IncomingMessageCopy := *msg.Message.Message
+	l1IncomingMessageCopy.L2msg = l2msgCopy
+
+	messageWithMetadataCopy := msg.Message
+	messageWithMetadataCopy.Message = &l1IncomingMessageCopy
+	copied.Message = messageWithMetadataCopy
+
+	return &copied
+}
+
 // Append will add the given messages to the backlogSegment at head until
 // that segment reaches its limit. If messages remain to be added a new segment
 // will be created.
@@ -103,8 +126,14 @@ func (b *backlog) Append(bm *m.BroadcastMessage) error {
 		}
 	}
 
+	enableDeepCopy := b.config().EnableBacklogDeepCopy
 	lookupByIndex := b.lookupByIndex.Load()
 	for _, msg := range bm.Messages {
+		// For memory debugging: deep copy L2msg to track allocations
+		msgToAppend := msg
+		if enableDeepCopy {
+			msgToAppend = deepCopyMessageForDebug(msg)
+		}
 		segment := b.tail.Load()
 		if segment == nil {
 			segment = newBacklogSegment()
@@ -124,7 +153,7 @@ func (b *backlog) Append(bm *m.BroadcastMessage) error {
 		if segment.count() >= b.config().SegmentLimit {
 			segment.messagesLock.RLock()
 			if len(segment.messages) > 0 {
-				msg.CumulativeSumMsgSize = segment.messages[len(segment.messages)-1].CumulativeSumMsgSize
+				msgToAppend.CumulativeSumMsgSize = segment.messages[len(segment.messages)-1].CumulativeSumMsgSize
 			}
 			segment.messagesLock.RUnlock()
 
@@ -135,7 +164,7 @@ func (b *backlog) Append(bm *m.BroadcastMessage) error {
 			b.tail.Store(segment)
 		}
 
-		err := segment.append(prevMsgIdx, msg)
+		err := segment.append(prevMsgIdx, msgToAppend)
 		if errors.Is(err, errDropSegments) {
 			head := b.head.Load()
 			b.removeFromLookup(head.Start(), uint64(msg.SequenceNumber))

broadcaster/backlog/config.go

@@ -7,18 +7,22 @@ import (
 type ConfigFetcher func() *Config
 
 type Config struct {
-	SegmentLimit int `koanf:"segment-limit" reload:"hot"`
+	SegmentLimit          int  `koanf:"segment-limit" reload:"hot"`
+	EnableBacklogDeepCopy bool `koanf:"enable-backlog-deep-copy" reload:"hot"`
 }
 
 func AddOptions(prefix string, f *flag.FlagSet) {
 	f.Int(prefix+".segment-limit", DefaultConfig.SegmentLimit, "the maximum number of messages each segment within the backlog can contain")
+	f.Bool(prefix+".enable-backlog-deep-copy", DefaultConfig.EnableBacklogDeepCopy, "enable deep copying of L2 messages for memory profiling (debug only)")
 }
 
 var (
 	DefaultConfig = Config{
-		SegmentLimit: 240,
+		SegmentLimit:          240,
+		EnableBacklogDeepCopy: false,
 	}
 	DefaultTestConfig = Config{
-		SegmentLimit: 3,
+		SegmentLimit:          3,
+		EnableBacklogDeepCopy: false,
 	}
 )