Add timeboosted information to broadcast feed

arbnode/inbox_tracker.go

@@ -307,7 +307,12 @@ func (t *InboxTracker) PopulateFeedBacklog(broadcastServer *broadcaster.Broadcas
 			blockHash = &msgResult.BlockHash
 		}
 
-		feedMessage, err := broadcastServer.NewBroadcastFeedMessage(*message, seqNum, blockHash)
+		blockMetadata, err := t.txStreamer.BlockMetadataAtCount(seqNum + 1)
+		if err != nil {
+			log.Warn("Error getting blockMetadata byte array from tx streamer", "err", err)
+		}
+
+		feedMessage, err := broadcastServer.NewBroadcastFeedMessage(*message, seqNum, blockHash, blockMetadata)
 		if err != nil {
 			return fmt.Errorf("error creating broadcast feed message %v: %w", seqNum, err)
 		}

arbnode/node.go

@@ -1046,8 +1046,8 @@ func (n *Node) GetFinalizedMsgCount(ctx context.Context) (arbutil.MessageIndex,
 	return n.InboxReader.GetFinalizedMsgCount(ctx)
 }
 
-func (n *Node) WriteMessageFromSequencer(pos arbutil.MessageIndex, msgWithMeta arbostypes.MessageWithMetadata, msgResult execution.MessageResult) error {
-	return n.TxStreamer.WriteMessageFromSequencer(pos, msgWithMeta, msgResult)
+func (n *Node) WriteMessageFromSequencer(pos arbutil.MessageIndex, msgWithMeta arbostypes.MessageWithMetadata, msgResult execution.MessageResult, blockMetadata arbostypes.BlockMetadata) error {
+	return n.TxStreamer.WriteMessageFromSequencer(pos, msgWithMeta, msgResult, blockMetadata)
 }
 
 func (n *Node) ExpectChosenSequencer() error {

arbnode/schema.go

@@ -6,6 +6,7 @@ package arbnode
 var (
 	messagePrefix                []byte = []byte("m") // maps a message sequence number to a message
 	blockHashInputFeedPrefix     []byte = []byte("b") // maps a message sequence number to a block hash received through the input feed
+	blockMetadataInputFeedPrefix []byte = []byte("t") // maps a message sequence number to a blockMetaData byte array received through the input feed
 	messageResultPrefix          []byte = []byte("r") // maps a message sequence number to a message result
 	legacyDelayedMessagePrefix   []byte = []byte("d") // maps a delayed sequence number to an accumulator and a message as serialized on L1
 	rlpDelayedMessagePrefix      []byte = []byte("e") // maps a delayed sequence number to an accumulator and an RLP encoded message

arbnode/transaction_streamer.go

@@ -60,7 +60,7 @@ type TransactionStreamer struct {
 
 	nextAllowedFeedReorgLog time.Time
 
-	broadcasterQueuedMessages            []arbostypes.MessageWithMetadataAndBlockHash
+	broadcasterQueuedMessages            []arbostypes.MessageWithMetadataAndBlockInfo
 	broadcasterQueuedMessagesPos         atomic.Uint64
 	broadcasterQueuedMessagesActiveReorg bool
 
@@ -264,7 +264,7 @@ func deleteFromRange(ctx context.Context, db ethdb.Database, prefix []byte, star
 
 // The insertion mutex must be held. This acquires the reorg mutex.
 // Note: oldMessages will be empty if reorgHook is nil
-func (s *TransactionStreamer) reorg(batch ethdb.Batch, count arbutil.MessageIndex, newMessages []arbostypes.MessageWithMetadataAndBlockHash) error {
+func (s *TransactionStreamer) reorg(batch ethdb.Batch, count arbutil.MessageIndex, newMessages []arbostypes.MessageWithMetadataAndBlockInfo) error {
 	if count == 0 {
 		return errors.New("cannot reorg out init message")
 	}
@@ -359,9 +359,9 @@ func (s *TransactionStreamer) reorg(batch ethdb.Batch, count arbutil.MessageInde
 		return err
 	}
 
-	messagesWithComputedBlockHash := make([]arbostypes.MessageWithMetadataAndBlockHash, 0, len(messagesResults))
+	messagesWithComputedBlockHash := make([]arbostypes.MessageWithMetadataAndBlockInfo, 0, len(messagesResults))
 	for i := 0; i < len(messagesResults); i++ {
-		messagesWithComputedBlockHash = append(messagesWithComputedBlockHash, arbostypes.MessageWithMetadataAndBlockHash{
+		messagesWithComputedBlockHash = append(messagesWithComputedBlockHash, arbostypes.MessageWithMetadataAndBlockInfo{
 			MessageWithMeta: newMessages[i].MessageWithMeta,
 			BlockHash:       &messagesResults[i].BlockHash,
 		})
@@ -383,6 +383,10 @@ func (s *TransactionStreamer) reorg(batch ethdb.Batch, count arbutil.MessageInde
 	if err != nil {
 		return err
 	}
+	err = deleteStartingAt(s.db, batch, blockMetadataInputFeedPrefix, uint64ToKey(uint64(count)))
+	if err != nil {
+		return err
+	}
 	err = deleteStartingAt(s.db, batch, messagePrefix, uint64ToKey(uint64(count)))
 	if err != nil {
 		return err
@@ -449,7 +453,7 @@ func (s *TransactionStreamer) GetMessage(seqNum arbutil.MessageIndex) (*arbostyp
 	return &message, nil
 }
 
-func (s *TransactionStreamer) getMessageWithMetadataAndBlockHash(seqNum arbutil.MessageIndex) (*arbostypes.MessageWithMetadataAndBlockHash, error) {
+func (s *TransactionStreamer) getMessageWithMetadataAndBlockInfo(seqNum arbutil.MessageIndex) (*arbostypes.MessageWithMetadataAndBlockInfo, error) {
 	msg, err := s.GetMessage(seqNum)
 	if err != nil {
 		return nil, err
@@ -472,11 +476,18 @@ func (s *TransactionStreamer) getMessageWithMetadataAndBlockHash(seqNum arbutil.
 		return nil, err
 	}
 
-	msgWithBlockHash := arbostypes.MessageWithMetadataAndBlockHash{
+	key = dbKey(blockMetadataInputFeedPrefix, uint64(seqNum))
+	blockMetadata, err := s.db.Get(key)
+	if err != nil && !dbutil.IsErrNotFound(err) {
+		return nil, err
+	}
+
+	msgWithBlockInfo := arbostypes.MessageWithMetadataAndBlockInfo{
 		MessageWithMeta: *msg,
 		BlockHash:       blockHash,
+		BlockMetadata:   blockMetadata,
 	}
-	return &msgWithBlockHash, nil
+	return &msgWithBlockInfo, nil
 }
 
 // Note: if changed to acquire the mutex, some internal users may need to be updated to a non-locking version.
@@ -532,7 +543,7 @@ func (s *TransactionStreamer) AddBroadcastMessages(feedMessages []*m.BroadcastFe
 		return nil
 	}
 	broadcastStartPos := feedMessages[0].SequenceNumber
-	var messages []arbostypes.MessageWithMetadataAndBlockHash
+	var messages []arbostypes.MessageWithMetadataAndBlockInfo
 	broadcastAfterPos := broadcastStartPos
 	for _, feedMessage := range feedMessages {
 		if broadcastAfterPos != feedMessage.SequenceNumber {
@@ -541,11 +552,12 @@ func (s *TransactionStreamer) AddBroadcastMessages(feedMessages []*m.BroadcastFe
 		if feedMessage.Message.Message == nil || feedMessage.Message.Message.Header == nil {
 			return fmt.Errorf("invalid feed message at sequence number %v", feedMessage.SequenceNumber)
 		}
-		msgWithBlockHash := arbostypes.MessageWithMetadataAndBlockHash{
+		msgWithBlockInfo := arbostypes.MessageWithMetadataAndBlockInfo{
 			MessageWithMeta: feedMessage.Message,
 			BlockHash:       feedMessage.BlockHash,
+			BlockMetadata:   feedMessage.BlockMetadata,
 		}
-		messages = append(messages, msgWithBlockHash)
+		messages = append(messages, msgWithBlockInfo)
 		broadcastAfterPos++
 	}
 
@@ -666,9 +678,9 @@ func endBatch(batch ethdb.Batch) error {
 }
 
 func (s *TransactionStreamer) AddMessagesAndEndBatch(pos arbutil.MessageIndex, messagesAreConfirmed bool, messages []arbostypes.MessageWithMetadata, batch ethdb.Batch) error {
-	messagesWithBlockHash := make([]arbostypes.MessageWithMetadataAndBlockHash, 0, len(messages))
+	messagesWithBlockInfo := make([]arbostypes.MessageWithMetadataAndBlockInfo, 0, len(messages))
 	for _, message := range messages {
-		messagesWithBlockHash = append(messagesWithBlockHash, arbostypes.MessageWithMetadataAndBlockHash{
+		messagesWithBlockInfo = append(messagesWithBlockInfo, arbostypes.MessageWithMetadataAndBlockInfo{
 			MessageWithMeta: message,
 		})
 	}
@@ -677,7 +689,7 @@ func (s *TransactionStreamer) AddMessagesAndEndBatch(pos arbutil.MessageIndex, m
 		// Trim confirmed messages from l1pricedataCache
 		s.exec.MarkFeedStart(pos + arbutil.MessageIndex(len(messages)))
 		s.reorgMutex.RLock()
-		dups, _, _, err := s.countDuplicateMessages(pos, messagesWithBlockHash, nil)
+		dups, _, _, err := s.countDuplicateMessages(pos, messagesWithBlockInfo, nil)
 		s.reorgMutex.RUnlock()
 		if err != nil {
 			return err
@@ -694,7 +706,7 @@ func (s *TransactionStreamer) AddMessagesAndEndBatch(pos arbutil.MessageIndex, m
 	s.insertionMutex.Lock()
 	defer s.insertionMutex.Unlock()
 
-	return s.addMessagesAndEndBatchImpl(pos, messagesAreConfirmed, messagesWithBlockHash, batch)
+	return s.addMessagesAndEndBatchImpl(pos, messagesAreConfirmed, messagesWithBlockInfo, batch)
 }
 
 func (s *TransactionStreamer) getPrevPrevDelayedRead(pos arbutil.MessageIndex) (uint64, error) {
@@ -715,7 +727,7 @@ func (s *TransactionStreamer) getPrevPrevDelayedRead(pos arbutil.MessageIndex) (
 
 func (s *TransactionStreamer) countDuplicateMessages(
 	pos arbutil.MessageIndex,
-	messages []arbostypes.MessageWithMetadataAndBlockHash,
+	messages []arbostypes.MessageWithMetadataAndBlockInfo,
 	batch *ethdb.Batch,
 ) (uint64, bool, *arbostypes.MessageWithMetadata, error) {
 	var curMsg uint64
@@ -809,7 +821,7 @@ func (s *TransactionStreamer) logReorg(pos arbutil.MessageIndex, dbMsg *arbostyp
 
 }
 
-func (s *TransactionStreamer) addMessagesAndEndBatchImpl(messageStartPos arbutil.MessageIndex, messagesAreConfirmed bool, messages []arbostypes.MessageWithMetadataAndBlockHash, batch ethdb.Batch) error {
+func (s *TransactionStreamer) addMessagesAndEndBatchImpl(messageStartPos arbutil.MessageIndex, messagesAreConfirmed bool, messages []arbostypes.MessageWithMetadataAndBlockInfo, batch ethdb.Batch) error {
 	var confirmedReorg bool
 	var oldMsg *arbostypes.MessageWithMetadata
 	var lastDelayedRead uint64
@@ -953,6 +965,7 @@ func (s *TransactionStreamer) WriteMessageFromSequencer(
 	pos arbutil.MessageIndex,
 	msgWithMeta arbostypes.MessageWithMetadata,
 	msgResult execution.MessageResult,
+	blockMetadata arbostypes.BlockMetadata,
 ) error {
 	if err := s.ExpectChosenSequencer(); err != nil {
 		return err
@@ -977,15 +990,16 @@ func (s *TransactionStreamer) WriteMessageFromSequencer(
 		}
 	}
 
-	msgWithBlockHash := arbostypes.MessageWithMetadataAndBlockHash{
+	msgWithBlockInfo := arbostypes.MessageWithMetadataAndBlockInfo{
 		MessageWithMeta: msgWithMeta,
 		BlockHash:       &msgResult.BlockHash,
+		BlockMetadata:   blockMetadata,
 	}
 
-	if err := s.writeMessages(pos, []arbostypes.MessageWithMetadataAndBlockHash{msgWithBlockHash}, nil); err != nil {
+	if err := s.writeMessages(pos, []arbostypes.MessageWithMetadataAndBlockInfo{msgWithBlockInfo}, nil); err != nil {
 		return err
 	}
-	s.broadcastMessages([]arbostypes.MessageWithMetadataAndBlockHash{msgWithBlockHash}, pos)
+	s.broadcastMessages([]arbostypes.MessageWithMetadataAndBlockInfo{msgWithBlockInfo}, pos)
 
 	return nil
 }
@@ -1006,7 +1020,7 @@ func (s *TransactionStreamer) PopulateFeedBacklog() error {
 	return s.inboxReader.tracker.PopulateFeedBacklog(s.broadcastServer)
 }
 
-func (s *TransactionStreamer) writeMessage(pos arbutil.MessageIndex, msg arbostypes.MessageWithMetadataAndBlockHash, batch ethdb.Batch) error {
+func (s *TransactionStreamer) writeMessage(pos arbutil.MessageIndex, msg arbostypes.MessageWithMetadataAndBlockInfo, batch ethdb.Batch) error {
 	// write message with metadata
 	key := dbKey(messagePrefix, uint64(pos))
 	msgBytes, err := rlp.EncodeToBytes(msg.MessageWithMeta)
@@ -1026,11 +1040,22 @@ func (s *TransactionStreamer) writeMessage(pos arbutil.MessageIndex, msg arbosty
 	if err != nil {
 		return err
 	}
-	return batch.Put(key, msgBytes)
+	if err := batch.Put(key, msgBytes); err != nil {
+		return err
+	}
+
+	if msg.BlockMetadata != nil {
+		// Only store non-nil BlockMetadata to db. In case of a reorg, we dont have to explicitly
+		// clear out BlockMetadata of the reorged message, since those messages will be handled by s.reorg()
+		// This also allows update of BatchGasCost in message without mistakenly erasing BlockMetadata
+		key = dbKey(blockMetadataInputFeedPrefix, uint64(pos))
+		return batch.Put(key, msg.BlockMetadata)
+	}
+	return nil
 }
 
 func (s *TransactionStreamer) broadcastMessages(
-	msgs []arbostypes.MessageWithMetadataAndBlockHash,
+	msgs []arbostypes.MessageWithMetadataAndBlockInfo,
 	pos arbutil.MessageIndex,
 ) {
 	if s.broadcastServer == nil {
@@ -1043,7 +1068,7 @@ func (s *TransactionStreamer) broadcastMessages(
 
 // 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.MessageWithMetadataAndBlockHash, batch ethdb.Batch) error {
+func (s *TransactionStreamer) writeMessages(pos arbutil.MessageIndex, messages []arbostypes.MessageWithMetadataAndBlockInfo, batch ethdb.Batch) error {
 	if batch == nil {
 		batch = s.db.NewBatch()
 	}
@@ -1072,6 +1097,20 @@ func (s *TransactionStreamer) writeMessages(pos arbutil.MessageIndex, messages [
 	return nil
 }
 
+func (s *TransactionStreamer) BlockMetadataAtCount(count arbutil.MessageIndex) (arbostypes.BlockMetadata, error) {
+	if count == 0 {
+		return []byte{}, nil
+	}
+	pos := count - 1
+
+	key := dbKey(blockMetadataInputFeedPrefix, uint64(pos))
+	blockMetadata, err := s.db.Get(key)
+	if err != nil && !dbutil.IsErrNotFound(err) {
+		return nil, err
+	}
+	return blockMetadata, nil
+}
+
 func (s *TransactionStreamer) ResultAtCount(count arbutil.MessageIndex) (*execution.MessageResult, error) {
 	if count == 0 {
 		return &execution.MessageResult{}, nil
@@ -1164,7 +1203,7 @@ func (s *TransactionStreamer) ExecuteNextMsg(ctx context.Context) bool {
 	if pos >= msgCount {
 		return false
 	}
-	msgAndBlockHash, err := s.getMessageWithMetadataAndBlockHash(pos)
+	msgAndBlockInfo, err := s.getMessageWithMetadataAndBlockInfo(pos)
 	if err != nil {
 		log.Error("feedOneMsg failed to readMessage", "err", err, "pos", pos)
 		return false
@@ -1178,7 +1217,7 @@ func (s *TransactionStreamer) ExecuteNextMsg(ctx context.Context) bool {
 		}
 		msgForPrefetch = msg
 	}
-	msgResult, err := s.exec.DigestMessage(pos, &msgAndBlockHash.MessageWithMeta, msgForPrefetch)
+	msgResult, err := s.exec.DigestMessage(pos, &msgAndBlockInfo.MessageWithMeta, msgForPrefetch)
 	if err != nil {
 		logger := log.Warn
 		if prevMessageCount < msgCount {
@@ -1188,7 +1227,8 @@ func (s *TransactionStreamer) ExecuteNextMsg(ctx context.Context) bool {
 		return false
 	}
 
-	s.checkResult(msgResult, msgAndBlockHash.BlockHash)
+	// we just log the error but not update the value in db itself with msgResult.BlockHash? and instead forward the new block hash
+	s.checkResult(msgResult, msgAndBlockInfo.BlockHash)
 
 	batch := s.db.NewBatch()
 	err = s.storeResult(pos, *msgResult, batch)
@@ -1202,11 +1242,13 @@ func (s *TransactionStreamer) ExecuteNextMsg(ctx context.Context) bool {
 		return false
 	}
 
-	msgWithBlockHash := arbostypes.MessageWithMetadataAndBlockHash{
-		MessageWithMeta: msgAndBlockHash.MessageWithMeta,
+	msgWithBlockInfo := arbostypes.MessageWithMetadataAndBlockInfo{
+		MessageWithMeta: msgAndBlockInfo.MessageWithMeta,
 		BlockHash:       &msgResult.BlockHash,
+		// maybe if blockhash is differing we clear out previous timeboosted and not send timeboosted info to broadcasting?
+		BlockMetadata: msgAndBlockInfo.BlockMetadata,
 	}
-	s.broadcastMessages([]arbostypes.MessageWithMetadataAndBlockHash{msgWithBlockHash}, pos)
+	s.broadcastMessages([]arbostypes.MessageWithMetadataAndBlockInfo{msgWithBlockInfo}, pos)
 
 	return pos+1 < msgCount
 }

arbos/arbostypes/messagewithmeta.go

@@ -18,9 +18,12 @@ type MessageWithMetadata struct {
 	DelayedMessagesRead uint64             `json:"delayedMessagesRead"`
 }
 
-type MessageWithMetadataAndBlockHash struct {
+type BlockMetadata []byte
+
+type MessageWithMetadataAndBlockInfo struct {
 	MessageWithMeta MessageWithMetadata
 	BlockHash       *common.Hash
+	BlockMetadata   BlockMetadata
 }
 
 var EmptyTestMessageWithMetadata = MessageWithMetadata{
@@ -32,6 +35,15 @@ var TestMessageWithMetadataAndRequestId = MessageWithMetadata{
 	Message: &TestIncomingMessageWithRequestId,
 }
 
+// IsTxTimeboosted given a tx's index in the block returns whether the tx was timeboosted or not
+func (b BlockMetadata) IsTxTimeboosted(txIndex int) bool {
+	maxTxCount := (len(b) - 1) * 8
+	if txIndex >= maxTxCount {
+		return false
+	}
+	return b[1+(txIndex/8)]&(1<<(txIndex%8)) != 0
+}
+
 func (m *MessageWithMetadata) Hash(sequenceNumber arbutil.MessageIndex, chainId uint64) (common.Hash, error) {
 	serializedExtraData := make([]byte, 24)
 	binary.BigEndian.PutUint64(serializedExtraData[:8], uint64(sequenceNumber))

broadcastclient/broadcastclient_test.go

@@ -106,7 +106,7 @@
 	go func() {
 		for i := 0; i < messageCount; i++ {
 			// #nosec G115
-			Require(t, b.BroadcastSingle(arbostypes.TestMessageWithMetadataAndRequestId, arbutil.MessageIndex(i), nil))
+			Require(t, b.BroadcastSingle(arbostypes.TestMessageWithMetadataAndRequestId, arbutil.MessageIndex(i), nil, nil))
 		}
 	}()
 
@@ -138,7 +138,7 @@
 	badPrivateKey, err := crypto.GenerateKey()
 	Require(t, err)
 	badPublicKey := badPrivateKey.Public()
 	badSequencerAddr := crypto.PubkeyToAddress(*badPublicKey.(*ecdsa.PublicKey))
 	config := DefaultTestConfig
 
 	ts := NewDummyTransactionStreamer(chainId, &badSequencerAddr)
@@ -158,7 +158,7 @@
 	go func() {
 		for i := 0; i < messageCount; i++ {
 			// #nosec G115
-			Require(t, b.BroadcastSingle(arbostypes.TestMessageWithMetadataAndRequestId, arbutil.MessageIndex(i), nil))
+			Require(t, b.BroadcastSingle(arbostypes.TestMessageWithMetadataAndRequestId, arbutil.MessageIndex(i), nil, nil))
 		}
 	}()
 
@@ -318,7 +318,7 @@
 	broadcastClient.Start(ctx)
 
 	t.Log("broadcasting seq 0 message")
-	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil))
+	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil, nil))
 
 	// Wait for client to receive batch to ensure it is connected
 	timer := time.NewTimer(5 * time.Second)
@@ -389,7 +389,7 @@
 	broadcastClient.Start(ctx)
 
 	t.Log("broadcasting seq 0 message")
-	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil))
+	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil, nil))
 
 	// Wait for client to receive batch to ensure it is connected
 	timer := time.NewTimer(5 * time.Second)
@@ -726,8 +726,8 @@
 	Require(t, b.Start(ctx))
 	defer b.StopAndWait()
 
-	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil))
-	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 1, nil))
+	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 0, nil, nil))
+	Require(t, b.BroadcastSingle(arbostypes.EmptyTestMessageWithMetadata, 1, nil, nil))
 
 	var wg sync.WaitGroup
 	for i := 0; i < 2; i++ {