Merge branch 'master' into merge-v1.14.7

Author: amsanghi

arbitrum/apibackend.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"net/url"
 	"strconv"
 	"strings"
 	"time"
@@ -51,6 +52,22 @@ type APIBackend struct {
 	sync           SyncProgressBackend
 }
 
+type errorFilteredFallbackClient struct {
+	impl types.FallbackClient
+	url  string
+}
+
+func (c *errorFilteredFallbackClient) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
+	err := c.impl.CallContext(ctx, result, method, args...)
+	if err != nil && strings.Contains(err.Error(), c.url) {
+		// avoids leaking the URL in the error message.
+		// URL can contain sensitive information such as API keys.
+		log.Warn("fallback client error", "error", err)
+		return errors.New("Failed to call fallback API")
+	}
+	return err
+}
+
 type timeoutFallbackClient struct {
 	impl    types.FallbackClient
 	timeout time.Duration
@@ -77,12 +94,22 @@ func CreateFallbackClient(fallbackClientUrl string, fallbackClientTimeout time.D
 		types.SetFallbackError(strings.Join(fields, ":"), int(errNumber))
 		return nil, nil
 	}
+
 	var fallbackClient types.FallbackClient
 	var err error
 	fallbackClient, err = rpc.Dial(fallbackClientUrl)
 	if err != nil {
 		return nil, fmt.Errorf("failed creating fallback connection: %w", err)
 	}
+	url, err := url.Parse(fallbackClientUrl)
+	if err != nil {
+		return nil, fmt.Errorf("failed parsing fallback URL: %w", err)
+	}
+	fallbackClient = &errorFilteredFallbackClient{
+		impl: fallbackClient,
+		url:  url.String(),
+	}
+
 	if fallbackClientTimeout != 0 {
 		fallbackClient = &timeoutFallbackClient{
 			impl:    fallbackClient,
@@ -94,8 +121,7 @@ func CreateFallbackClient(fallbackClientUrl string, fallbackClientTimeout time.D
 
 type SyncProgressBackend interface {
 	SyncProgressMap() map[string]interface{}
-	SafeBlockNumber(ctx context.Context) (uint64, error)
-	FinalizedBlockNumber(ctx context.Context) (uint64, error)
+	BlockMetadataByNumber(blockNum uint64) (common.BlockMetadata, error)
 }
 
 func createRegisterAPIBackend(backend *Backend, filterConfig filters.Config, fallbackClientUrl string, fallbackClientTimeout time.Duration) (*filters.FilterSystem, error) {
@@ -379,13 +405,23 @@ func (a *APIBackend) blockNumberToUint(ctx context.Context, number rpc.BlockNumb
 		if a.sync == nil {
 			return 0, errors.New("block number not supported: object not set")
 		}
-		return a.sync.SafeBlockNumber(ctx)
+
+		currentSafeBlock := a.BlockChain().CurrentSafeBlock()
+		if currentSafeBlock == nil {
+			return 0, errors.New("safe block not found")
+		}
+		return currentSafeBlock.Number.Uint64(), nil
 	}
 	if number == rpc.FinalizedBlockNumber {
 		if a.sync == nil {
 			return 0, errors.New("block number not supported: object not set")
 		}
-		return a.sync.FinalizedBlockNumber(ctx)
+
+		currentFinalizedBlock := a.BlockChain().CurrentFinalBlock()
+		if currentFinalizedBlock == nil {
+			return 0, errors.New("finalized block not found")
+		}
+		return currentFinalizedBlock.Number.Uint64(), nil
 	}
 	if number < 0 {
 		return 0, errors.New("block number not supported")
@@ -460,6 +496,10 @@ func (a *APIBackend) BlockByNumberOrHash(ctx context.Context, blockNrOrHash rpc.
 	return nil, errors.New("invalid arguments; neither block nor hash specified")
 }
 
+func (a *APIBackend) BlockMetadataByNumber(blockNum uint64) (common.BlockMetadata, error) {
+	return a.sync.BlockMetadataByNumber(blockNum)
+}
+
 func StateAndHeaderFromHeader(ctx context.Context, chainDb ethdb.Database, bc *core.BlockChain, maxRecreateStateDepth int64, header *types.Header, err error) (*state.StateDB, *types.Header, error) {
 	if err != nil {
 		return nil, header, err

arbitrum/recordingdb.go

@@ -32,17 +32,19 @@ type RecordingKV struct {
 	inner         *triedb.Database
 	diskDb        ethdb.KeyValueStore
 	readDbEntries map[common.Hash][]byte
+	mutex         sync.Mutex
 	enableBypass  bool
 }
 
 func newRecordingKV(inner *triedb.Database, diskDb ethdb.KeyValueStore) *RecordingKV {
-	return &RecordingKV{inner, diskDb, make(map[common.Hash][]byte), false}
+	return &RecordingKV{inner, diskDb, make(map[common.Hash][]byte), sync.Mutex{}, false}
 }
 
 func (db *RecordingKV) Has(key []byte) (bool, error) {
 	return false, errors.New("recording KV doesn't support Has")
 }
 
+// Get may be called concurrently with other Get calls
 func (db *RecordingKV) Get(key []byte) ([]byte, error) {
 	var hash common.Hash
 	var res []byte
@@ -66,6 +68,8 @@ func (db *RecordingKV) Get(key []byte) ([]byte, error) {
 	if crypto.Keccak256Hash(res) != hash {
 		return nil, fmt.Errorf("recording KV attempted to access non-hash key %v", hash)
 	}
+	db.mutex.Lock()
+	defer db.mutex.Unlock()
 	db.readDbEntries[hash] = res
 	return res, nil
 }
@@ -190,7 +194,7 @@ func (r *RecordingDatabase) StateFor(header *types.Header) (*state.StateDB, erro
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 
-	sdb, err := state.NewDeterministic(header.Root, r.db)
+	sdb, err := state.NewRecording(header.Root, r.db)
 	if err == nil {
 		r.referenceRootLockHeld(header.Root)
 	}
@@ -263,7 +267,7 @@ func (r *RecordingDatabase) PrepareRecording(ctx context.Context, lastBlockHeade
 	if lastBlockHeader != nil {
 		prevRoot = lastBlockHeader.Root
 	}
-	recordingStateDb, err := state.NewDeterministic(prevRoot, recordingStateDatabase)
+	recordingStateDb, err := state.NewRecording(prevRoot, recordingStateDatabase)
 	if err != nil {
 		return nil, nil, nil, fmt.Errorf("failed to create recordingStateDb: %w", err)
 	}

common/lru/blob_lru.go

@@ -82,3 +82,21 @@ func (c *SizeConstrainedCache[K, V]) Get(key K) (V, bool) {
 
 	return c.lru.Get(key)
 }
+
+func (c *SizeConstrainedCache[K, V]) Remove(key K) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	if v, ok := c.lru.Peek(key); ok {
+		c.size -= uint64(len(v))
+		c.lru.Remove(key)
+	}
+}
+
+func (c *SizeConstrainedCache[K, V]) Clear() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	c.lru.Purge()
+	c.size = 0
+}

common/types.go

@@ -486,3 +486,24 @@ func (b PrettyBytes) TerminalString() string {
 	}
 	return fmt.Sprintf("%#x...%x (%dB)", b[:3], b[len(b)-3:], len(b))
 }
+
+type BlockMetadata []byte
+
+// IsTxTimeboosted given a tx's index in the block returns whether the tx was timeboosted
+// or not. The first byte of blockMetadata byte array is reserved to indicate the version,
+// starting from the second byte, (N)th bit would represent if (N)th tx is timeboosted or not, 1 means yes and 0 means no
+// blockMetadata[index / 8 + 1] & (1 << (index % 8)) != 0; where index = (N - 1), implies whether (N)th tx in a block is timeboosted
+// note that number of txs in a block will always lag behind (len(blockMetadata) - 1) * 8 but it wont lag more than a value of 7
+func (b BlockMetadata) IsTxTimeboosted(txIndex int) (bool, error) {
+	if len(b) == 0 {
+		return false, errors.New("blockMetadata is not set")
+	}
+	if txIndex < 0 {
+		return false, fmt.Errorf("invalid transaction index- %d, should be positive", txIndex)
+	}
+	maxTxCount := (len(b) - 1) * 8
+	if txIndex >= maxTxCount {
+		return false, nil
+	}
+	return b[1+(txIndex/8)]&(1<<(txIndex%8)) != 0, nil
+}

core/blockchain.go

@@ -79,6 +79,9 @@ var (
 
 	triedbCommitTimer = metrics.NewRegisteredResettingTimer("chain/triedb/commits", nil)
 
+	triedbSizeGauge   = metrics.NewRegisteredGauge("chain/triedb/size", nil)
+	triedbGCProcGauge = metrics.NewRegisteredGauge("chain/triedb/gcproc", nil)
+
 	blockInsertTimer     = metrics.NewRegisteredResettingTimer("chain/inserts", nil)
 	blockValidationTimer = metrics.NewRegisteredResettingTimer("chain/validation", nil)
 	blockExecutionTimer  = metrics.NewRegisteredResettingTimer("chain/execution", nil)
@@ -1656,6 +1659,11 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
 			bc.triedb.Dereference(prevEntry.Root)
 		}
 	}
+
+	_, dirtyNodesBufferedSize, _ := bc.triedb.Size()
+	triedbSizeGauge.Update(int64(dirtyNodesBufferedSize))
+	triedbGCProcGauge.Update(int64(bc.gcproc))
+
 	return nil
 }
 

core/blockchain_arbitrum.go

@@ -21,12 +21,50 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
+func (bc *BlockChain) FlushTrieDB(capLimit common.StorageSize) error {
+	if bc.triedb.Scheme() == rawdb.PathScheme {
+		return nil
+	}
+
+	if !bc.chainmu.TryLock() {
+		return errChainStopped
+	}
+	defer bc.chainmu.Unlock()
+
+	if !bc.triegc.Empty() {
+		_, triegcBlockNumber := bc.triegc.Peek()
+		blockNumber := uint64(-triegcBlockNumber)
+
+		header := bc.GetHeaderByNumber(blockNumber)
+		if header == nil {
+			log.Warn("Reorg in progress, trie commit postponed")
+		} else {
+			err := bc.triedb.Commit(header.Root, true)
+			if err != nil {
+				return err
+			}
+
+			bc.gcproc = 0
+			bc.lastWrite = blockNumber
+		}
+	}
+
+	err := bc.triedb.Cap(capLimit)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // WriteBlockAndSetHeadWithTime also counts processTime, which will cause intermittent TrieDirty cache writes
 func (bc *BlockChain) WriteBlockAndSetHeadWithTime(block *types.Block, receipts []*types.Receipt, logs []*types.Log, state *state.StateDB, emitHeadEvent bool, processTime time.Duration) (status WriteStatus, err error) {
 	if !bc.chainmu.TryLock() {