Merge remote-tracking branch 'origin/master' into wasmer7

Author: rory-ocl

arbnode/delayed_seq_reorg_test.go

@@ -20,7 +20,7 @@ func TestSequencerReorgFromDelayed(t *testing.T) {
 	defer cancel()
 
 	exec, streamer, db, _ := NewTransactionStreamerForTest(t, ctx, common.Address{})
-	tracker, err := NewInboxTracker(db, streamer, nil, DefaultSnapSyncConfig)
+	tracker, err := NewInboxTracker(db, streamer, nil)
 	Require(t, err)
 
 	err = streamer.Start(ctx)
@@ -221,7 +221,7 @@ func TestSequencerReorgFromLastDelayedMsg(t *testing.T) {
 	defer cancel()
 
 	exec, streamer, db, _ := NewTransactionStreamerForTest(t, ctx, common.Address{})
-	tracker, err := NewInboxTracker(db, streamer, nil, DefaultSnapSyncConfig)
+	tracker, err := NewInboxTracker(db, streamer, nil)
 	Require(t, err)
 
 	err = streamer.Start(ctx)

arbnode/inbox_reader.go

@@ -149,9 +149,6 @@ func (r *InboxReader) Start(ctxIn context.Context) error {
 			return err
 		}
 		if batchCount > 0 {
-			if r.tracker.snapSyncConfig.Enabled {
-				break
-			}
 			// Validate the init message matches our L2 blockchain
 			ctx, err := r.StopWaiter.GetContextSafe()
 			if err != nil {

arbnode/inbox_test.go

@@ -149,7 +149,7 @@ func NewTransactionStreamerForTest(t *testing.T, ctx context.Context, ownerAddre
 		Fail(t, err)
 	}
 	execSeq := &execClientWrapper{execEngine, t}
-	inbox, err := NewTransactionStreamer(ctx, consensusDB, bc.Config(), execSeq, nil, make(chan error, 1), transactionStreamerConfigFetcher, &DefaultSnapSyncConfig)
+	inbox, err := NewTransactionStreamer(ctx, consensusDB, bc.Config(), execSeq, nil, make(chan error, 1), transactionStreamerConfigFetcher)
 	if err != nil {
 		Fail(t, err)
 	}

arbnode/inbox_tracker.go

@@ -35,24 +35,22 @@ var (
 )
 
 type InboxTracker struct {
-	db             ethdb.Database
-	txStreamer     *TransactionStreamer
-	mutex          sync.Mutex
-	validator      *staker.BlockValidator
-	dapReaders     *daprovider.DAProviderRegistry
-	snapSyncConfig SnapSyncConfig
+	db         ethdb.Database
+	txStreamer *TransactionStreamer
+	mutex      sync.Mutex
+	validator  *staker.BlockValidator
+	dapReaders *daprovider.DAProviderRegistry
 
 	batchMetaMutex sync.Mutex
 	batchMeta      *containers.LruCache[uint64, BatchMetadata]
 }
 
-func NewInboxTracker(db ethdb.Database, txStreamer *TransactionStreamer, dapReaders *daprovider.DAProviderRegistry, snapSyncConfig SnapSyncConfig) (*InboxTracker, error) {
+func NewInboxTracker(db ethdb.Database, txStreamer *TransactionStreamer, dapReaders *daprovider.DAProviderRegistry) (*InboxTracker, error) {
 	tracker := &InboxTracker{
-		db:             db,
-		txStreamer:     txStreamer,
-		dapReaders:     dapReaders,
-		batchMeta:      containers.NewLruCache[uint64, BatchMetadata](1000),
-		snapSyncConfig: snapSyncConfig,
+		db:         db,
+		txStreamer: txStreamer,
+		dapReaders: dapReaders,
+		batchMeta:  containers.NewLruCache[uint64, BatchMetadata](1000),
 	}
 	return tracker, nil
 }
@@ -434,37 +432,13 @@ func (t *InboxTracker) GetDelayedMessageBytes(ctx context.Context, seqNum uint64
 
 func (t *InboxTracker) AddDelayedMessages(messages []*DelayedInboxMessage) error {
 	var nextAcc common.Hash
-	firstDelayedMsgToKeep := uint64(0)
 	if len(messages) == 0 {
 		return nil
 	}
 	pos, err := messages[0].Message.Header.SeqNum()
 	if err != nil {
 		return err
 	}
-	if t.snapSyncConfig.Enabled && pos < t.snapSyncConfig.DelayedCount {
-		firstDelayedMsgToKeep = t.snapSyncConfig.DelayedCount
-		if firstDelayedMsgToKeep > 0 {
-			firstDelayedMsgToKeep--
-		}
-		for {
-			if len(messages) == 0 {
-				return nil
-			}
-			pos, err = messages[0].Message.Header.SeqNum()
-			if err != nil {
-				return err
-			}
-			if pos+1 == firstDelayedMsgToKeep {
-				nextAcc = messages[0].AfterInboxAcc()
-			}
-			if pos < firstDelayedMsgToKeep {
-				messages = messages[1:]
-			} else {
-				break
-			}
-		}
-	}
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
@@ -479,7 +453,7 @@ func (t *InboxTracker) AddDelayedMessages(messages []*DelayedInboxMessage) error
 		return err
 	}
 
-	if pos > firstDelayedMsgToKeep {
+	if pos > 0 {
 		var err error
 		nextAcc, err = t.GetDelayedAcc(pos - 1)
 		if err != nil {
@@ -689,42 +663,16 @@ var delayedMessagesMismatch = errors.New("sequencer batch delayed messages missi
 func (t *InboxTracker) AddSequencerBatches(ctx context.Context, client *ethclient.Client, batches []*SequencerInboxBatch) error {
 	var nextAcc common.Hash
 	var prevbatchmeta BatchMetadata
-	sequenceNumberToKeep := uint64(0)
 	if len(batches) == 0 {
 		return nil
 	}
-	if t.snapSyncConfig.Enabled && batches[0].SequenceNumber < t.snapSyncConfig.BatchCount {
-		sequenceNumberToKeep = t.snapSyncConfig.BatchCount
-		if sequenceNumberToKeep > 0 {
-			sequenceNumberToKeep--
-		}
-		for {
-			if len(batches) == 0 {
-				return nil
-			}
-			if batches[0].SequenceNumber+1 == sequenceNumberToKeep {
-				nextAcc = batches[0].AfterInboxAcc
-				prevbatchmeta = BatchMetadata{
-					Accumulator:         batches[0].AfterInboxAcc,
-					DelayedMessageCount: batches[0].AfterDelayedCount,
-					MessageCount:        arbutil.MessageIndex(t.snapSyncConfig.PrevBatchMessageCount),
-					ParentChainBlock:    batches[0].ParentChainBlockNumber,
-				}
-			}
-			if batches[0].SequenceNumber < sequenceNumberToKeep {
-				batches = batches[1:]
-			} else {
-				break
-			}
-		}
-	}
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
 	pos := batches[0].SequenceNumber
 	startPos := pos
 
-	if pos > sequenceNumberToKeep {
+	if pos > 0 {
 		var err error
 		prevbatchmeta, err = t.GetBatchMetadata(pos - 1)
 		nextAcc = prevbatchmeta.Accumulator

arbnode/node.go

@@ -89,8 +89,6 @@ type Config struct {
 	RPCServer                rpcserver.Config                 `koanf:"rpc-server"`
 	ExecutionRPCClient       rpcclient.ClientConfig           `koanf:"execution-rpc-client" reload:"hot"`
 	VersionAlerterServer     nitroversionalerter.ServerConfig `koanf:"version-alerter-server" reload:"hot"`
-	// SnapSyncConfig is only used for testing purposes, these should not be configured in production.
-	SnapSyncTest SnapSyncConfig
 }
 
 func (c *Config) Validate() error {
@@ -212,7 +210,6 @@ var ConfigDefault = Config{
 	Maintenance:              DefaultMaintenanceConfig,
 	ConsensusExecutionSyncer: DefaultConsensusExecutionSyncerConfig,
 	VersionAlerterServer:     nitroversionalerter.DefaultServerConfig,
-	SnapSyncTest:             DefaultSnapSyncConfig,
 	RPCServer:                rpcserver.DefaultConfig,
 	ExecutionRPCClient: rpcclient.ClientConfig{
 		URL:                       "",
@@ -330,24 +327,6 @@ type Node struct {
 	ConsensusExecutionSyncer *ConsensusExecutionSyncer
 }
 
-type SnapSyncConfig struct {
-	Enabled                   bool
-	PrevBatchMessageCount     uint64
-	PrevDelayedRead           uint64
-	BatchCount                uint64
-	DelayedCount              uint64
-	ParentChainAssertionBlock uint64
-}
-
-var DefaultSnapSyncConfig = SnapSyncConfig{
-	Enabled:                   false,
-	PrevBatchMessageCount:     0,
-	PrevDelayedRead:           0,
-	BatchCount:                0,
-	DelayedCount:              0,
-	ParentChainAssertionBlock: 0,
-}
-
 type ConfigFetcher interface {
 	Get() *Config
 	Start(context.Context)
@@ -743,49 +722,21 @@ func getDAProviders(
 }
 
 func getInboxTrackerAndReader(
-	ctx context.Context,
 	consensusDB ethdb.Database,
 	txStreamer *TransactionStreamer,
 	dapReaders *daprovider.DAProviderRegistry,
-	config *Config,
 	configFetcher ConfigFetcher,
 	l1client *ethclient.Client,
 	l1Reader *headerreader.HeaderReader,
 	deployInfo *chaininfo.RollupAddresses,
 	delayedBridge *DelayedBridge,
 	sequencerInbox *SequencerInbox,
-	exec execution.ExecutionSequencer,
 ) (*InboxTracker, *InboxReader, error) {
-	inboxTracker, err := NewInboxTracker(consensusDB, txStreamer, dapReaders, config.SnapSyncTest)
+	inboxTracker, err := NewInboxTracker(consensusDB, txStreamer, dapReaders)
 	if err != nil {
 		return nil, nil, err
 	}
 	firstMessageBlock := new(big.Int).SetUint64(deployInfo.DeployedAt)
-	if config.SnapSyncTest.Enabled {
-		if exec == nil {
-			return nil, nil, errors.New("snap sync test requires an execution sequencer")
-		}
-
-		batchCount := config.SnapSyncTest.BatchCount
-		delayedMessageNumber, err := exec.NextDelayedMessageNumber()
-		if err != nil {
-			return nil, nil, err
-		}
-		if batchCount > delayedMessageNumber {
-			batchCount = delayedMessageNumber
-		}
-		// Find the first block containing the batch count.
-		// Subtract 1 to get the block before the needed batch count,
-		// this is done to fetch previous batch metadata needed for snap sync.
-		if batchCount > 0 {
-			batchCount--
-		}
-		block, err := FindBlockContainingBatchCount(ctx, deployInfo.Bridge, l1client, config.SnapSyncTest.ParentChainAssertionBlock, batchCount)
-		if err != nil {
-			return nil, nil, err
-		}
-		firstMessageBlock.SetUint64(block)
-	}
 	inboxReader, err := NewInboxReader(inboxTracker, l1client, l1Reader, firstMessageBlock, delayedBridge, sequencerInbox, func() *InboxReaderConfig { return &configFetcher.Get().InboxReader })
 	if err != nil {
 		return nil, nil, err
@@ -927,7 +878,7 @@ func getTransactionStreamer(
 	fatalErrChan chan error,
 ) (*TransactionStreamer, error) {
 	transactionStreamerConfigFetcher := func() *TransactionStreamerConfig { return &configFetcher.Get().TransactionStreamer }
-	txStreamer, err := NewTransactionStreamer(ctx, consensusDB, l2Config, exec, broadcastServer, fatalErrChan, transactionStreamerConfigFetcher, &configFetcher.Get().SnapSyncTest)
+	txStreamer, err := NewTransactionStreamer(ctx, consensusDB, l2Config, exec, broadcastServer, fatalErrChan, transactionStreamerConfigFetcher)
 	if err != nil {
 		return nil, err
 	}
@@ -1224,7 +1175,7 @@ func createNodeImpl(
 		return nil, err
 	}
 
-	inboxTracker, inboxReader, err := getInboxTrackerAndReader(ctx, consensusDB, txStreamer, dapRegistry, config, configFetcher, l1client, l1Reader, deployInfo, delayedBridge, sequencerInbox, executionSequencer)
+	inboxTracker, inboxReader, err := getInboxTrackerAndReader(consensusDB, txStreamer, dapRegistry, configFetcher, l1client, l1Reader, deployInfo, delayedBridge, sequencerInbox)
 	if err != nil {
 		return nil, err
 	}
@@ -1290,53 +1241,6 @@ func createNodeImpl(
 	}, nil
 }
 
-func FindBlockContainingBatchCount(ctx context.Context, bridgeAddress common.Address, l1Client *ethclient.Client, parentChainAssertionBlock uint64, batchCount uint64) (uint64, error) {
-	bridge, err := bridgegen.NewIBridge(bridgeAddress, l1Client)
-	if err != nil {
-		return 0, err
-	}
-	high := parentChainAssertionBlock
-	low := uint64(0)
-	reduceBy := uint64(100)
-	if high > reduceBy {
-		low = high - reduceBy
-	}
-	// Reduce high and low by 100 until lowNode.InboxMaxCount < batchCount
-	// This will give us a range (low to high) of blocks that contain the batch count.
-	for low > 0 {
-		lowCount, err := bridge.SequencerMessageCount(&bind.CallOpts{Context: ctx, BlockNumber: new(big.Int).SetUint64(low)})
-		if err != nil {
-			return 0, err
-		}
-		if lowCount.Uint64() > batchCount {
-			high = low
-			reduceBy = reduceBy * 2
-			if low > reduceBy {
-				low = low - reduceBy
-			} else {
-				low = 0
-			}
-		} else {
-			break
-		}
-	}
-	// Then binary search between low and high to find the block containing the batch count.
-	for low < high {
-		mid := low + (high-low)/2
-
-		midCount, err := bridge.SequencerMessageCount(&bind.CallOpts{Context: ctx, BlockNumber: new(big.Int).SetUint64(mid)})
-		if err != nil {
-			return 0, err
-		}
-		if midCount.Uint64() < batchCount {
-			low = mid + 1
-		} else {
-			high = mid
-		}
-	}
-	return low, nil
-}
-
 func (n *Node) OnConfigReload(_ *Config, _ *Config) error {
 	// TODO
 	return nil

arbnode/transaction_streamer.go

@@ -54,10 +54,9 @@ type TransactionStreamer struct {
 	prevHeadMsgIdx *arbutil.MessageIndex
 	validator      *staker.BlockValidator
 
-	db             ethdb.Database
-	fatalErrChan   chan<- error
-	config         TransactionStreamerConfigFetcher
-	snapSyncConfig *SnapSyncConfig
+	db           ethdb.Database
+	fatalErrChan chan<- error
+	config       TransactionStreamerConfigFetcher
 
 	insertionMutex     sync.Mutex // cannot be acquired while reorgMutex is held
 	reorgMutex         sync.RWMutex
@@ -124,7 +123,6 @@ func NewTransactionStreamer(
 	broadcastServer *broadcaster.Broadcaster,
 	fatalErrChan chan<- error,
 	config TransactionStreamerConfigFetcher,
-	snapSyncConfig *SnapSyncConfig,
 ) (*TransactionStreamer, error) {
 	streamer := &TransactionStreamer{
 		exec:               exec,
@@ -134,7 +132,6 @@ func NewTransactionStreamer(
 		broadcastServer:    broadcastServer,
 		fatalErrChan:       fatalErrChan,
 		config:             config,
-		snapSyncConfig:     snapSyncConfig,
 	}
 	err := streamer.cleanupInconsistentState()
 	if err != nil {
@@ -794,9 +791,6 @@ func (s *TransactionStreamer) AddMessagesAndEndBatch(firstMsgIdx arbutil.Message
 }
 
 func (s *TransactionStreamer) getPrevPrevDelayedRead(msgIdx arbutil.MessageIndex) (uint64, error) {
-	if s.snapSyncConfig.Enabled && uint64(msgIdx) == s.snapSyncConfig.PrevBatchMessageCount {
-		return s.snapSyncConfig.PrevDelayedRead, nil
-	}
 	var prevDelayedRead uint64
 	if msgIdx > 0 {
 		prevMsg, err := s.GetMessage(msgIdx - 1)

changelog/diegoximenes-nit-4565.md

@@ -0,0 +1,2 @@
+### Removed
+- Remove aborted snap sync code

changelog/jeongkyun-oh-nit-4431.md

@@ -0,0 +1,2 @@
+### Fixed
+ - Use defer to release createBlocksMutex in sequencerWrapper to prevent deadlock on panic

changelog/mkulawik-nit-4138.md

@@ -0,0 +1,2 @@
+### Fixed
+ - Fix opening classic-msg database

changelog/mkulawik-nit-4530.md

@@ -0,0 +1,2 @@
+### Fixed
+- Fix system test triggered panic in `updateFilterMapsHeads`