cmd publish (#608)

* add command publish

* cmd publish scanner fix; add worker pool safe job; add rate-limit

* gen-doc

* changes requested due to PR review

* gen-docs

* publish: improve output; script to create tx samples

* fix linter issue

* fix(monitor): handle missing safe/finalized blocks gracefully

Instead of failing when nodes don't support safe/finalized block queries,
log debug messages and default to 0. This improves compatibility with
older or non-standard node implementations.

---------

Co-authored-by: John Hilliard <[email protected]>

Author: tclemos

README.md

@@ -75,6 +75,8 @@ Note: Do not modify this section! It is auto-generated by `cobra` using `make ge
 
 - [polycli parseethwallet](doc/polycli_parseethwallet.md) - Extract the private key from an eth wallet.
 
+- [polycli publish](doc/polycli_publish.md) - Publish transactions to the network with high-throughput
+
 - [polycli retest](doc/polycli_retest.md) - Convert the standard ETH test fillers into something to be replayed against an RPC
 
 - [polycli rpcfuzz](doc/polycli_rpcfuzz.md) - Continually run a variety of RPC calls and fuzzers.

cmd/monitor/monitor.go

@@ -241,17 +241,17 @@ func getChainState(ctx context.Context, ec *ethclient.Client, txPoolStatusSuppor
 
 	safeBlock, err := ec.HeaderByNumber(ctx, big.NewInt(int64(rpc.SafeBlockNumber)))
 	if err != nil {
-		return nil, fmt.Errorf("couldn't fetch safe block number: %s", err.Error())
-	}
-	if safeBlock != nil {
+		log.Debug().Err(err).Msg("Unable to fetch safe block number")
+		cs.SafeBlock = 0
+	} else if safeBlock != nil {
 		cs.SafeBlock = safeBlock.Number.Uint64()
 	}
 
 	finalizedBlock, err := ec.HeaderByNumber(ctx, big.NewInt(int64(rpc.FinalizedBlockNumber)))
 	if err != nil {
-		return nil, fmt.Errorf("couldn't fetch finalized block number: %s", err.Error())
-	}
-	if finalizedBlock != nil {
+		log.Debug().Err(err).Msg("Unable to fetch finalized block number")
+		cs.FinalizedBlock = 0
+	} else if finalizedBlock != nil {
 		cs.FinalizedBlock = finalizedBlock.Number.Uint64()
 	}
 

cmd/publish/input.go

@@ -0,0 +1,135 @@
+package publish
+
+import (
+	"bufio"
+	"encoding/hex"
+	"fmt"
+	"iter"
+	"os"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/rs/zerolog/log"
+)
+
+const scannerBufferSize = 1024 * 1024
+
+func getInputData(inputFileName *string, args []string) (iter.Seq[string], inputDataSource, error) {
+	// firstly check and see if we have an input file
+	if inputFileName != nil && *inputFileName != "" {
+		// If we get here, we're going to assume the user
+		// wants to load transactions from a file and we're not going to look
+		// for other input sources
+		return dataFromFile(*inputFileName)
+	}
+
+	// secondly check and see if we have any args
+	if len(args) > 0 {
+		// checks if any of the args start with 0x, if so, return only the args
+		// that start with 0x
+		txArgs := make([]string, 0)
+		for _, arg := range args {
+			if arg[:2] == "0x" {
+				txArgs = append(txArgs, arg)
+			}
+		}
+
+		// If we get here, we're going to assume the user wants to load transactions
+		// from the command line and we're not going to look for other input sources
+		if len(txArgs) > 0 {
+			return dataFromArgs(txArgs)
+		}
+	}
+
+	// if we get here, we're going to assume the user wants to load transactions
+	// from stdin or from the command line
+	return dataFromStdin()
+}
+
+func dataFromArgs(args []string) (iter.Seq[string], inputDataSource, error) {
+	log.Info().Msg("Reading data from args")
+
+	return func(yield func(string) bool) {
+		for _, arg := range args {
+			if !yield(arg) {
+				return
+			}
+		}
+	}, InputDataSourceArgs, nil
+}
+
+// dataFromFile returns an iterator that reads lines from a file
+func dataFromFile(filename string) (iter.Seq[string], inputDataSource, error) {
+	f, err := os.Open(filename)
+	if err != nil {
+		return func(yield func(string) bool) {}, InputDataSourceFile, err
+	}
+	log.Info().
+		Str("filename", filename).
+		Msg("Reading data from file")
+
+	return func(yield func(string) bool) {
+		// Ensure the file is closed after the function exits
+		defer f.Close()
+		s := bufio.NewScanner(f)
+		sBuf := make([]byte, 0)
+		s.Buffer(sBuf, scannerBufferSize)
+		for s.Scan() {
+			if !yield(s.Text()) {
+				return
+			}
+		}
+		if err := s.Err(); err != nil {
+			log.Error().
+				Err(err).
+				Str("filename", filename).
+				Msg("error scanning file")
+		}
+	}, InputDataSourceFile, nil
+}
+
+// dataFromStdin returns an iterator that reads lines from stdin
+func dataFromStdin() (iter.Seq[string], inputDataSource, error) {
+	fmt.Println("Reading data from Stdin, type the transactions you want to publish:")
+
+	return func(yield func(string) bool) {
+		s := bufio.NewScanner(os.Stdin)
+		sBuf := make([]byte, 0)
+		s.Buffer(sBuf, scannerBufferSize)
+		for s.Scan() {
+			if !yield(s.Text()) {
+				return
+			}
+		}
+		if err := s.Err(); err != nil {
+			log.Error().
+				Err(err).
+				Msg("error scanning stdin")
+		}
+	}, InputDataSourceStdin, nil
+}
+
+// inputDataItemToTx converts an input data item that represents a transaction
+// rlp hex encoded into a transaction
+func inputDataItemToTx(inputDataItem string) (*types.Transaction, error) {
+	tx := new(types.Transaction)
+
+	inputDataItem = strings.TrimPrefix(inputDataItem, "0x")
+
+	// Check if the string has an odd length
+	if len(inputDataItem)%2 != 0 {
+		// Prepend a '0' to make it even-length
+		inputDataItem = "0" + inputDataItem
+	}
+
+	b, err := hex.DecodeString(inputDataItem)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := tx.UnmarshalBinary(b); err != nil {
+		return nil, err
+	}
+
+	return tx, nil
+}

cmd/publish/output.go

@@ -0,0 +1,88 @@
+package publish
+
+import (
+	"fmt"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	InputDataSourceFile  = "file"
+	InputDataSourceArgs  = "args"
+	InputDataSourceStdin = "stdin"
+)
+
+type inputDataSource string
+
+type output struct {
+	InputDataSource       inputDataSource
+	InputDataCount        atomic.Uint64
+	ValidInputs           atomic.Uint64
+	InvalidInputs         atomic.Uint64
+	StartTime             time.Time
+	EndTime               time.Time
+	TxsSentSuccessfully   atomic.Uint64
+	TxsSentUnsuccessfully atomic.Uint64
+}
+
+func (s *output) Start() {
+	s.StartTime = time.Now()
+}
+
+func (s *output) Stop() {
+	s.EndTime = time.Now()
+}
+
+func (s *output) Print() {
+	elapsed := s.EndTime.Sub(s.StartTime)
+	elapsedSeconds := elapsed.Seconds()
+	if elapsedSeconds == 0 {
+		elapsedSeconds = 1
+	}
+	txSent := s.TxsSentSuccessfully.Load() + s.TxsSentUnsuccessfully.Load()
+	txsSendPerSecond := 0.0
+	if elapsedSeconds > 0.0001 {
+		txsSendPerSecond = float64(txSent) / elapsedSeconds
+	}
+	successRatio := float64(0)
+	if txSent > 0 {
+		successRatio = float64(s.TxsSentSuccessfully.Load()) / float64(txSent) * 100
+	}
+
+	summaryString := fmt.Sprintf(`-----------------------------------
+              Summary              
+-----------------------------------
+Concurrency: %d
+JobQueueSize: %d
+RateLimit: %d
+-----------------------------------
+Input Data Source: %s
+Input Data Count: %d
+Valid Inputs: %d
+Invalid Inputs: %d
+-----------------------------------
+Elapsed Time: %s
+Txs Sent: %d
+Txs Sent Per Second: %.2f
+Txs Sent Successfully: %d
+Txs Sent Unsuccessfully: %d
+Success Ratio: %.2f%%
+-----------------------------------`,
+		*publishInputArgs.concurrency,
+		*publishInputArgs.jobQueueSize,
+		*publishInputArgs.rateLimit,
+
+		s.InputDataSource,
+		s.InputDataCount.Load(),
+		s.ValidInputs.Load(),
+		s.InvalidInputs.Load(),
+
+		elapsed,
+		txSent,
+		txsSendPerSecond,
+		s.TxsSentSuccessfully.Load(),
+		s.TxsSentUnsuccessfully.Load(),
+		successRatio)
+
+	fmt.Println(summaryString)
+}