[NIT-2845] Add a nitro option to stop syncing at a given block number

arbnode/delayed_seq_reorg_test.go

@@ -22,9 +22,9 @@ func TestSequencerReorgFromDelayed(t *testing.T) {
 	tracker, err := NewInboxTracker(db, streamer, nil, DefaultSnapSyncConfig)
 	Require(t, err)
 
-	err = streamer.Start(ctx)
+	err = streamer.Start(ctx, 0)
 	Require(t, err)
-	exec.Start(ctx)
+	exec.Start(ctx, 0)
 	init, err := streamer.GetMessage(0)
 	Require(t, err)
 

arbnode/delayed_sequencer.go

@@ -211,11 +211,20 @@ func (d *DelayedSequencer) ForceSequenceDelayed(ctx context.Context) error {
 	return d.sequenceWithoutLockout(ctx, lastBlockHeader)
 }
 
-func (d *DelayedSequencer) run(ctx context.Context) {
+func (d *DelayedSequencer) run(ctx context.Context, syncTillBlock uint64) {
 	headerChan, cancel := d.l1Reader.Subscribe(false)
 	defer cancel()
 
 	for {
+		delayedCount, err := d.inbox.GetDelayedCount()
+		if err != nil {
+			log.Warn("error reading delayed count", "err", err)
+			continue
+		}
+		if syncTillBlock > 0 && delayedCount >= syncTillBlock {
+			log.Info("stopping block creation in delayed sequencer", "syncTillBlock", syncTillBlock)
+			return
+		}
 		select {
 		case nextHeader, ok := <-headerChan:
 			if !ok {
@@ -232,7 +241,9 @@ func (d *DelayedSequencer) run(ctx context.Context) {
 	}
 }
 
-func (d *DelayedSequencer) Start(ctxIn context.Context) {
+func (d *DelayedSequencer) Start(ctxIn context.Context, syncTillBlock uint64) {
 	d.StopWaiter.Start(ctxIn, d)
-	d.LaunchThread(d.run)
+	d.LaunchThread(func(ctx context.Context) {
+		d.run(ctx, syncTillBlock)
+	})
 }

arbnode/inbox_reader.go

@@ -119,18 +119,36 @@ func NewInboxReader(tracker *InboxTracker, client *ethclient.Client, l1Reader *h
 	}, nil
 }
 
-func (r *InboxReader) Start(ctxIn context.Context) error {
+func (r *InboxReader) Start(ctxIn context.Context, syncTillBlock uint64) error {
 	r.StopWaiter.Start(ctxIn, r)
 	hadError := false
-	r.CallIteratively(func(ctx context.Context) time.Duration {
-		err := r.run(ctx, hadError)
-		if err != nil && !errors.Is(err, context.Canceled) && !strings.Contains(err.Error(), "header not found") {
-			log.Warn("error reading inbox", "err", err)
-			hadError = true
-		} else {
-			hadError = false
+	r.LaunchThread(func(ctx context.Context) {
+		for {
+			delayedCount, err := r.tracker.GetDelayedCount()
+			if err != nil {
+				log.Warn("error reading delayed count", "err", err)
+				hadError = true
+			}
+			if syncTillBlock > 0 && delayedCount >= syncTillBlock {
+				log.Info("stopping block creation in inbox reader", "syncTillBlock", syncTillBlock)
+				return
+			}
+			err = r.run(ctx, hadError)
+			if err != nil && !errors.Is(err, context.Canceled) && !strings.Contains(err.Error(), "header not found") {
+				log.Warn("error reading inbox", "err", err)
+				hadError = true
+			} else {
+				hadError = false
+			}
+			interval := time.Second
+			timer := time.NewTimer(interval)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return
+			case <-timer.C:
+			}
 		}
-		return time.Second
 	})
 
 	// Ensure we read the init message before other things start up

arbnode/inbox_test.go

@@ -106,9 +106,9 @@ func TestTransactionStreamer(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	err := inbox.Start(ctx)
+	err := inbox.Start(ctx, 0)
 	Require(t, err)
-	exec.Start(ctx)
+	exec.Start(ctx, 0)
 
 	maxExpectedGasCost := big.NewInt(l2pricing.InitialBaseFeeWei)
 	maxExpectedGasCost.Mul(maxExpectedGasCost, big.NewInt(2100*2))

arbnode/node.go

@@ -79,6 +79,7 @@ func GenerateRollupConfig(prod bool, wasmModuleRoot common.Hash, rollupOwner com
 
 type Config struct {
 	Sequencer           bool                        `koanf:"sequencer"`
+	SyncTillBlock       uint64                      `koanf:"sync-till-block"`
 	ParentChainReader   headerreader.Config         `koanf:"parent-chain-reader" reload:"hot"`
 	InboxReader         InboxReaderConfig           `koanf:"inbox-reader" reload:"hot"`
 	DelayedSequencer    DelayedSequencerConfig      `koanf:"delayed-sequencer" reload:"hot"`
@@ -145,6 +146,7 @@ func (c *Config) ValidatorRequired() bool {
 
 func ConfigAddOptions(prefix string, f *flag.FlagSet, feedInputEnable bool, feedOutputEnable bool) {
 	f.Bool(prefix+".sequencer", ConfigDefault.Sequencer, "enable sequencer")
+	f.Uint64(prefix+".sync-till-block", ConfigDefault.SyncTillBlock, "sync till block")
 	headerreader.AddOptions(prefix+".parent-chain-reader", f)
 	InboxReaderConfigAddOptions(prefix+".inbox-reader", f)
 	DelayedSequencerConfigAddOptions(prefix+".delayed-sequencer", f)
@@ -163,6 +165,7 @@ func ConfigAddOptions(prefix string, f *flag.FlagSet, feedInputEnable bool, feed
 
 var ConfigDefault = Config{
 	Sequencer:           false,
+	SyncTillBlock:       0,
 	ParentChainReader:   headerreader.DefaultConfig,
 	InboxReader:         DefaultInboxReaderConfig,
 	DelayedSequencer:    DefaultDelayedSequencerConfig,
@@ -839,7 +842,7 @@ func (n *Node) Start(ctx context.Context) error {
 	if execClient != nil {
 		execClient.SetConsensusClient(n)
 	}
-	err = n.Execution.Start(ctx)
+	err = n.Execution.Start(ctx, n.configFetcher.Get().SyncTillBlock)
 	if err != nil {
 		return fmt.Errorf("error starting exec client: %w", err)
 	}
@@ -869,12 +872,12 @@ func (n *Node) Start(ctx context.Context) error {
 			return fmt.Errorf("error populating feed backlog on startup: %w", err)
 		}
 	}
-	err = n.TxStreamer.Start(ctx)
+	err = n.TxStreamer.Start(ctx, n.configFetcher.Get().SyncTillBlock)
 	if err != nil {
 		return fmt.Errorf("error starting transaction streamer: %w", err)
 	}
 	if n.InboxReader != nil {
-		err = n.InboxReader.Start(ctx)
+		err = n.InboxReader.Start(ctx, n.configFetcher.Get().SyncTillBlock)
 		if err != nil {
 			return fmt.Errorf("error starting inbox reader: %w", err)
 		}
@@ -887,15 +890,15 @@ func (n *Node) Start(ctx context.Context) error {
 		}
 	}
 	if n.SeqCoordinator != nil {
-		n.SeqCoordinator.Start(ctx)
+		n.SeqCoordinator.Start(ctx, n.configFetcher.Get().SyncTillBlock)
 	} else {
 		n.Execution.Activate()
 	}
 	if n.MaintenanceRunner != nil {
 		n.MaintenanceRunner.Start(ctx)
 	}
 	if n.DelayedSequencer != nil {
-		n.DelayedSequencer.Start(ctx)
+		n.DelayedSequencer.Start(ctx, n.configFetcher.Get().SyncTillBlock)
 	}
 	if n.BatchPoster != nil {
 		n.BatchPoster.Start(ctx)
@@ -945,7 +948,7 @@ func (n *Node) Start(ctx context.Context) error {
 					return
 				}
 			}
-			n.BroadcastClients.Start(ctx)
+			n.BroadcastClients.Start(ctx, n.configFetcher.Get().SyncTillBlock)
 		}()
 	}
 	if n.configFetcher != nil {

arbnode/seq_coordinator.go

@@ -854,7 +854,7 @@ func (c *SeqCoordinator) launchHealthcheckServer(ctx context.Context) {
 	}
 }
 
-func (c *SeqCoordinator) Start(ctxIn context.Context) {
+func (c *SeqCoordinator) Start(ctxIn context.Context, syncTillBlock uint64) {
 	c.StopWaiter.Start(ctxIn, c)
 	var newRedisCoordinator *redisutil.RedisCoordinator
 	if c.config.NewRedisUrl != "" {
@@ -865,7 +865,33 @@ func (c *SeqCoordinator) Start(ctxIn context.Context) {
 				err, "newRedisUrl", c.config.NewRedisUrl)
 		}
 	}
-	c.CallIteratively(func(ctx context.Context) time.Duration { return c.chooseRedisAndUpdate(ctx, newRedisCoordinator) })
+
+	c.LaunchThread(func(ctx context.Context) {
+		for {
+			count, err := c.streamer.GetMessageCount()
+			if err != nil {
+				log.Warn("failed to get message count", "err", err)
+			}
+			if syncTillBlock > 0 && uint64(count) >= syncTillBlock {
+				log.Info("stopping block creation in sequencer", "syncTillBlock", syncTillBlock)
+				return
+			}
+			interval := c.chooseRedisAndUpdate(ctx, newRedisCoordinator)
+			if ctx.Err() != nil {
+				return
+			}
+			if interval == time.Duration(0) {
+				continue
+			}
+			timer := time.NewTimer(interval)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return
+			case <-timer.C:
+			}
+		}
+	})
 	if c.config.ChosenHealthcheckAddr != "" {
 		c.StopWaiter.LaunchThread(c.launchHealthcheckServer)
 	}

arbnode/transaction_streamer.go

@@ -1218,7 +1218,32 @@ func (s *TransactionStreamer) executeMessages(ctx context.Context, ignored struc
 	return s.config().ExecuteMessageLoopDelay
 }
 
-func (s *TransactionStreamer) Start(ctxIn context.Context) error {
+func (s *TransactionStreamer) Start(ctxIn context.Context, syncTillBlock uint64) error {
 	s.StopWaiter.Start(ctxIn, s)
-	return stopwaiter.CallIterativelyWith[struct{}](&s.StopWaiterSafe, s.executeMessages, s.newMessageNotifier)
+	return s.LaunchThreadSafe(func(ctx context.Context) {
+		var defaultVal struct{}
+		var val struct{}
+		for {
+			if syncTillBlock > 0 && uint64(s.execLastMsgCount) >= syncTillBlock {
+				log.Info("stopping block creation in transaction streamer", "syncTillBlock", syncTillBlock)
+				return
+			}
+			interval := s.executeMessages(ctx, val)
+			if ctx.Err() != nil {
+				return
+			}
+			val = defaultVal
+			if interval == time.Duration(0) {
+				continue
+			}
+			timer := time.NewTimer(interval)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return
+			case <-timer.C:
+			case val = <-s.newMessageNotifier:
+			}
+		}
+	})
 }

broadcastclient/broadcastclient.go

@@ -176,7 +176,7 @@ func NewBroadcastClient(
 	}, err
 }
 
-func (bc *BroadcastClient) Start(ctxIn context.Context) {
+func (bc *BroadcastClient) Start(ctxIn context.Context, syncTillBlock uint64) {
 	bc.StopWaiter.Start(ctxIn, bc)
 	if bc.StopWaiter.Stopped() {
 		log.Info("broadcast client has already been stopped, not starting")
@@ -185,6 +185,11 @@ func (bc *BroadcastClient) Start(ctxIn context.Context) {
 	bc.LaunchThread(func(ctx context.Context) {
 		backoffDuration := bc.config().ReconnectInitialBackoff
 		for {
+
+			if syncTillBlock > 0 && uint64(bc.nextSeqNum) >= syncTillBlock {
+				log.Info("stopping block creation in broadcast client", "syncTillBlock", syncTillBlock)
+				return
+			}
 			earlyFrameData, err := bc.connect(ctx, bc.nextSeqNum)
 			if errors.Is(err, ErrMissingChainId) ||
 				errors.Is(err, ErrIncorrectChainId) ||

broadcastclient/broadcastclient_test.go

@@ -153,7 +153,7 @@ func TestInvalidSignature(t *testing.T) {
 		&badSequencerAddr,
 	)
 	Require(t, err)
-	broadcastClient.Start(ctx)
+	broadcastClient.Start(ctx, 0)
 
 	go func() {
 		for i := 0; i < messageCount; i++ {
@@ -227,7 +227,7 @@ func startMakeBroadcastClient(ctx context.Context, t *testing.T, clientConfig Co
 		sequencerAddr,
 	)
 	Require(t, err)
-	broadcastClient.Start(ctx)
+	broadcastClient.Start(ctx, 0)
 	messageCount := 0
 
 	wg.Add(1)
@@ -315,7 +315,7 @@ func TestServerClientDisconnect(t *testing.T) {
 		&sequencerAddr,
 	)
 	Require(t, err)
-	broadcastClient.Start(ctx)
+	broadcastClient.Start(ctx, 0)
 
 	t.Log("broadcasting seq 0 message")
 	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil))
@@ -386,7 +386,7 @@ func TestBroadcastClientConfirmedMessage(t *testing.T) {
 		&sequencerAddr,
 	)
 	Require(t, err)
-	broadcastClient.Start(ctx)
+	broadcastClient.Start(ctx, 0)
 
 	t.Log("broadcasting seq 0 message")
 	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil))
@@ -458,7 +458,7 @@ func TestServerIncorrectChainId(t *testing.T) {
 		&sequencerAddr,
 	)
 	Require(t, err)
-	badBroadcastClient.Start(ctx)
+	badBroadcastClient.Start(ctx, 0)
 	badTimer := time.NewTimer(5 * time.Second)
 	select {
 	case err := <-feedErrChan:
@@ -517,7 +517,7 @@ func TestServerMissingChainId(t *testing.T) {
 		&sequencerAddr,
 	)
 	Require(t, err)
-	badBroadcastClient.Start(ctx)
+	badBroadcastClient.Start(ctx, 0)
 	badTimer := time.NewTimer(5 * time.Second)
 	select {
 	case err := <-feedErrChan:
@@ -574,7 +574,7 @@ func TestServerIncorrectFeedServerVersion(t *testing.T) {
 		&sequencerAddr,
 	)
 	Require(t, err)
-	badBroadcastClient.Start(ctx)
+	badBroadcastClient.Start(ctx, 0)
 	badTimer := time.NewTimer(5 * time.Second)
 	select {
 	case err := <-feedErrChan:
@@ -633,7 +633,7 @@ func TestServerMissingFeedServerVersion(t *testing.T) {
 		&sequencerAddr,
 	)
 	Require(t, err)
-	badBroadcastClient.Start(ctx)
+	badBroadcastClient.Start(ctx, 0)
 	badTimer := time.NewTimer(5 * time.Second)
 	select {
 	case err := <-feedErrChan:
@@ -684,7 +684,7 @@ func TestBroadcastClientReconnectsOnServerDisconnect(t *testing.T) {
 		&sequencerAddr,
 	)
 	Require(t, err)
-	broadcastClient.Start(ctx)
+	broadcastClient.Start(ctx, 0)
 	defer broadcastClient.StopAndWait()
 
 	// Client set to timeout connection at 200 milliseconds, and server set to send ping every 50 seconds,
@@ -796,7 +796,7 @@ func connectAndGetCachedMessages(ctx context.Context, addr net.Addr, chainId uin
 		sequencerAddr,
 	)
 	Require(t, err)
-	broadcastClient.Start(ctx)
+	broadcastClient.Start(ctx, 0)
 
 	go func() {
 		defer wg.Done()