Merge branch 'master' into merkle-perf

arbnode/dataposter/data_poster.go

@@ -217,6 +217,10 @@ func NewDataPoster(ctx context.Context, opts *DataPosterOpts) (*DataPoster, erro
 func rpcClient(ctx context.Context, opts *ExternalSignerCfg) (*rpc.Client, error) {
 	tlsCfg := &tls.Config{
 		MinVersion: tls.VersionTLS12,
+		// Dataposter verifies that signed transaction was signed by the account
+		// that it expects to be signed with. So signer is already authenticated
+		// on application level and does not need to rely on TLS for authentication.
+		InsecureSkipVerify: opts.InsecureSkipVerify, // #nosec G402
 	}
 
 	if opts.ClientCert != "" && opts.ClientPrivateKey != "" {
@@ -1223,6 +1227,8 @@ type ExternalSignerCfg struct {
 	// (Optional) Client certificate key for mtls.
 	// This is required when client-cert is set.
 	ClientPrivateKey string `koanf:"client-private-key"`
+	// TLS config option, when enabled skips certificate verification of external signer.
+	InsecureSkipVerify bool `koanf:"insecure-skip-verify"`
 }
 
 type DangerousConfig struct {
@@ -1276,6 +1282,7 @@ func addExternalSignerOptions(prefix string, f *pflag.FlagSet) {
 	f.String(prefix+".root-ca", DefaultDataPosterConfig.ExternalSigner.RootCA, "external signer root CA")
 	f.String(prefix+".client-cert", DefaultDataPosterConfig.ExternalSigner.ClientCert, "rpc client cert")
 	f.String(prefix+".client-private-key", DefaultDataPosterConfig.ExternalSigner.ClientPrivateKey, "rpc client private key")
+	f.Bool(prefix+".insecure-skip-verify", DefaultDataPosterConfig.ExternalSigner.InsecureSkipVerify, "skip TLS certificate verification")
 }
 
 var DefaultDataPosterConfig = DataPosterConfig{
@@ -1297,7 +1304,7 @@ var DefaultDataPosterConfig = DataPosterConfig{
 	UseNoOpStorage:         false,
 	LegacyStorageEncoding:  false,
 	Dangerous:              DangerousConfig{ClearDBStorage: false},
-	ExternalSigner:         ExternalSignerCfg{Method: "eth_signTransaction"},
+	ExternalSigner:         ExternalSignerCfg{Method: "eth_signTransaction", InsecureSkipVerify: false},
 	MaxFeeCapFormula:       "((BacklogOfBatches * UrgencyGWei) ** 2) + ((ElapsedTime/ElapsedTimeBase) ** 2) * ElapsedTimeImportance + TargetPriceGWei",
 	ElapsedTimeBase:        10 * time.Minute,
 	ElapsedTimeImportance:  10,
@@ -1330,7 +1337,7 @@ var TestDataPosterConfig = DataPosterConfig{
 	UseDBStorage:           false,
 	UseNoOpStorage:         false,
 	LegacyStorageEncoding:  false,
-	ExternalSigner:         ExternalSignerCfg{Method: "eth_signTransaction"},
+	ExternalSigner:         ExternalSignerCfg{Method: "eth_signTransaction", InsecureSkipVerify: true},
 	MaxFeeCapFormula:       "((BacklogOfBatches * UrgencyGWei) ** 2) + ((ElapsedTime/ElapsedTimeBase) ** 2) * ElapsedTimeImportance + TargetPriceGWei",
 	ElapsedTimeBase:        10 * time.Minute,
 	ElapsedTimeImportance:  10,

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