@@ -4,9 +4,10 @@ import (
44 "bytes"
55 "context"
66 "crypto/tls"
7- "encoding/json"
87 "errors"
98 "fmt"
9+ "github.com/goccy/go-json"
10+ "github.com/valyala/fasthttp"
1011 "io"
1112 "math"
1213 "math/rand"
@@ -111,10 +112,19 @@ var (
111112 HTTPErrorCode : 500 ,
112113 }
113114
115+ ErrBackendResponseDecompressionError = & RPCErr {
116+ Code : JSONRPCErrorInternal - 22 ,
117+ Message : "backend response too large" ,
118+ HTTPErrorCode : 500 ,
119+ }
120+
114121 ErrBackendUnexpectedJSONRPC = errors .New ("backend returned an unexpected JSON-RPC response" )
115122
116123 ErrConsensusGetReceiptsCantBeBatched = errors .New ("consensus_getReceipts cannot be batched" )
117124 ErrConsensusGetReceiptsInvalidTarget = errors .New ("unsupported consensus_receipts_target" )
125+
126+ // For the new fastHTTP client
127+ headerContentTypeJson = []byte ("application/json" )
118128)
119129
120130func ErrInvalidRequest (msg string ) * RPCErr {
@@ -141,6 +151,7 @@ type Backend struct {
141151 authUsername string
142152 authPassword string
143153 headers map [string ]string
154+ fastClient * fasthttp.Client
144155 client * LimitedHTTPClient
145156 dialer * websocket.Dialer
146157 maxRetries int
@@ -318,12 +329,29 @@ func NewBackend(
318329 rpcSemaphore * semaphore.Weighted ,
319330 opts ... BackendOpt ,
320331) * Backend {
332+ readTimeout , _ := time .ParseDuration ("10000ms" )
333+ writeTimeout , _ := time .ParseDuration ("10000ms" )
334+ maxIdleConnDuration , _ := time .ParseDuration ("1h" )
321335 backend := & Backend {
322336 Name : name ,
323337 rpcURL : rpcURL ,
324338 wsURL : wsURL ,
325339 maxResponseSize : math .MaxInt64 ,
326- client : & LimitedHTTPClient {
340+ fastClient : & fasthttp.Client {
341+ MaxConnsPerHost : 16384 ,
342+ ReadTimeout : readTimeout ,
343+ WriteTimeout : writeTimeout ,
344+ MaxIdleConnDuration : maxIdleConnDuration ,
345+ NoDefaultUserAgentHeader : true , // Don't send: User-Agent: fasthttp
346+ DisableHeaderNamesNormalizing : true , // If you set the case on your headers correctly you can enable this
347+ DisablePathNormalizing : true ,
348+ // increase DNS cache time to five minutes
349+ Dial : (& fasthttp.TCPDialer {
350+ Concurrency : 32768 ,
351+ DNSCacheDuration : time .Minute * 5 ,
352+ }).Dial ,
353+ },
354+ client : & LimitedHTTPClient { // Keep legacy client for now, TODO remove
327355 Client : http.Client {Timeout : 5 * time .Second },
328356 sem : rpcSemaphore ,
329357 backendName : name ,
@@ -550,21 +578,10 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
550578 body = mustMarshalJSON (rpcReqs )
551579 }
552580
553- httpReq , err := http .NewRequestWithContext (ctx , "POST" , b .rpcURL , bytes .NewReader (body ))
554- if err != nil {
555- b .intermittentErrorsSlidingWindow .Incr ()
556- RecordBackendNetworkErrorRateSlidingWindow (b , b .ErrorRate ())
557- return nil , wrapErr (err , "error creating backend request" )
558- }
559-
560- if b .authPassword != "" {
561- httpReq .SetBasicAuth (b .authUsername , b .authPassword )
562- }
563-
564- opTxProxyAuth := GetOpTxProxyAuthHeader (ctx )
565- if opTxProxyAuth != "" {
566- httpReq .Header .Set (DefaultOpTxProxyAuthHeader , opTxProxyAuth )
567- }
581+ req := fasthttp .AcquireRequest ()
582+ req .SetRequestURI (b .rpcURL )
583+ req .Header .SetMethod (fasthttp .MethodPost )
584+ req .SetBody (body )
568585
569586 xForwardedFor := GetXForwardedFor (ctx )
570587 if b .stripTrailingXFF {
@@ -573,42 +590,63 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
573590 xForwardedFor = fmt .Sprintf ("%s, %s" , xForwardedFor , b .proxydIP )
574591 }
575592
576- httpReq .Header .Set ("content-type" , "application/json" )
577- httpReq .Header .Set ("X-Forwarded-For" , xForwardedFor )
593+ req .Header .SetContentTypeBytes (headerContentTypeJson )
594+ req .Header .Set ("X-Forwarded-For" , xForwardedFor )
595+
596+ // Geth native backend supports gzip
597+ req .Header .Set ("Accept-Encoding" , "gzip" )
578598
579599 for name , value := range b .headers {
580- httpReq .Header .Set (name , value )
600+ req .Header .Set (name , value )
581601 }
582602
583603 start := time .Now ()
584- httpRes , err := b .client .DoLimited (httpReq )
604+
605+ httpRes := fasthttp .AcquireResponse ()
606+
607+ reqTimeout := time .Duration (5000 ) * time .Millisecond
608+
609+ err := b .fastClient .DoTimeout (req , httpRes , reqTimeout )
585610 if err != nil {
586611 b .intermittentErrorsSlidingWindow .Incr ()
587612 RecordBackendNetworkErrorRateSlidingWindow (b , b .ErrorRate ())
588- return nil , wrapErr (err , "error in backend request" )
613+ return nil , wrapErr (err , "err in backend request" )
589614 }
615+ fasthttp .ReleaseRequest (req )
616+ defer fasthttp .ReleaseResponse (httpRes )
590617
591618 metricLabelMethod := rpcReqs [0 ].Method
592619 if isBatch {
593620 metricLabelMethod = "<batch>"
594621 }
622+
623+ sc := httpRes .StatusCode ()
624+
595625 rpcBackendHTTPResponseCodesTotal .WithLabelValues (
596626 GetAuthCtx (ctx ),
597627 b .Name ,
598628 metricLabelMethod ,
599- strconv .Itoa (httpRes . StatusCode ),
629+ strconv .Itoa (sc ),
600630 strconv .FormatBool (isBatch ),
601631 ).Inc ()
602632
603633 // Alchemy returns a 400 on bad JSONs, so handle that case
604- if httpRes . StatusCode != 200 && httpRes . StatusCode != 400 {
634+ if sc != 200 && sc != 400 {
605635 b .intermittentErrorsSlidingWindow .Incr ()
606636 RecordBackendNetworkErrorRateSlidingWindow (b , b .ErrorRate ())
607637 return nil , fmt .Errorf ("response code %d" , httpRes .StatusCode )
608638 }
609639
610- defer httpRes .Body .Close ()
611- resB , err := io .ReadAll (LimitReader (httpRes .Body , b .maxResponseSize ))
640+ // defer httpRes.Body.Close()
641+
642+ // This should intelligently choose decompression based on whether the downstream
643+ // backend supported our request or not.
644+ bodyUncomp , err := httpRes .BodyUncompressed ()
645+ if err != nil {
646+ return nil , ErrBackendResponseDecompressionError
647+ }
648+
649+ resB , err := io .ReadAll (LimitReader (bytes .NewReader (bodyUncomp ), b .maxResponseSize ))
612650 if errors .Is (err , ErrLimitReaderOverLimit ) {
613651 return nil , ErrBackendResponseTooLarge
614652 }
@@ -649,9 +687,9 @@ func (b *Backend) doForward(ctx context.Context, rpcReqs []*RPCReq, isBatch bool
649687
650688 // capture the HTTP status code in the response. this will only
651689 // ever be 400 given the status check on line 318 above.
652- if httpRes . StatusCode != 200 {
690+ if sc != 200 {
653691 for _ , res := range rpcRes {
654- res .Error .HTTPErrorCode = httpRes . StatusCode
692+ res .Error .HTTPErrorCode = sc
655693 }
656694 }
657695 duration := time .Since (start )
@@ -1423,6 +1461,14 @@ func (bg *BackendGroup) ForwardRequestToBackendGroup(
14231461 error : err ,
14241462 }
14251463 }
1464+
1465+ if errors .Is (err , ErrBackendResponseDecompressionError ) {
1466+ return & BackendGroupRPCResponse {
1467+ RPCRes : nil ,
1468+ ServedBy : "" ,
1469+ error : err ,
1470+ }
1471+ }
14261472 if errors .Is (err , ErrBackendOffline ) {
14271473 log .Warn (
14281474 "skipping offline backend" ,
0 commit comments