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

arbnode/message_pruner.go

@@ -23,13 +23,14 @@ import (
 
 type MessagePruner struct {
 	stopwaiter.StopWaiter
-	transactionStreamer         *TransactionStreamer
-	inboxTracker                *InboxTracker
-	config                      MessagePrunerConfigFetcher
-	pruningLock                 sync.Mutex
-	lastPruneDone               time.Time
-	cachedPrunedMessages        uint64
-	cachedPrunedDelayedMessages uint64
+	transactionStreamer              *TransactionStreamer
+	inboxTracker                     *InboxTracker
+	config                           MessagePrunerConfigFetcher
+	pruningLock                      sync.Mutex
+	lastPruneDone                    time.Time
+	cachedPrunedMessages             uint64
+	cachedPrunedBlockHashesInputFeed uint64
+	cachedPrunedDelayedMessages      uint64
 }
 
 type MessagePrunerConfig struct {
@@ -115,7 +116,15 @@ func (m *MessagePruner) prune(ctx context.Context, count arbutil.MessageIndex, g
 }
 
 func (m *MessagePruner) deleteOldMessagesFromDB(ctx context.Context, messageCount arbutil.MessageIndex, delayedMessageCount uint64) error {
-	prunedKeysRange, err := deleteFromLastPrunedUptoEndKey(ctx, m.transactionStreamer.db, messagePrefix, &m.cachedPrunedMessages, uint64(messageCount))
+	prunedKeysRange, err := deleteFromLastPrunedUptoEndKey(ctx, m.transactionStreamer.db, blockHashInputFeedPrefix, &m.cachedPrunedBlockHashesInputFeed, uint64(messageCount))
+	if err != nil {
+		return fmt.Errorf("error deleting expected block hashes: %w", err)
+	}
+	if len(prunedKeysRange) > 0 {
+		log.Info("Pruned expected block hashes:", "first pruned key", prunedKeysRange[0], "last pruned key", prunedKeysRange[len(prunedKeysRange)-1])
+	}
+
+	prunedKeysRange, err = deleteFromLastPrunedUptoEndKey(ctx, m.transactionStreamer.db, messagePrefix, &m.cachedPrunedMessages, uint64(messageCount))
 	if err != nil {
 		return fmt.Errorf("error deleting last batch messages: %w", err)
 	}

arbnode/message_pruner_test.go

@@ -22,8 +22,8 @@ func TestMessagePrunerWithPruningEligibleMessagePresent(t *testing.T) {
 	Require(t, err)
 
 	checkDbKeys(t, messagesCount, transactionStreamerDb, messagePrefix)
+	checkDbKeys(t, messagesCount, transactionStreamerDb, blockHashInputFeedPrefix)
 	checkDbKeys(t, messagesCount, inboxTrackerDb, rlpDelayedMessagePrefix)
-
 }
 
 func TestMessagePrunerTwoHalves(t *testing.T) {
@@ -71,16 +71,18 @@ func TestMessagePrunerWithNoPruningEligibleMessagePresent(t *testing.T) {
 	Require(t, err)
 
 	checkDbKeys(t, uint64(messagesCount), transactionStreamerDb, messagePrefix)
+	checkDbKeys(t, uint64(messagesCount), transactionStreamerDb, blockHashInputFeedPrefix)
 	checkDbKeys(t, messagesCount, inboxTrackerDb, rlpDelayedMessagePrefix)
 
 }
 
 func setupDatabase(t *testing.T, messageCount, delayedMessageCount uint64) (ethdb.Database, ethdb.Database, *MessagePruner) {
-
 	transactionStreamerDb := rawdb.NewMemoryDatabase()
 	for i := uint64(0); i < uint64(messageCount); i++ {
 		err := transactionStreamerDb.Put(dbKey(messagePrefix, i), []byte{})
 		Require(t, err)
+		err = transactionStreamerDb.Put(dbKey(blockHashInputFeedPrefix, i), []byte{})
+		Require(t, err)
 	}
 
 	inboxTrackerDb := rawdb.NewMemoryDatabase()

arbnode/schema.go

@@ -5,6 +5,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
 	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
 	parentChainBlockNumberPrefix []byte = []byte("p") // maps a delayed sequence number to a parent chain block number