Adding support for authorization header

cmd/loadtest/app.go

@@ -78,6 +78,7 @@ type (
 		ContractCallFunctionArgs      *[]string
 		ContractCallPayable           *bool
 		InscriptionContent            *string
+		RawHTTPHeaders                *[]string
 
 		// Computed
 		CurrentGasPrice     *big.Int
@@ -93,6 +94,7 @@ type (
 		Mode                loadTestMode
 		ParsedModes         []loadTestMode
 		MultiMode           bool
+		ParsedHTTPHeaders   map[string]string
 	}
 )
 
@@ -190,6 +192,14 @@ func checkLoadtestFlags() error {
 		return errors.New("the backoff factor needs to be non-zero positive")
 	}
 
+	if ltp.RawHTTPHeaders != nil {
+		var err error
+		inputLoadTestParams.ParsedHTTPHeaders, err = util.ParseHeaderStrings(*ltp.RawHTTPHeaders)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -260,6 +270,7 @@ inscription - sending inscription transactions`)
 	ltp.ContractCallFunctionArgs = LoadtestCmd.Flags().StringSlice("function-arg", []string{}, `The arguments that will be passed to a contract function call. This must be paired up with "--mode contract-call" and "--contract-address". Args can be passed multiple times: "--function-arg 'test' --function-arg 999" or comma separated values "--function-arg "test",9". The ordering of the arguments must match the ordering of the function parameters.`)
 	ltp.ContractCallPayable = LoadtestCmd.Flags().Bool("contract-call-payable", false, "Use this flag if the function is payable, the value amount passed will be from --eth-amount. This must be paired up with --mode contract-call and --contract-address")
 	ltp.InscriptionContent = LoadtestCmd.Flags().String("inscription-content", `data:,{"p":"erc-20","op":"mint","tick":"TEST","amt":"1"}`, "The inscription content that will be encoded as calldata. This must be paired up with --mode inscription")
+	ltp.RawHTTPHeaders = LoadtestCmd.Flags().StringSliceP("header", "H", nil, "Header to be added to each HTTP request. E.g. \"X-First-Name: Joe\"")
 
 	inputLoadTestParams = *ltp
 

cmd/loadtest/loadtest.go

@@ -347,7 +347,15 @@ func runLoadTest(ctx context.Context) error {
 	}
 
 	// Dial the Ethereum RPC server.
-	rpc, err := ethrpc.DialContext(ctx, *inputLoadTestParams.RPCUrl)
+	var rpc *ethrpc.Client
+	var err error
+	if inputLoadTestParams.ParsedHTTPHeaders == nil {
+		log.Trace().Msg("No HeadersAdding custom headers")
+		rpc, err = ethrpc.DialContext(ctx, *inputLoadTestParams.RPCUrl)
+	} else {
+		log.Trace().Msg("Adding custom headers")
+		rpc, err = ethrpc.DialOptions(ctx, *inputLoadTestParams.RPCUrl, ethrpc.WithHTTPAuth(util.GetHTTPAuth(inputLoadTestParams.ParsedHTTPHeaders)))
+	}
 	if err != nil {
 		log.Error().Err(err).Msg("Unable to dial rpc")
 		return err

cmd/monitor/cmd.go

@@ -16,10 +16,12 @@ var (
 	usage string
 
 	// flags
-	rpcUrl          string
-	batchSizeValue  string
-	blockCacheLimit int
-	intervalStr     string
+	rpcUrl            string
+	batchSizeValue    string
+	blockCacheLimit   int
+	intervalStr       string
+	rawHttpHeaders    []string
+	parsedHttpHeaders map[string]string
 
 	defaultBatchSize = 100
 )
@@ -79,13 +81,22 @@ func init() {
 	MonitorCmd.PersistentFlags().StringVarP(&batchSizeValue, "batch-size", "b", "auto", "Number of requests per batch")
 	MonitorCmd.PersistentFlags().IntVarP(&blockCacheLimit, "cache-limit", "c", 200, "Number of cached blocks for the LRU block data structure (Min 100)")
 	MonitorCmd.PersistentFlags().StringVarP(&intervalStr, "interval", "i", "5s", "Amount of time between batch block rpc calls")
+	MonitorCmd.PersistentFlags().StringSliceVarP(&rawHttpHeaders, "header", "H", nil, "Header to be added to each HTTP request. E.g. \"X-First-Name: Joe\"")
+
 }
 
 func checkFlags() (err error) {
 	if err = util.ValidateUrl(rpcUrl); err != nil {
 		return
 	}
 
+	if rawHttpHeaders != nil {
+		parsedHttpHeaders, err = util.ParseHeaderStrings(rawHttpHeaders)
+		if err != nil {
+			return err
+		}
+	}
+
 	interval, err = time.ParseDuration(intervalStr)
 	if err != nil {
 		return err

cmd/monitor/monitor.go

@@ -27,13 +27,34 @@ import (
 var errBatchRequestsNotSupported = errors.New("batch requests are not supported")
 
 var (
-	windowSize         int
-	batchSize          SafeBatchSize
-	interval           time.Duration
-	one                = big.NewInt(1)
-	zero               = big.NewInt(0)
+	// windowSize determines the number of blocks to display in the monitor UI at one time.
+	windowSize int
+
+	// batchSize holds the number of blocks to fetch in one batch.
+	// It can be adjusted dynamically based on network conditions.
+	batchSize SafeBatchSize
+
+	// interval specifies the time duration to wait between each update cycle.
+	interval time.Duration
+
+	// one and zero are big.Int representations of 1 and 0, used for convenience in calculations.
+	one  = big.NewInt(1)
+	zero = big.NewInt(0)
+
+	// observedPendingTxs holds a historical record of the number of pending transactions.
 	observedPendingTxs historicalRange
-	maxDataPoints      = 1000
+
+	// maxDataPoints defines the maximum number of data points to keep in historical records.
+	maxDataPoints = 1000
+
+	// maxConcurrency defines the maximum number of goroutines that can fetch block data concurrently.
+	maxConcurrency = 10
+
+	// semaphore is a channel used to control the concurrency of block data fetch operations.
+	semaphore = make(chan struct{}, maxConcurrency)
+
+	// size of the sub batches to divide and conquer the total batch size with
+	subBatchSize = 50
 )
 
 type (
@@ -75,7 +96,14 @@ const (
 
 func monitor(ctx context.Context) error {
 	// Dial rpc.
-	rpc, err := ethrpc.DialContext(ctx, rpcUrl)
+	var rpc *ethrpc.Client
+	var err error
+	if parsedHttpHeaders == nil {
+		rpc, err = ethrpc.DialContext(ctx, rpcUrl)
+	} else {
+		rpc, err = ethrpc.DialOptions(ctx, rpcUrl, ethrpc.WithHTTPAuth(util.GetHTTPAuth(parsedHttpHeaders)))
+	}
+
 	if err != nil {
 		log.Error().Err(err).Msg("Unable to dial rpc")
 		return err
@@ -277,26 +305,64 @@ func (ms *monitorStatus) getBlockRange(ctx context.Context, to *big.Int, rpc *et
 		return nil
 	}
 
-	b := backoff.NewExponentialBackOff()
-	b.MaxElapsedTime = 3 * time.Minute
-	retryable := func() error {
-		return rpc.BatchCallContext(ctx, blms)
-	}
-	if err := backoff.Retry(retryable, b); err != nil {
+	err := ms.processBatchesConcurrently(ctx, rpc, blms)
+	if err != nil {
+		log.Error().Err(err).Msg("Error processing batches concurrently")
 		return err
 	}
 
-	ms.BlocksLock.Lock()
-	defer ms.BlocksLock.Unlock()
-	for _, b := range blms {
-		if b.Error != nil {
-			continue
-		}
-		pb := rpctypes.NewPolyBlock(b.Result.(*rpctypes.RawBlockResponse))
-		ms.BlockCache.Add(pb.Number().String(), pb)
+	return nil
+}
+
+func (ms *monitorStatus) processBatchesConcurrently(ctx context.Context, rpc *ethrpc.Client, blms []ethrpc.BatchElem) error {
+	var wg sync.WaitGroup
+	var errs []error = make([]error, 0)
+	var errorsMutex sync.Mutex
+
+	for i := 0; i < len(blms); i += subBatchSize {
+		semaphore <- struct{}{}
+		wg.Add(1)
+		go func(i int) {
+			defer func() {
+				<-semaphore
+				wg.Done()
+			}()
+			end := i + subBatchSize
+			if end > len(blms) {
+				end = len(blms)
+			}
+			subBatch := blms[i:end]
+
+			b := backoff.NewExponentialBackOff()
+			b.MaxElapsedTime = 3 * time.Minute
+			retryable := func() error {
+				return rpc.BatchCallContext(ctx, subBatch)
+			}
+			if err := backoff.Retry(retryable, b); err != nil {
+				log.Error().Err(err).Msg("unable to retry")
+				errorsMutex.Lock()
+				errs = append(errs, err)
+				errorsMutex.Unlock()
+				return
+			}
+
+			for _, elem := range subBatch {
+				if elem.Error != nil {
+					log.Error().Str("Method", elem.Method).Interface("Args", elem.Args).Err(elem.Error).Msg("Failed batch element")
+				} else {
+					pb := rpctypes.NewPolyBlock(elem.Result.(*rpctypes.RawBlockResponse))
+					ms.BlocksLock.Lock()
+					ms.BlockCache.Add(pb.Number().String(), pb)
+					ms.BlocksLock.Unlock()
+				}
+			}
+
+		}(i)
 	}
 
-	return nil
+	wg.Wait()
+
+	return errors.Join(errs...)
 }
 
 func renderMonitorUI(ctx context.Context, ec *ethclient.Client, ms *monitorStatus, rpc *ethrpc.Client) error {
@@ -315,6 +381,8 @@ func renderMonitorUI(ctx context.Context, ec *ethclient.Client, ms *monitorStatu
 	grid.SetRect(0, 0, termWidth, termHeight)
 	blockGrid.SetRect(0, 0, termWidth, termHeight)
 	transactionGrid.SetRect(0, 0, termWidth, termHeight)
+	// Initial render needed I assume to avoid the first bad redraw
+	termui.Render(grid)
 
 	var setBlock = false
 	var renderedBlocks rpctypes.SortableBlocks
@@ -401,7 +469,8 @@ func renderMonitorUI(ctx context.Context, ec *ethclient.Client, ms *monitorStatu
 		ms.BlocksLock.RUnlock()
 		renderedBlocks = renderedBlocksTemp
 
-		skeleton.Current.Text = ui.GetCurrentBlockInfo(ms.HeadBlock, ms.GasPrice, ms.PeerCount, ms.PendingCount, ms.ChainID, renderedBlocks)
+		log.Warn().Int("skeleton.Current.Inner.Dy()", skeleton.Current.Inner.Dy()).Int("skeleton.Current.Inner.Dx()", skeleton.Current.Inner.Dx()).Msg("the dimension of the current box")
+		skeleton.Current.Text = ui.GetCurrentBlockInfo(ms.HeadBlock, ms.GasPrice, ms.PeerCount, ms.PendingCount, ms.ChainID, renderedBlocks, skeleton.Current.Inner.Dx(), skeleton.Current.Inner.Dy())
 		skeleton.TxPerBlockChart.Data = metrics.GetTxsPerBlock(renderedBlocks)
 		skeleton.GasPriceChart.Data = metrics.GetMeanGasPricePerBlock(renderedBlocks)
 		skeleton.BlockSizeChart.Data = metrics.GetSizePerBlock(renderedBlocks)