rpc: compression with libdeflate (#20665)

close #17112 
This PR replaces the existing single-path compress/gzip middleware in
node/rpcstack.go with a dual-path implementation that distinguishes
between streaming and non-streaming responses.
a) Non-streaming responses (standard JSON-RPC calls such as
eth_getBlockByNumber):
- are now compressed in one shot using go-libdeflate, a lightweight CGo
wrapper around ebiggers/libdeflate. Benchmarks show ~1.75x speedup vs
stdlib gzip on ~30 KB JSON payloads (4.6 ms vs 8.0 ms, 122 MB/s vs 70
MB/s); under high concurrency the advantage is larger due to lower CPU
usage per request.
- Since the compressed size is known before writing, Content-Length is
now set to the exact compressed size, avoiding unnecessary
Transfer-Encoding: chunked.
- Responses smaller than 1 KB are sent uncompressed: at that size the
CPU overhead of initialising the compressor outweighs the benefit, and
for very small payloads the compressed output can exceed the input size
due to gzip framing overhead.
b) Streaming responses (e.g. debug_traceTransaction, trace_filter) are
detected via http.Flusher: when the RPC handler calls Flush() before
writing, the middleware switches to stdlib compress/gzip in streaming
mode, compressing trace data incrementally without buffering the full
response.
   
  Changes:                                                  
- node/rpcstack.go: new gzipResponseWriter with buffer/stream dual mode;
4 sync.Pool instances for zero-allocation hot path
- rpc/http.go: injects http.Flusher into request context via
httpFlusherContextKey
- rpc/handler.go: streaming methods call flush before writing to
activate streaming mode
  - go.mod: adds github.com/erigontech/go-libdeflate v0.1.0 
  
# 🚀 Performance Benchmarks: Gzip Optimization
---
<details>
<summary><b>1. Isolated Compression Benchmarks (libdeflate vs
stdlib)</b></summary>

Diff

  Benchmark (Gzip isolation)    Latency      Throughput   Mem Alloc
- BenchmarkStdlibGzip           8.0ms/req    70 MB/s      66KB
+ BenchmarkLibdeflateGzip 4.6ms/req 122 MB/s 63KB (2x faster)
Note: We observed a ~1.75x speedup on a single thread. Under high
concurrency, the advantage is even greater due to reduced CPU overhead.

</details>

<details>
<summary><b>2. eth_getBlockByNumber with txs (Old SW vs Main
SW)</b></summary>

Old SW - Results with instability and errors:

Diff

- [2. 1] qps: 2000 -> [R=97.74% max=9.103s error=503 Service
Unavailable]
- [3. 3] qps: 2300 -> [R=98.50% max=5.841s error=503 Service
Unavailable]
- [3. 4] qps: 2300 -> [R=99.80% max=5.634s error=503 Service
Unavailable]
- [3. 5] qps: 2300 -> [R=98.25% max=5.877s error=503 Service
Unavailable]


Main SW - Stable results with 100% success rate:
Diff

+ [2. 1] qps: 2000 -> [R=100.00% max=141.22ms]
+ [2. 5] qps: 2000 -> [R=100.00% max=157.7ms]
+ [3. 1] qps: 2300 -> [R=100.00% max=217.23ms]
+ [3. 5] qps: 2300 -> [R=100.00% max=224.174ms]
</details>

<details>
<summary><b>3. trace_block </b></summary>

| Metric | Old SW | Main SW | Speedup |
| :--- | :--- | :--- | :--- |
| **Throughput** | 375.9 r/s | **455.0 r/s** | **+21%** |
| **p50 Latency** | 114.4 ms | **96.9 ms** | **1.18x** |
| **p90 Latency** | 232.0 ms | **181.2 ms** | **1.28x** |
| **Mean Latency** | 132.4 ms | **109.5 ms** | **1.21x** |

</details>

<details>
<summary><b>4. Executes all RPC using http, http-compressed and
websockets</b></summary>

./run_all.sh -T http,http_comp,websocket 
Run tests in parallel on localhost:8545/localhost:8551
Result directory:
/home/simon/silkworm/tests/rpc-tests3/integration/results
Time:                         2026-04-19 08:58:05.257624
Total round_trip time:        3:40:01.359044
Total marshalling time:       0:00:00.063308
Total unmarshalling time:     0:01:40.509521
No of json Diffs:             0
Test time-elapsed:            0:08:37.467460
Available tests:              1436
Available tested api:         112
Number of loop:               1
Number of executed tests:     4152
Number of NOT executed tests: 156
Number of success tests:      4152
Number of failed tests:       0
</details>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: lupin012

go.mod

@@ -6,6 +6,7 @@ replace github.com/holiman/bloomfilter/v2 => github.com/AskAlexSharov/bloomfilte
 
 require (
 	github.com/erigontech/erigon-snapshot v1.3.1-0.20260402120223-7bb412bc89cd
+	github.com/erigontech/go-libdeflate v0.1.0
 	github.com/erigontech/mdbx-go v0.39.17
 	github.com/erigontech/secp256k1 v1.2.1-0.20260218182123-377cc1bd6410
 )

go.sum

@@ -344,6 +344,8 @@ github.com/erigontech/fastkeccak v0.1.1-0.20260408010752-08e7b6602268 h1:NHl4+Dj
 github.com/erigontech/fastkeccak v0.1.1-0.20260408010752-08e7b6602268/go.mod h1:CwJFJVKFVWpQyKSfRrQyY56IqV16sR5xnQoFThGHiZA=
 github.com/erigontech/go-eth-kzg v0.0.0-20260401161010-070339460d07 h1:0VpFIthaJzGs3iwyrKCj67wptrNq7KKiLhilDCRPqwk=
 github.com/erigontech/go-eth-kzg v0.0.0-20260401161010-070339460d07/go.mod h1:J9/u5sWfznSObptgfa92Jq8rTswn6ahQWEuiLHOjCUI=
+github.com/erigontech/go-libdeflate v0.1.0 h1:WtKiuNpuAP/Qyq3mbPwFB0DoQikabtOeZFUgo9VGRsY=
+github.com/erigontech/go-libdeflate v0.1.0/go.mod h1:OCQBWlynNqNCYb1ckqNjY69H1LAhg2TyyfYZalO03BM=
 github.com/erigontech/mdbx-go v0.39.17 h1:+FQMaCuH/IB+t9HEXVDxTJaV5jvZTwC/6YgDHHKDDMo=
 github.com/erigontech/mdbx-go v0.39.17/go.mod h1:VTGwYzepS9Wg38jVfreOsSVlh73OBGPZluu7kHo6X6g=
 github.com/erigontech/secp256k1 v1.2.1-0.20260218182123-377cc1bd6410 h1:5YD7JJ5PaqOdjKA84lTDtQby9nbI4podqrkUhyIyFDw=

node/rpcstack.go

@@ -20,6 +20,7 @@
 package node
 
 import (
+	"bytes"
 	"compress/gzip"
 	"context"
 	"errors"
@@ -28,10 +29,13 @@ import (
 	"net"
 	"net/http"
 	"slices"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
 
+	libdeflate "github.com/erigontech/go-libdeflate"
+
 	"github.com/rs/cors"
 
 	"github.com/erigontech/erigon/common/log/v3"
@@ -508,25 +512,181 @@ func (h *virtualHostHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	http.Error(w, "invalid host specified", http.StatusForbidden)
 }
 
+// gzPoolBufCap is the maximum buffer capacity retained in pools to bound RSS growth.
+const gzPoolBufCap = 1 << 20
+
+// minGzipBodySize is the minimum response body size to compress. Responses
+// smaller than this are sent as-is: gzip framing overhead would exceed savings.
+const minGzipBodySize = 1024
+
 var gzPool = sync.Pool{
+	New: func() any { w, _ := gzip.NewWriterLevel(io.Discard, gzip.BestSpeed); return w },
+}
+
+var libdeflateWarnOnce sync.Once
+var libdeflateCompressWarnOnce sync.Once
+var libdeflateDisabled atomic.Bool
+
+var gzCompressorPool = sync.Pool{
 	New: func() any {
-		w := gzip.NewWriter(io.Discard)
-		return w
+		c, err := libdeflate.NewCompressor(libdeflate.DefaultCompression)
+		if err != nil {
+			libdeflateDisabled.Store(true)
+			libdeflateWarnOnce.Do(func() {
+				log.Warn("libdeflate unavailable, falling back to stdlib gzip", "err", err)
+			})
+			return nil
+		}
+		return c
 	},
 }
 
+var gzBufPool = sync.Pool{
+	New: func() any { return new(bytes.Buffer) },
+}
+
+var gzDstPool = sync.Pool{
+	New: func() any { return make([]byte, 0, 64*1024) },
+}
+
+func getBuf() *bytes.Buffer {
+	buf := gzBufPool.Get().(*bytes.Buffer)
+	buf.Reset()
+	return buf
+}
+
+func putBuf(buf *bytes.Buffer) {
+	if buf.Cap() <= gzPoolBufCap {
+		gzBufPool.Put(buf)
+	}
+}
+
+func putDst(dst []byte) {
+	if cap(dst) <= gzPoolBufCap {
+		gzDstPool.Put(dst)
+	}
+}
+
+func gzDstGrow(b []byte, wantLen int) []byte {
+	if cap(b) >= wantLen {
+		return b[:wantLen]
+	}
+	return make([]byte, wantLen, max(wantLen, 2*cap(b)))
+}
+
 type gzipResponseWriter struct {
-	io.Writer
+	buf    *bytes.Buffer
+	gzw    *gzip.Writer
+	status int
 	http.ResponseWriter
 }
 
 func (w *gzipResponseWriter) WriteHeader(status int) {
-	w.Header().Del("Content-Length")
-	w.ResponseWriter.WriteHeader(status)
+	if w.gzw != nil {
+		w.ResponseWriter.WriteHeader(status)
+	} else {
+		w.status = status
+	}
 }
 
 func (w *gzipResponseWriter) Write(b []byte) (int, error) {
-	return w.Writer.Write(b)
+	if w.gzw != nil {
+		return w.gzw.Write(b)
+	}
+	return w.buf.Write(b)
+}
+
+// Flush switches to streaming gzip on first call; subsequent calls flush incrementally.
+func (w *gzipResponseWriter) Flush() {
+	if w.gzw == nil {
+		w.ResponseWriter.Header().Set("Content-Encoding", "gzip")
+		w.ResponseWriter.Header().Del("Content-Length")
+		if w.status != 0 {
+			w.ResponseWriter.WriteHeader(w.status)
+		}
+		w.gzw = gzPool.Get().(*gzip.Writer)
+		w.gzw.Reset(w.ResponseWriter)
+		if w.buf.Len() > 0 {
+			_, _ = w.gzw.Write(w.buf.Bytes())
+			w.buf.Reset()
+		}
+	}
+	_ = w.gzw.Flush()
+	if f, ok := w.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+func writeStdlibGzip(w http.ResponseWriter, src []byte, status int) {
+	gz := gzPool.Get().(*gzip.Writer)
+	defer gzPool.Put(gz)
+	gz.Reset(w)
+	w.Header().Set("Content-Encoding", "gzip")
+	w.Header().Del("Content-Length")
+	if status != 0 {
+		w.WriteHeader(status)
+	}
+	_, _ = gz.Write(src)
+	_ = gz.Close()
+}
+
+// compressLibdeflate tries to compress src with libdeflate and write the response.
+// Returns false if libdeflate is unavailable or compression fails; the caller
+// should then fall back to writeStdlibGzip.
+func compressLibdeflate(w http.ResponseWriter, src []byte, status int) bool {
+	if libdeflateDisabled.Load() {
+		return false
+	}
+	raw := gzCompressorPool.Get()
+	if raw == nil {
+		return false
+	}
+	c := raw.(*libdeflate.Compressor)
+	defer gzCompressorPool.Put(c)
+
+	dst := gzDstPool.Get().([]byte)
+	dst = gzDstGrow(dst, c.GzipCompressBound(len(src)))
+	defer putDst(dst)
+
+	n, err := c.CompressGzip(dst, src)
+	if err != nil {
+		libdeflateCompressWarnOnce.Do(func() {
+			log.Warn("libdeflate compression failed, falling back to stdlib gzip", "err", err)
+		})
+		return false
+	}
+
+	w.Header().Set("Content-Encoding", "gzip")
+	w.Header().Set("Content-Length", strconv.Itoa(n))
+	if status != 0 {
+		w.WriteHeader(status)
+	}
+	w.Write(dst[:n]) //nolint:errcheck
+	return true
+}
+
+func sendGzipResponse(w http.ResponseWriter, grw *gzipResponseWriter) {
+	defer putBuf(grw.buf)
+
+	if grw.gzw != nil {
+		defer gzPool.Put(grw.gzw)
+		defer grw.gzw.Close() //nolint:errcheck
+		return
+	}
+
+	src := grw.buf.Bytes()
+	if len(src) < minGzipBodySize {
+		w.Header().Set("Content-Length", strconv.Itoa(len(src)))
+		if grw.status != 0 {
+			w.WriteHeader(grw.status)
+		}
+		w.Write(src) //nolint:errcheck
+		return
+	}
+
+	if !compressLibdeflate(w, src, grw.status) {
+		writeStdlibGzip(w, src, grw.status)
+	}
 }
 
 func newGzipHandler(next http.Handler) http.Handler {
@@ -536,15 +696,12 @@ func newGzipHandler(next http.Handler) http.Handler {
 			return
 		}
 
-		w.Header().Set("Content-Encoding", "gzip")
-
-		gz := gzPool.Get().(*gzip.Writer)
-		defer gzPool.Put(gz)
-
-		gz.Reset(w)
-		defer gz.Close()
-
-		next.ServeHTTP(&gzipResponseWriter{ResponseWriter: w, Writer: gz}, r)
+		grw := &gzipResponseWriter{buf: getBuf(), ResponseWriter: w}
+		// The hook activates streaming mode before the first write; absent when gzip
+		// is off so it cannot prematurely commit HTTP headers (e.g. 200 before 503).
+		r = r.WithContext(rpc.WithGzipStreamingHook(r.Context(), grw.Flush))
+		next.ServeHTTP(grw, r)
+		sendGzipResponse(w, grw)
 	})
 }