refactor aggregator/nonaggregator loop start

block/manager.go

@@ -151,31 +151,24 @@ func (m *Manager) Start(ctx context.Context) error {
 		}
 	}
 
+	//Fullnode loop can start before syncing from DA
 	if !isSequencer {
-		go uevent.MustSubscribe(ctx, m.Pubsub, "applyGossipedBlocksLoop", p2p.EventQueryNewNewGossipedBlock, m.onNewGossipedBlock, m.logger)
+		m.startNonSequencerLoop(ctx)
 	}
 
+	// TODO: populate the accumulatedSize on startup
+	//Wait till DA is up and running
+	<-m.DAClient.Started()
+	//Start syncing from DA
 	err = m.syncBlockManager()
 	if err != nil {
 		return fmt.Errorf("sync block manager: %w", err)
 	}
 
+	//Aggregator producer loop is started only when finished syncing from DA
 	if isSequencer {
-		// TODO: populate the accumulatedSize on startup
-		//Wait till DA is up and running
-		<-m.DAClient.Started()
-		err = m.syncBlockManager()
-		if err != nil {
-			return fmt.Errorf("sync block manager: %w", err)
-		}
-		go m.ProduceBlockLoop(ctx)
-		go m.SubmitLoop(ctx)
-	} else {
-		go uevent.MustSubscribe(ctx, m.Pubsub, "applyGossipedBlocksLoop", p2p.EventQueryNewNewGossipedBlock, m.onNewGossipedBlock, m.logger)
-		go m.RetrieveLoop(ctx)
-		go m.SyncTargetLoop(ctx)
+		m.startSequencerLoop(ctx)
 	}
-
 	return nil
 }
 
@@ -224,15 +217,13 @@ func (m *Manager) onNewGossipedBlock(event pubsub.Message) {
 	eventData := event.Data().(p2p.GossipedBlock)
 	block := eventData.Block
 	commit := eventData.Commit
-	m.retrieverMutex.Lock() // needed to protect blockCache access
+
 	_, found := m.blockCache[block.Header.Height]
-	// It is not strictly necessary to return early, for correctness, but doing so helps us avoid mutex pressure and unnecessary repeated attempts to apply cached blocks
-	if found {
+	if found || block.Header.Height < m.Store.NextHeight() {
 		m.retrieverMutex.Unlock()
 		return
 	}
-
-	m.logger.Debug("Received new block via gossip", "block height", block.Header.Height, "store height", m.Store.Height(), "n cachedBlocks", len(m.blockCache))
+	m.logger.Debug("Received new block via gossip", "height", block.Header.Height, "n cachedBlocks", len(m.blockCache))
 
 	nextHeight := m.Store.NextHeight()
 	if block.Header.Height >= nextHeight {
@@ -259,3 +250,14 @@ func getInitialState(store store.Store, genesis *tmtypes.GenesisDoc, logger type
 
 	return s, err
 }
+
+func (m *Manager) startNonSequencerLoop(ctx context.Context) {
+	go uevent.MustSubscribe(ctx, m.Pubsub, "applyGossipedBlocksLoop", p2p.EventQueryNewNewGossipedBlock, m.onNewGossipedBlock, m.logger)
+	go m.RetrieveLoop(ctx)
+	go m.SyncTargetLoop(ctx)
+}
+
+func (m *Manager) startSequencerLoop(ctx context.Context) {
+	go m.ProduceBlockLoop(ctx)
+	go m.SubmitLoop(ctx)
+}