allow parallel workers count

Author: TymKh

cmd/receiver-proxy/main.go

@@ -75,6 +75,12 @@ var flags = []cli.Flag{
 		Usage:   "address of the orderflow archive endpoint (block-processor)",
 		EnvVars: []string{"ORDERFLOW_ARCHIVE_ENDPOINT"},
 	},
+	&cli.IntFlag{
+		Name:    "archive-worker-count",
+		Value:   5,
+		Usage:   "number of parallel workers sending orderflow to archive",
+		EnvVars: []string{"ARCHIVE_WORKER_COUNT"},
+	},
 
 	// Various configs
 	&cli.StringFlag{
@@ -237,7 +243,7 @@ func runMain(cCtx *cli.Context) error {
 	flashbotsSignerAddress := eth.HexToAddress(flashbotsSignerStr)
 	maxRequestBodySizeBytes := cCtx.Int64("max-request-body-size-bytes")
 	connectionsPerPeer := cCtx.Int("connections-per-peer")
-
+	archiveWorkerCount := cCtx.Int("archive-worker-count")
 	maxUserRPS := cCtx.Int(flagMaxUserRPS)
 
 	certDuration := cCtx.Duration("cert-duration")
@@ -262,6 +268,7 @@ func runMain(cCtx *cli.Context) error {
 		MaxRequestBodySizeBytes:     maxRequestBodySizeBytes,
 		ConnectionsPerPeer:          connectionsPerPeer,
 		MaxUserRPS:                  maxUserRPS,
+		ArchiveWorkerCount:          archiveWorkerCount,
 	}
 
 	instance, err := proxy.NewReceiverProxy(*proxyConfig)

proxy/archive.go

@@ -26,7 +26,7 @@ var (
 	ArchiveRequestTimeout = time.Second * 15
 	ArchiveRetryMaxTime   = time.Second * 120
 
-	ArchiveWorkerQueueSize = 10000
+	ArchiveWorkerQueueSize = 20000
 )
 
 type ArchiveQueue struct {

proxy/receiver_proxy.go

@@ -94,6 +94,7 @@ type ReceiverProxyConfig struct {
 
 	ConnectionsPerPeer int
 	MaxUserRPS         int
+	ArchiveWorkerCount int
 }
 
 func NewReceiverProxy(config ReceiverProxyConfig) (*ReceiverProxy, error) {
@@ -185,6 +186,7 @@ func NewReceiverProxy(config ReceiverProxyConfig) (*ReceiverProxy, error) {
 		flushQueue:        archiveFlushCh,
 		archiveClient:     archiveClient,
 		blockNumberSource: NewBlockNumberSource(config.EthRPC),
+		workerCount:       config.ArchiveWorkerCount,
 	}
 	go archiveQueue.Run()