broadcast messages after producing a block and not after writing them…

… in the db

arbnode/node.go

@@ -997,6 +997,10 @@ func (n *Node) GetFinalizedMsgCount(ctx context.Context) (arbutil.MessageIndex,
 	return n.InboxReader.GetFinalizedMsgCount(ctx)
 }
 
+func (n *Node) BroadcastMessage(msg arbostypes.MessageWithMetadata, pos arbutil.MessageIndex) {
+	n.TxStreamer.BroadcastMessage(msg, pos)
+}
+
 func (n *Node) WriteMessageFromSequencer(pos arbutil.MessageIndex, msgWithMeta arbostypes.MessageWithMetadata) error {
 	return n.TxStreamer.WriteMessageFromSequencer(pos, msgWithMeta)
 }

arbnode/transaction_streamer.go

@@ -1000,6 +1000,7 @@ func (s *TransactionStreamer) WriteMessageFromSequencer(pos arbutil.MessageIndex
 	if err := s.writeMessages(pos, []arbostypes.MessageWithMetadata{msgWithMeta}, nil); err != nil {
 		return err
 	}
+	s.BroadcastMessage(msgWithMeta, pos)
 
 	return nil
 }
@@ -1029,6 +1030,15 @@ func (s *TransactionStreamer) writeMessage(pos arbutil.MessageIndex, msg arbosty
 	return batch.Put(key, msgBytes)
 }
 
+func (s *TransactionStreamer) BroadcastMessage(msg arbostypes.MessageWithMetadata, pos arbutil.MessageIndex) {
+	if s.broadcastServer == nil {
+		return
+	}
+	if err := s.broadcastServer.BroadcastSingle(msg, pos); err != nil {
+		log.Error("failed broadcasting message", "pos", pos, "err", err)
+	}
+}
+
 // The mutex must be held, and pos must be the latest message count.
 // `batch` may be nil, which initializes a new batch. The batch is closed out in this function.
 func (s *TransactionStreamer) writeMessages(pos arbutil.MessageIndex, messages []arbostypes.MessageWithMetadata, batch ethdb.Batch) error {
@@ -1056,12 +1066,6 @@ func (s *TransactionStreamer) writeMessages(pos arbutil.MessageIndex, messages [
 	default:
 	}
 
-	if s.broadcastServer != nil {
-		if err := s.broadcastServer.BroadcastMessages(messages, pos); err != nil {
-			log.Error("failed broadcasting message", "pos", pos, "err", err)
-		}
-	}
-
 	return nil
 }
 

broadcaster/broadcaster.go

@@ -82,27 +82,6 @@ func (b *Broadcaster) BroadcastSingleFeedMessage(bfm *m.BroadcastFeedMessage) {
 	b.BroadcastFeedMessages(broadcastFeedMessages)
 }
 
-func (b *Broadcaster) BroadcastMessages(messages []arbostypes.MessageWithMetadata, seq arbutil.MessageIndex) (err error) {
-	defer func() {
-		if r := recover(); r != nil {
-			log.Error("recovered error in BroadcastMessages", "recover", r, "backtrace", string(debug.Stack()))
-			err = errors.New("panic in BroadcastMessages")
-		}
-	}()
-	var feedMessages []*m.BroadcastFeedMessage
-	for i, msg := range messages {
-		bfm, err := b.NewBroadcastFeedMessage(msg, seq+arbutil.MessageIndex(i))
-		if err != nil {
-			return err
-		}
-		feedMessages = append(feedMessages, bfm)
-	}
-
-	b.BroadcastFeedMessages(feedMessages)
-
-	return nil
-}
-
 func (b *Broadcaster) BroadcastFeedMessages(messages []*m.BroadcastFeedMessage) {
 
 	bm := &m.BroadcastMessage{

execution/gethexec/executionengine.go

@@ -630,6 +630,14 @@ func (s *ExecutionEngine) digestMessageWithBlockMutex(num arbutil.MessageIndex,
 	if err != nil {
 		return err
 	}
+
+	if s.consensus != nil {
+		l2BlockHash := block.Hash()
+		msg.L2BlockHash = &l2BlockHash
+
+		s.consensus.BroadcastMessage(*msg, num)
+	}
+
 	err = s.appendBlock(block, statedb, receipts, time.Since(startTime))
 	if err != nil {
 		return err

execution/interface.go

@@ -90,6 +90,7 @@ type ConsensusInfo interface {
 }
 
 type ConsensusSequencer interface {
+	BroadcastMessage(msg arbostypes.MessageWithMetadata, pos arbutil.MessageIndex)
 	WriteMessageFromSequencer(pos arbutil.MessageIndex, msgWithMeta arbostypes.MessageWithMetadata) error
 	ExpectChosenSequencer() error
 	CacheL1PriceDataOfMsg(pos arbutil.MessageIndex, callDataUnits uint64, l1GasCharged uint64)