From 91ee486a4ed16f68eaf6377e3c8a21897a3eb8f0 Mon Sep 17 00:00:00 2001 From: Rodrigo Daniel Pino Trueba Date: Wed, 19 Aug 2026 14:11:45 +0100 Subject: [PATCH 1/5] refactor: move compression utilities into their own package --- adapters/core2sn/class.go | 3 ++- adapters/sn2core/sn2core.go | 3 ++- core/class_hash.go | 4 ++-- rpc/v10/transaction_test.go | 4 ++-- rpc/v8/class.go | 3 ++- rpc/v8/transaction.go | 4 ++-- rpc/v9/class.go | 4 ++-- rpc/v9/transaction.go | 4 ++-- utils/{ => compression}/compression.go | 2 +- utils/{ => compression}/compression_test.go | 12 ++++++------ 10 files changed, 23 insertions(+), 20 deletions(-) rename utils/{ => compression}/compression.go (97%) rename utils/{ => compression}/compression_test.go (63%) diff --git a/adapters/core2sn/class.go b/adapters/core2sn/class.go index c096de74ac..3235177619 100644 --- a/adapters/core2sn/class.go +++ b/adapters/core2sn/class.go @@ -6,6 +6,7 @@ import ( "github.com/NethermindEth/juno/core" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) func AdaptSegmentLengths(l core.SegmentLengths) starknet.SegmentLengths { @@ -115,7 +116,7 @@ func AdaptDeprecatedEntryPoint(ep *core.DeprecatedEntryPoint) starknet.EntryPoin func AdaptDeprecatedCairoClass( class *core.DeprecatedCairoClass, ) (starknet.DeprecatedCairoClass, error) { - decompressedProgram, err := utils.Gzip64Decode(class.Program) + decompressedProgram, err := compression.Gzip64Decode(class.Program) if err != nil { return starknet.DeprecatedCairoClass{}, err } diff --git a/adapters/sn2core/sn2core.go b/adapters/sn2core/sn2core.go index 24de85fe6e..d25d3e1eb5 100644 --- a/adapters/sn2core/sn2core.go +++ b/adapters/sn2core/sn2core.go @@ -12,6 +12,7 @@ import ( "github.com/NethermindEth/juno/l1/eth" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) // ErrPreConfirmedIdentifierMismatch is returned by AdaptPreConfirmedWithDelta @@ -430,7 +431,7 @@ func AdaptDeprecatedCairoClass( } var err error - class.Program, err = utils.Gzip64Encode(response.Program) + class.Program, err = compression.Gzip64Encode(response.Program) if err != nil { return nil, err } diff --git a/core/class_hash.go b/core/class_hash.go index ab1d72ea52..630273a08a 100644 --- a/core/class_hash.go +++ b/core/class_hash.go @@ -5,11 +5,11 @@ import ( "github.com/NethermindEth/juno/core/crypto" "github.com/NethermindEth/juno/core/felt" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) func deprecatedCairoClassHash(class *DeprecatedCairoClass) (felt.Felt, error) { - decompressedProgram, err := utils.Gzip64Decode(class.Program) + decompressedProgram, err := compression.Gzip64Decode(class.Program) if err != nil { return felt.Felt{}, err } diff --git a/rpc/v10/transaction_test.go b/rpc/v10/transaction_test.go index 2fdc5a7046..7724c2d659 100644 --- a/rpc/v10/transaction_test.go +++ b/rpc/v10/transaction_test.go @@ -29,7 +29,7 @@ import ( "github.com/NethermindEth/juno/starknet" adaptfeeder "github.com/NethermindEth/juno/starknetdata/feeder" "github.com/NethermindEth/juno/sync/preconfirmed" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/log" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -2285,7 +2285,7 @@ func TestContractClassToGatewayPayload(t *testing.T) { require.Equal(t, class.EntryPoints, decoded.EntryPoints) require.Equal(t, class.ABI, decoded.ABI) - sierraJSON, err := utils.Gzip64Decode(decoded.SierraProgram) + sierraJSON, err := compression.Gzip64Decode(decoded.SierraProgram) require.NoError(t, err, "sierra_program must be gzip+base64 encoded") var roundTripped []felt.Felt diff --git a/rpc/v8/class.go b/rpc/v8/class.go index 184647155a..10ddaf0c61 100644 --- a/rpc/v8/class.go +++ b/rpc/v8/class.go @@ -14,6 +14,7 @@ import ( "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) // https://github.com/starkware-libs/starknet-specs/blob/v0.8.1/api/starknet_api_openrpc.json#L3159 @@ -66,7 +67,7 @@ func adaptDeclaredClass( } base64Program := string(program[1 : len(program)-1]) - feederClass.DeprecatedCairo.Program, err = utils.Gzip64Decode(base64Program) + feederClass.DeprecatedCairo.Program, err = compression.Gzip64Decode(base64Program) if err != nil { return nil, err } diff --git a/rpc/v8/transaction.go b/rpc/v8/transaction.go index 2b17ce5edd..51079d47c9 100644 --- a/rpc/v8/transaction.go +++ b/rpc/v8/transaction.go @@ -20,7 +20,7 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/throttler" "go.uber.org/zap" ) @@ -711,7 +711,7 @@ func (h *Handler) pushToFeederGateway(ctx context.Context, tx *BroadcastedTransa return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } - gwSierraProg, errIn := utils.Gzip64Encode(sierraProgBytes) + gwSierraProg, errIn := compression.Gzip64Encode(sierraProgBytes) if errIn != nil { return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } diff --git a/rpc/v9/class.go b/rpc/v9/class.go index 35973e8a9b..5803cab3ad 100644 --- a/rpc/v9/class.go +++ b/rpc/v9/class.go @@ -13,7 +13,7 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" ) type CalldataInputs = rpccore.LimitSlice[felt.Felt, rpccore.FunctionCalldataLimit] @@ -73,7 +73,7 @@ func AdaptDeclaredClass( } base64Program := string(program[1 : len(program)-1]) - feederClass.DeprecatedCairo.Program, err = utils.Gzip64Decode(base64Program) + feederClass.DeprecatedCairo.Program, err = compression.Gzip64Decode(base64Program) if err != nil { return nil, err } diff --git a/rpc/v9/transaction.go b/rpc/v9/transaction.go index 1aecf1a3df..8b2879cafb 100644 --- a/rpc/v9/transaction.go +++ b/rpc/v9/transaction.go @@ -21,7 +21,7 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/throttler" "go.uber.org/zap" ) @@ -789,7 +789,7 @@ func (h *Handler) pushToFeederGateway( return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } - gwSierraProg, errIn := utils.Gzip64Encode(sierraProgBytes) + gwSierraProg, errIn := compression.Gzip64Encode(sierraProgBytes) if errIn != nil { return AddTxResponse{}, jsonrpc.Err(jsonrpc.InternalError, errIn.Error()) } diff --git a/utils/compression.go b/utils/compression/compression.go similarity index 97% rename from utils/compression.go rename to utils/compression/compression.go index 6150f28c19..941d6c3805 100644 --- a/utils/compression.go +++ b/utils/compression/compression.go @@ -1,4 +1,4 @@ -package utils +package compression import ( "bytes" diff --git a/utils/compression_test.go b/utils/compression/compression_test.go similarity index 63% rename from utils/compression_test.go rename to utils/compression/compression_test.go index 2aacb58c95..dc83b77da7 100644 --- a/utils/compression_test.go +++ b/utils/compression/compression_test.go @@ -1,9 +1,9 @@ -package utils_test +package compression_test import ( "testing" - "github.com/NethermindEth/juno/utils" + "github.com/NethermindEth/juno/utils/compression" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -11,20 +11,20 @@ import ( func TestGzip64(t *testing.T) { bytes := []byte{0} expectedComBytes := "H4sIAAAAAAAA/2IABAAA//+N7wLSAQAAAA==" - comBytes, err := utils.Gzip64Encode(bytes) + comBytes, err := compression.Gzip64Encode(bytes) require.NoError(t, err) assert.Equal(t, comBytes, expectedComBytes) - decompBytes, err := utils.Gzip64Decode(comBytes) + decompBytes, err := compression.Gzip64Decode(comBytes) require.NoError(t, err) assert.Equal(t, bytes, decompBytes) } func FuzzGzip64(f *testing.F) { f.Fuzz(func(t *testing.T, data []byte) { - compressed, err := utils.Gzip64Encode(data) + compressed, err := compression.Gzip64Encode(data) require.NoError(t, err) - decompressed, err := utils.Gzip64Decode(compressed) + decompressed, err := compression.Gzip64Decode(compressed) require.NoError(t, err) assert.Equal(t, data, decompressed) }) From fe3531916037c70cf48ddca8602e32dc2ed240a7 Mon Sep 17 00:00:00 2001 From: Rodrigo Daniel Pino Trueba Date: Wed, 19 Aug 2026 14:37:38 +0100 Subject: [PATCH 2/5] feat(utils/compression): implement shared compression pool --- utils/compression/compression.go | 70 +++++++++- utils/compression/compression_bench_test.go | 91 +++++++++++++ utils/compression/compression_test.go | 135 ++++++++++++++++++++ 3 files changed, 295 insertions(+), 1 deletion(-) create mode 100644 utils/compression/compression_bench_test.go diff --git a/utils/compression/compression.go b/utils/compression/compression.go index 941d6c3805..fbc0de1d0b 100644 --- a/utils/compression/compression.go +++ b/utils/compression/compression.go @@ -6,11 +6,79 @@ import ( "encoding/base64" "fmt" "io" + "sync" ) +// Inclusive range of levels gzip accepts: HuffmanOnly (-2) to BestCompression (9). +const ( + minLevel = gzip.HuffmanOnly + maxLevel = gzip.BestCompression + levelCount = maxLevel - minLevel + 1 +) + +// Writer is a pooled gzip writer. It carries its compression level because +// gzip.Writer does not expose it and PutGzipWriter needs it to refile the writer. +type Writer struct { + *gzip.Writer + level int +} + +// gzipWriterPools holds one pool per compression level. Levels must not share a +// pool: Reset preserves the level a writer was built with, so a caller asking for +// one level could be handed another and silently compress at the wrong level. +// The pointer keeps the pools from being copied. +var gzipWriterPools = newGzipWriterPools() + +func newGzipWriterPools() *[levelCount]sync.Pool { + pools := new([levelCount]sync.Pool) + for i := range pools { + level := minLevel + i + pools[i].New = func() any { + // Only fails on an out-of-range level, and level comes from the index. + gzipWriter, err := gzip.NewWriterLevel(io.Discard, level) + if err != nil { + panic(fmt.Sprintf("compression: gzip writer at level %d: %v", level, err)) + } + return &Writer{Writer: gzipWriter, level: level} + } + } + return pools +} + +// poolIndex maps a level onto its pool. Callers pass constants, so a level gzip +// would reject is a programming error rather than a runtime condition. +func poolIndex(level int) int { + if level < minLevel || level > maxLevel { + panic(fmt.Sprintf("compression: gzip level %d outside [%d, %d]", level, minLevel, maxLevel)) + } + return level - minLevel +} + +// GzipWriter returns a gzip writer reset onto `dst`, compressing at the default +// level. Once used, it should be sent back to the pool via `PutGzipWriter` +func GzipWriter(dst io.Writer) *Writer { + return GzipWriterLevel(dst, gzip.DefaultCompression) +} + +// GzipWriterLevel returns a gzip writer reset onto `dst`, compressing at `level`. +// Once used, it should be sent back to the pool via `PutGzipWriter` +func GzipWriterLevel(dst io.Writer, level int) *Writer { + writer := gzipWriterPools[poolIndex(level)].Get().(*Writer) + writer.Reset(dst) + return writer +} + +// PutGzipWriter returns `writer` to the pool for its level. The writer need not be +// in a good state: Reset clears any sticky write error, so error paths should +// return it rather than drop it. +func PutGzipWriter(writer *Writer) { + gzipWriterPools[poolIndex(writer.level)].Put(writer) +} + func Gzip64Encode(data []byte) (string, error) { var compressedBuffer bytes.Buffer - gzipWriter := gzip.NewWriter(&compressedBuffer) + gzipWriter := GzipWriter(&compressedBuffer) + defer PutGzipWriter(gzipWriter) if _, err := gzipWriter.Write(data); err != nil { return "", fmt.Errorf("gzip data: %v", err) } diff --git a/utils/compression/compression_bench_test.go b/utils/compression/compression_bench_test.go new file mode 100644 index 0000000000..2ad70ee58b --- /dev/null +++ b/utils/compression/compression_bench_test.go @@ -0,0 +1,91 @@ +package compression_test + +import ( + "bytes" + "strconv" + "testing" + + "github.com/NethermindEth/juno/utils/compression" +) + +// Run with: go test -bench=. -benchmem -count=3 ./utils/compression/ +// +// Sizes bracket the deprecated Cairo 0 and sierra programs these helpers run on, +// which reach from tens of kilobytes to a few megabytes. The pooled writer removes +// a fixed ~800 KB allocation per call, so the payoff shows up in B/op at every +// size and in ns/op only at the small end, where that allocation is not yet +// dominated by compression itself. +var benchSizes = []int{16 << 10, 256 << 10, 2 << 20} + +// programLike builds a deterministic payload that compresses like the JSON class +// programs these helpers see in production. A payload of random bytes would be +// incompressible and would benchmark the wrong thing. +func programLike(size int) []byte { + tokens := []string{ + `{"prime":"0x800000000000011000000000000000000000000000000000000000000000001",`, + `"data":["0x480680017fff8000","0x1","0x48127ffe7fff8000","0x208b7fff7fff7ffe"],`, + `"attributes":[],"debug_info":null,"builtins":["range_check","pedersen"],`, + `"hints":{"0":[{"accessible_scopes":["starkware.cairo.common.math"]}]},`, + } + + var buf bytes.Buffer + buf.Grow(size) + for i := 0; buf.Len() < size; i++ { + buf.WriteString(tokens[i%len(tokens)]) + } + return buf.Bytes()[:size] +} + +func BenchmarkGzip64Encode(b *testing.B) { + for _, size := range benchSizes { + data := programLike(size) + b.Run(strconv.Itoa(size>>10)+"KiB", func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(int64(size)) + for b.Loop() { + if _, err := compression.Gzip64Encode(data); err != nil { + b.Fatal(err) + } + } + }) + } +} + +func BenchmarkGzip64Decode(b *testing.B) { + for _, size := range benchSizes { + encoded, err := compression.Gzip64Encode(programLike(size)) + if err != nil { + b.Fatal(err) + } + b.Run(strconv.Itoa(size>>10)+"KiB", func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(int64(size)) + for b.Loop() { + if _, err := compression.Gzip64Decode(encoded); err != nil { + b.Fatal(err) + } + } + }) + } +} + +// GzipWriter is also used directly by callers that stream into a destination +// rather than building a base64 string, e.g. the JSON-RPC HTTP handler. +func BenchmarkGzipWriter(b *testing.B) { + data := programLike(256 << 10) + var sink bytes.Buffer + + b.ReportAllocs() + b.SetBytes(int64(len(data))) + for b.Loop() { + sink.Reset() + gzipWriter := compression.GzipWriter(&sink) + if _, err := gzipWriter.Write(data); err != nil { + b.Fatal(err) + } + if err := gzipWriter.Close(); err != nil { + b.Fatal(err) + } + compression.PutGzipWriter(gzipWriter) + } +} diff --git a/utils/compression/compression_test.go b/utils/compression/compression_test.go index dc83b77da7..36ad2f2edd 100644 --- a/utils/compression/compression_test.go +++ b/utils/compression/compression_test.go @@ -1,6 +1,13 @@ package compression_test import ( + "bytes" + "compress/gzip" + "encoding/base64" + "errors" + "io" + "strconv" + "sync" "testing" "github.com/NethermindEth/juno/utils/compression" @@ -29,3 +36,131 @@ func FuzzGzip64(f *testing.F) { assert.Equal(t, data, decompressed) }) } + +// Successive encodes draw the same writer back out of the pool. The sizes below +// are deliberately not monotonic: a writer that carried state over from the +// previous encode would trail bytes of a longer payload into a shorter one, which +// a run of same-or-growing sizes would hide. +func TestGzip64EncodeAcrossSuccessiveCalls(t *testing.T) { + for _, size := range []int{4096, 2048, 64, 8192, 1} { + t.Run(strconv.Itoa(size), func(t *testing.T) { + payload := bytes.Repeat([]byte("a"), size) + + encoded, err := compression.Gzip64Encode(payload) + require.NoError(t, err) + decoded, err := compression.Gzip64Decode(encoded) + require.NoError(t, err) + assert.Equal(t, payload, decoded) + }) + } +} + +type failingWriter struct{} + +func (failingWriter) Write([]byte) (int, error) { + return 0, errors.New("destination unavailable") +} + +// A writer whose destination failed is still safe to hand back: the next caller +// gets output identical to what an untouched writer produces. +func TestGzipWriterAfterFailedDestination(t *testing.T) { + const repeats = 4096 + payload := bytes.Repeat([]byte("compress me "), repeats) + + poisoned := compression.GzipWriter(failingWriter{}) + _, writeErr := poisoned.Write(payload) + closeErr := poisoned.Close() + require.Error(t, errors.Join(writeErr, closeErr), "failing destination must fault the writer") + compression.PutGzipWriter(poisoned) + + var buf bytes.Buffer + reused := compression.GzipWriter(&buf) + _, err := reused.Write(payload) + require.NoError(t, err) + require.NoError(t, reused.Close()) + compression.PutGzipWriter(reused) + + expected, err := compression.Gzip64Encode(payload) + require.NoError(t, err) + assert.Equal(t, expected, base64.StdEncoding.EncodeToString(buf.Bytes())) +} + +// gzipAtLevel is the reference encoding: a writer built at `level` and used once, +// so no pool can influence the result. +func gzipAtLevel(t *testing.T, data []byte, level int) []byte { + t.Helper() + + var buf bytes.Buffer + gzipWriter, err := gzip.NewWriterLevel(&buf, level) + require.NoError(t, err) + _, err = gzipWriter.Write(data) + require.NoError(t, err) + require.NoError(t, gzipWriter.Close()) + return buf.Bytes() +} + +// Each level keeps its own writers. Returning a writer at one level must never +// let it come back out at another, which would silently compress at the wrong +// level: the only symptom is a differently sized output, never an error. +func TestGzipWriterLevelsDoNotMix(t *testing.T) { + const repeats = 8192 + payload := bytes.Repeat([]byte("mixed levels must not share writers "), repeats) + + levels := []int{gzip.BestSpeed, gzip.DefaultCompression, gzip.BestCompression} + + // Cycle the levels so every pool has been drawn from and returned to before the + // assertions below, giving a misfiled writer the chance to surface. + for range 2 { + for _, level := range levels { + compression.PutGzipWriter(compression.GzipWriterLevel(io.Discard, level)) + } + } + + for _, level := range levels { + t.Run(strconv.Itoa(level), func(t *testing.T) { + var buf bytes.Buffer + gzipWriter := compression.GzipWriterLevel(&buf, level) + _, err := gzipWriter.Write(payload) + require.NoError(t, err) + require.NoError(t, gzipWriter.Close()) + compression.PutGzipWriter(gzipWriter) + + assert.Equal(t, gzipAtLevel(t, payload, level), buf.Bytes()) + }) + } + + // Guard the premise: these levels really do produce different output, so the + // assertions above could actually fail if writers were shared. + fastest := gzipAtLevel(t, payload, gzip.BestSpeed) + smallest := gzipAtLevel(t, payload, gzip.BestCompression) + require.NotEqual(t, len(fastest), len(smallest), "levels must be distinguishable") +} + +func TestGzipWriterLevelRejectsOutOfRange(t *testing.T) { + assert.Panics(t, func() { compression.GzipWriterLevel(io.Discard, gzip.BestCompression+1) }) + assert.Panics(t, func() { compression.GzipWriterLevel(io.Discard, gzip.HuffmanOnly-1) }) +} + +// Concurrent callers must not end up sharing a writer. Run under -race. +func TestGzip64EncodeConcurrent(t *testing.T) { + const ( + goroutines = 16 + chunk = 512 + ) + + var wg sync.WaitGroup + for i := range goroutines { + wg.Add(1) + go func() { + defer wg.Done() + + payload := bytes.Repeat([]byte{byte('a' + i)}, chunk*(i+1)) + encoded, err := compression.Gzip64Encode(payload) + assert.NoError(t, err) + decoded, err := compression.Gzip64Decode(encoded) + assert.NoError(t, err) + assert.Equal(t, payload, decoded) + }() + } + wg.Wait() +} From b19d5e5cae1cd2163881747dff7f2d4941584e92 Mon Sep 17 00:00:00 2001 From: Rodrigo Daniel Pino Trueba Date: Wed, 19 Aug 2026 15:28:45 +0100 Subject: [PATCH 3/5] feat: integrate new compression pool to the rest of Juno --- clients/gateway/gateway.go | 5 +++-- jsonrpc/http.go | 14 +++----------- rpc/v10/transaction.go | 21 ++++++--------------- 3 files changed, 12 insertions(+), 28 deletions(-) diff --git a/clients/gateway/gateway.go b/clients/gateway/gateway.go index 0c6a9711e0..336784834f 100644 --- a/clients/gateway/gateway.go +++ b/clients/gateway/gateway.go @@ -2,7 +2,6 @@ package gateway import ( "bytes" - "compress/gzip" "context" "encoding/json" "errors" @@ -12,6 +11,7 @@ import ( "net/url" "time" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/log" ) @@ -146,7 +146,8 @@ func prepareRequestBody(jsonBody []byte) (io.Reader, bool, error) { } var buf bytes.Buffer - gzWriter := gzip.NewWriter(&buf) + gzWriter := compression.GzipWriter(&buf) + defer compression.PutGzipWriter(gzWriter) if _, err := gzWriter.Write(jsonBody); err != nil { return nil, false, fmt.Errorf("writing gzip content: %w", err) } diff --git a/jsonrpc/http.go b/jsonrpc/http.go index 4cdd8cc87e..843d188c06 100644 --- a/jsonrpc/http.go +++ b/jsonrpc/http.go @@ -1,7 +1,6 @@ package jsonrpc import ( - "compress/gzip" "context" "errors" "io" @@ -9,20 +8,14 @@ import ( "net/http" "strconv" "strings" - "sync" "time" "github.com/NethermindEth/juno/db" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/log" "go.uber.org/zap" ) -// gzipWriterPool holds gzip writers for reuse; -// callers must Reset a writer onto their destination before writing to it. -var gzipWriterPool = sync.Pool{ - New: func() any { return gzip.NewWriter(io.Discard) }, -} - type HTTP struct { rpc *Server logger log.StructuredLogger @@ -131,11 +124,10 @@ func (h *HTTP) ServeHTTP(writer http.ResponseWriter, req *http.Request) { var ioWriter io.Writer = writer if strings.Contains(req.Header.Get("Accept-Encoding"), "gzip") { writer.Header().Set("Content-Encoding", "gzip") - gw := gzipWriterPool.Get().(*gzip.Writer) - gw.Reset(writer) + gw := compression.GzipWriter(writer) defer func() { closeErr := gw.Close() - gzipWriterPool.Put(gw) + compression.PutGzipWriter(gw) if closeErr != nil { http.Error(writer, "gzip close error", http.StatusInternalServerError) return diff --git a/rpc/v10/transaction.go b/rpc/v10/transaction.go index 80112b5c3e..d24883fbce 100644 --- a/rpc/v10/transaction.go +++ b/rpc/v10/transaction.go @@ -8,7 +8,6 @@ import ( "encoding/json" "errors" "fmt" - "io" "sync" "github.com/NethermindEth/juno/adapters/sn2core" @@ -21,21 +20,14 @@ import ( "github.com/NethermindEth/juno/rpc/rpccore" "github.com/NethermindEth/juno/starknet" "github.com/NethermindEth/juno/starknet/compiler" + "github.com/NethermindEth/juno/utils/compression" "github.com/NethermindEth/juno/utils/throttler" "go.uber.org/zap" ) -var ( - gzPool = sync.Pool{ - New: func() any { - w, _ := gzip.NewWriterLevel(io.Discard, gzip.BestSpeed) - return w - }, - } - bufPool = sync.Pool{ - New: func() any { return new(bytes.Buffer) }, - } -) +var bufPool = sync.Pool{ + New: func() any { return new(bytes.Buffer) }, +} // AdaptTransaction adapts a core.Transaction to a local Transaction. // It's a wrapper around AdaptCoreTransaction that allows to exclude proof facts @@ -244,9 +236,8 @@ func ContractClassToGatewayPayload(class *ContractClass) ([]byte, error) { defer bufPool.Put(sierraBuf) b64 := base64.NewEncoder(base64.StdEncoding, sierraBuf) - gz := gzPool.Get().(*gzip.Writer) - gz.Reset(b64) - defer gzPool.Put(gz) + gz := compression.GzipWriterLevel(b64, gzip.BestSpeed) + defer compression.PutGzipWriter(gz) enc := json.NewEncoder(gz) enc.SetEscapeHTML(false) From c1386198a2b9316d5c63a5a61449eb8547aaec25 Mon Sep 17 00:00:00 2001 From: Rodrigo Daniel Pino Trueba Date: Wed, 19 Aug 2026 16:17:54 +0100 Subject: [PATCH 4/5] chore(compression): minor improvements --- clients/gateway/gateway.go | 2 +- jsonrpc/http.go | 2 +- rpc/v10/transaction.go | 2 +- utils/compression/compression.go | 102 ++++++++++++-------- utils/compression/compression_bench_test.go | 12 +-- utils/compression/compression_test.go | 49 +++++++--- 6 files changed, 105 insertions(+), 64 deletions(-) diff --git a/clients/gateway/gateway.go b/clients/gateway/gateway.go index 336784834f..a3010fdd1d 100644 --- a/clients/gateway/gateway.go +++ b/clients/gateway/gateway.go @@ -147,7 +147,7 @@ func prepareRequestBody(jsonBody []byte) (io.Reader, bool, error) { var buf bytes.Buffer gzWriter := compression.GzipWriter(&buf) - defer compression.PutGzipWriter(gzWriter) + defer gzWriter.Release() if _, err := gzWriter.Write(jsonBody); err != nil { return nil, false, fmt.Errorf("writing gzip content: %w", err) } diff --git a/jsonrpc/http.go b/jsonrpc/http.go index 843d188c06..9b0ea15fc4 100644 --- a/jsonrpc/http.go +++ b/jsonrpc/http.go @@ -127,7 +127,7 @@ func (h *HTTP) ServeHTTP(writer http.ResponseWriter, req *http.Request) { gw := compression.GzipWriter(writer) defer func() { closeErr := gw.Close() - compression.PutGzipWriter(gw) + gw.Release() if closeErr != nil { http.Error(writer, "gzip close error", http.StatusInternalServerError) return diff --git a/rpc/v10/transaction.go b/rpc/v10/transaction.go index d24883fbce..129995c452 100644 --- a/rpc/v10/transaction.go +++ b/rpc/v10/transaction.go @@ -237,7 +237,7 @@ func ContractClassToGatewayPayload(class *ContractClass) ([]byte, error) { b64 := base64.NewEncoder(base64.StdEncoding, sierraBuf) gz := compression.GzipWriterLevel(b64, gzip.BestSpeed) - defer compression.PutGzipWriter(gz) + defer gz.Release() enc := json.NewEncoder(gz) enc.SetEscapeHTML(false) diff --git a/utils/compression/compression.go b/utils/compression/compression.go index fbc0de1d0b..3dcd8fa20d 100644 --- a/utils/compression/compression.go +++ b/utils/compression/compression.go @@ -4,86 +4,110 @@ import ( "bytes" "compress/gzip" "encoding/base64" + "errors" "fmt" "io" "sync" ) -// Inclusive range of levels gzip accepts: HuffmanOnly (-2) to BestCompression (9). +// All gzip compression levels const ( minLevel = gzip.HuffmanOnly maxLevel = gzip.BestCompression levelCount = maxLevel - minLevel + 1 ) -// Writer is a pooled gzip writer. It carries its compression level because -// gzip.Writer does not expose it and PutGzipWriter needs it to refile the writer. -type Writer struct { - *gzip.Writer - level int -} +var ErrWriterNotAcquired = errors.New("using writer after release") -// gzipWriterPools holds one pool per compression level. Levels must not share a -// pool: Reset preserves the level a writer was built with, so a caller asking for -// one level could be handed another and silently compress at the wrong level. -// The pointer keeps the pools from being copied. -var gzipWriterPools = newGzipWriterPools() +// gzipWriterPools holds one pool per compression level. +var gzipWriterPools *[levelCount]sync.Pool = func() *[levelCount]sync.Pool { + pool := [levelCount]sync.Pool{} -func newGzipWriterPools() *[levelCount]sync.Pool { - pools := new([levelCount]sync.Pool) - for i := range pools { + for i := range pool { level := minLevel + i - pools[i].New = func() any { + pool[i].New = func() any { // Only fails on an out-of-range level, and level comes from the index. gzipWriter, err := gzip.NewWriterLevel(io.Discard, level) if err != nil { - panic(fmt.Sprintf("compression: gzip writer at level %d: %v", level, err)) + panic(fmt.Sprintf("creating new gzip writer for level %d: %v", level, err)) } - return &Writer{Writer: gzipWriter, level: level} + return &Writer{gz: gzipWriter} } } - return pools + + return &pool +}() + +// Writer is a pooled gzip writer. It references the pool it belongs to so it can +// track its own state +type Writer struct { + gz *gzip.Writer + pool *sync.Pool +} + +func (w *Writer) Write(p []byte) (int, error) { + if !w.isAcquired() { + return 0, ErrWriterNotAcquired + } + return w.gz.Write(p) +} + +func (w *Writer) Close() error { + if !w.isAcquired() { + return ErrWriterNotAcquired + } + return w.gz.Close() } -// poolIndex maps a level onto its pool. Callers pass constants, so a level gzip -// would reject is a programming error rather than a runtime condition. -func poolIndex(level int) int { - if level < minLevel || level > maxLevel { - panic(fmt.Sprintf("compression: gzip level %d outside [%d, %d]", level, minLevel, maxLevel)) +func (w *Writer) Flush() error { + if !w.isAcquired() { + return ErrWriterNotAcquired } - return level - minLevel + return w.gz.Flush() +} + +func (w *Writer) isAcquired() bool { + return w.pool != nil +} + +// Release returns the writer to the pool +func (w *Writer) Release() { + if w.pool == nil { + panic("re-releasing writer") + } + pool := w.pool + w.pool = nil + // Intentionally added to break all the references this writer is pointing too before + // putting it back to the pool. + w.gz.Reset(io.Discard) + pool.Put(w) } // GzipWriter returns a gzip writer reset onto `dst`, compressing at the default -// level. Once used, it should be sent back to the pool via `PutGzipWriter` +// level. Once used, it should be sent back to the pool via `Release` func GzipWriter(dst io.Writer) *Writer { return GzipWriterLevel(dst, gzip.DefaultCompression) } // GzipWriterLevel returns a gzip writer reset onto `dst`, compressing at `level`. -// Once used, it should be sent back to the pool via `PutGzipWriter` +// Once used, it should be sent back to the pool via `Release` func GzipWriterLevel(dst io.Writer, level int) *Writer { - writer := gzipWriterPools[poolIndex(level)].Get().(*Writer) - writer.Reset(dst) + pool := &gzipWriterPools[level-minLevel] + writer := pool.Get().(*Writer) + writer.pool = pool + writer.gz.Reset(dst) return writer } -// PutGzipWriter returns `writer` to the pool for its level. The writer need not be -// in a good state: Reset clears any sticky write error, so error paths should -// return it rather than drop it. -func PutGzipWriter(writer *Writer) { - gzipWriterPools[poolIndex(writer.level)].Put(writer) -} - func Gzip64Encode(data []byte) (string, error) { var compressedBuffer bytes.Buffer gzipWriter := GzipWriter(&compressedBuffer) - defer PutGzipWriter(gzipWriter) + defer gzipWriter.Release() if _, err := gzipWriter.Write(data); err != nil { - return "", fmt.Errorf("gzip data: %v", err) + return "", fmt.Errorf("writing data with gzip: %w", err) } if err := gzipWriter.Close(); err != nil { - return "", fmt.Errorf("close gzip writer: %v", err) + return "", fmt.Errorf("closing gzip writer: %w", err) } return base64.StdEncoding.EncodeToString(compressedBuffer.Bytes()), nil } diff --git a/utils/compression/compression_bench_test.go b/utils/compression/compression_bench_test.go index 2ad70ee58b..dd0919a413 100644 --- a/utils/compression/compression_bench_test.go +++ b/utils/compression/compression_bench_test.go @@ -8,18 +8,8 @@ import ( "github.com/NethermindEth/juno/utils/compression" ) -// Run with: go test -bench=. -benchmem -count=3 ./utils/compression/ -// -// Sizes bracket the deprecated Cairo 0 and sierra programs these helpers run on, -// which reach from tens of kilobytes to a few megabytes. The pooled writer removes -// a fixed ~800 KB allocation per call, so the payoff shows up in B/op at every -// size and in ns/op only at the small end, where that allocation is not yet -// dominated by compression itself. var benchSizes = []int{16 << 10, 256 << 10, 2 << 20} -// programLike builds a deterministic payload that compresses like the JSON class -// programs these helpers see in production. A payload of random bytes would be -// incompressible and would benchmark the wrong thing. func programLike(size int) []byte { tokens := []string{ `{"prime":"0x800000000000011000000000000000000000000000000000000000000000001",`, @@ -86,6 +76,6 @@ func BenchmarkGzipWriter(b *testing.B) { if err := gzipWriter.Close(); err != nil { b.Fatal(err) } - compression.PutGzipWriter(gzipWriter) + gzipWriter.Release() } } diff --git a/utils/compression/compression_test.go b/utils/compression/compression_test.go index 36ad2f2edd..394d702c32 100644 --- a/utils/compression/compression_test.go +++ b/utils/compression/compression_test.go @@ -3,7 +3,6 @@ package compression_test import ( "bytes" "compress/gzip" - "encoding/base64" "errors" "io" "strconv" @@ -20,7 +19,7 @@ func TestGzip64(t *testing.T) { expectedComBytes := "H4sIAAAAAAAA/2IABAAA//+N7wLSAQAAAA==" comBytes, err := compression.Gzip64Encode(bytes) require.NoError(t, err) - assert.Equal(t, comBytes, expectedComBytes) + assert.Equal(t, expectedComBytes, comBytes) decompBytes, err := compression.Gzip64Decode(comBytes) require.NoError(t, err) @@ -62,7 +61,9 @@ func (failingWriter) Write([]byte) (int, error) { } // A writer whose destination failed is still safe to hand back: the next caller -// gets output identical to what an untouched writer produces. +// gets output identical to what an untouched writer produces. The expected value +// comes from gzipAtLevel rather than Gzip64Encode so the oracle cannot itself be +// tainted by the pool under test. func TestGzipWriterAfterFailedDestination(t *testing.T) { const repeats = 4096 payload := bytes.Repeat([]byte("compress me "), repeats) @@ -71,18 +72,36 @@ func TestGzipWriterAfterFailedDestination(t *testing.T) { _, writeErr := poisoned.Write(payload) closeErr := poisoned.Close() require.Error(t, errors.Join(writeErr, closeErr), "failing destination must fault the writer") - compression.PutGzipWriter(poisoned) + poisoned.Release() var buf bytes.Buffer reused := compression.GzipWriter(&buf) _, err := reused.Write(payload) require.NoError(t, err) require.NoError(t, reused.Close()) - compression.PutGzipWriter(reused) + reused.Release() - expected, err := compression.Gzip64Encode(payload) - require.NoError(t, err) - assert.Equal(t, expected, base64.StdEncoding.EncodeToString(buf.Bytes())) + assert.Equal(t, gzipAtLevel(t, payload, gzip.DefaultCompression), buf.Bytes()) +} + +// Releasing twice would hand the same writer to two callers at once, so the +// second Release must fail loudly rather than corrupt a concurrent caller. +func TestGzipWriterDoubleReleasePanics(t *testing.T) { + gzipWriter := compression.GzipWriter(io.Discard) + gzipWriter.Release() + assert.Panics(t, func() { gzipWriter.Release() }) +} + +// A released writer may already be owned by another goroutine, so using it must +// be refused rather than silently corrupt the new owner's output. +func TestGzipWriterUseAfterReleaseErrors(t *testing.T) { + gzipWriter := compression.GzipWriter(io.Discard) + gzipWriter.Release() + + _, err := gzipWriter.Write([]byte("stale")) + assert.ErrorIs(t, err, compression.ErrWriterNotAcquired) + assert.ErrorIs(t, gzipWriter.Close(), compression.ErrWriterNotAcquired) + assert.ErrorIs(t, gzipWriter.Flush(), compression.ErrWriterNotAcquired) } // gzipAtLevel is the reference encoding: a writer built at `level` and used once, @@ -106,13 +125,21 @@ func TestGzipWriterLevelsDoNotMix(t *testing.T) { const repeats = 8192 payload := bytes.Repeat([]byte("mixed levels must not share writers "), repeats) - levels := []int{gzip.BestSpeed, gzip.DefaultCompression, gzip.BestCompression} + // HuffmanOnly and NoCompression are included because they take a different + // reset path inside flate than the levels that build a match chain. + levels := []int{ + gzip.HuffmanOnly, + gzip.NoCompression, + gzip.BestSpeed, + gzip.DefaultCompression, + gzip.BestCompression, + } // Cycle the levels so every pool has been drawn from and returned to before the // assertions below, giving a misfiled writer the chance to surface. for range 2 { for _, level := range levels { - compression.PutGzipWriter(compression.GzipWriterLevel(io.Discard, level)) + compression.GzipWriterLevel(io.Discard, level).Release() } } @@ -123,7 +150,7 @@ func TestGzipWriterLevelsDoNotMix(t *testing.T) { _, err := gzipWriter.Write(payload) require.NoError(t, err) require.NoError(t, gzipWriter.Close()) - compression.PutGzipWriter(gzipWriter) + gzipWriter.Release() assert.Equal(t, gzipAtLevel(t, payload, level), buf.Bytes()) }) From b892ba822046e7fad539afbb43cc66ee1acd2fed Mon Sep 17 00:00:00 2001 From: Rodrigo Daniel Pino Trueba Date: Thu, 20 Aug 2026 16:23:36 +0100 Subject: [PATCH 5/5] chore(utils): add inderaction to writer dst To get chepaer O(1) Writer release --- utils/compression/compression.go | 71 ++++++++++++++++++--------- utils/compression/compression_test.go | 29 +++++++++-- 2 files changed, 72 insertions(+), 28 deletions(-) diff --git a/utils/compression/compression.go b/utils/compression/compression.go index 3dcd8fa20d..b2ddb182b5 100644 --- a/utils/compression/compression.go +++ b/utils/compression/compression.go @@ -23,26 +23,52 @@ var ErrWriterNotAcquired = errors.New("using writer after release") var gzipWriterPools *[levelCount]sync.Pool = func() *[levelCount]sync.Pool { pool := [levelCount]sync.Pool{} - for i := range pool { - level := minLevel + i - pool[i].New = func() any { - // Only fails on an out-of-range level, and level comes from the index. - gzipWriter, err := gzip.NewWriterLevel(io.Discard, level) - if err != nil { - panic(fmt.Sprintf("creating new gzip writer for level %d: %v", level, err)) - } - return &Writer{gz: gzipWriter} - } + for i := range levelCount { + pool[i].New = func() any { return newWriter(minLevel + i) } } return &pool }() -// Writer is a pooled gzip writer. It references the pool it belongs to so it can -// track its own state +// proxy exists to solve the problem of putting a [Writer] `w` back to the pool +// (via [Writer.Release]) while `w` still references live data that is not going to be used +// again. There is the option to drop reference with `w.gz.Reset(io.Discard)` +// but it's expensive (cost in the micro-seconds). +// With `proxy` we create a middle pointer to which `w.gz` will point. +// `proxy` in turn will point to the actual data. +// When releasing a [Writer] it is enough for the proxy to drop the reference. +// When acquired: [Writer.gz] ---> [Writer.proxy] ---> data +// After release: [Writer.gz] ---> [Writer.proxy] ---> nil +type proxy struct { + dst io.Writer +} + +func (d *proxy) Write(p []byte) (int, error) { + if d.dst == nil { + return 0, ErrWriterNotAcquired + } + return d.dst.Write(p) +} + +// Writer is a pooled gzip writer. Its gzip writer is permanently wired to +// proxy, which forwards writes to the caller's dst while the writer is +// acquired. The indirection lets Release detach the caller's destination in +// O(1) instead of paying a second flate reset just to re-point the gzip +// writer at io.Discard. type Writer struct { - gz *gzip.Writer - pool *sync.Pool + gz *gzip.Writer + proxy proxy // dst of gz + pool *sync.Pool // pool this writer belongs to; nil once released +} + +func newWriter(level int) *Writer { + writer := &Writer{} + gzipWriter, err := gzip.NewWriterLevel(&writer.proxy, level) + if err != nil { + panic(fmt.Sprintf("creating new gzip writer for level %d: %v", level, err)) + } + writer.gz = gzipWriter + return writer } func (w *Writer) Write(p []byte) (int, error) { @@ -66,23 +92,21 @@ func (w *Writer) Flush() error { return w.gz.Flush() } -func (w *Writer) isAcquired() bool { - return w.pool != nil -} - // Release returns the writer to the pool func (w *Writer) Release() { - if w.pool == nil { + if !w.isAcquired() { panic("re-releasing writer") } pool := w.pool w.pool = nil - // Intentionally added to break all the references this writer is pointing too before - // putting it back to the pool. - w.gz.Reset(io.Discard) + w.proxy.dst = nil pool.Put(w) } +func (w *Writer) isAcquired() bool { + return w.pool != nil +} + // GzipWriter returns a gzip writer reset onto `dst`, compressing at the default // level. Once used, it should be sent back to the pool via `Release` func GzipWriter(dst io.Writer) *Writer { @@ -95,7 +119,8 @@ func GzipWriterLevel(dst io.Writer, level int) *Writer { pool := &gzipWriterPools[level-minLevel] writer := pool.Get().(*Writer) writer.pool = pool - writer.gz.Reset(dst) + writer.proxy.dst = dst + writer.gz.Reset(&writer.proxy) return writer } diff --git a/utils/compression/compression_test.go b/utils/compression/compression_test.go index 394d702c32..d9d1d06671 100644 --- a/utils/compression/compression_test.go +++ b/utils/compression/compression_test.go @@ -5,9 +5,11 @@ import ( "compress/gzip" "errors" "io" + "runtime" "strconv" "sync" "testing" + "weak" "github.com/NethermindEth/juno/utils/compression" "github.com/stretchr/testify/assert" @@ -84,6 +86,26 @@ func TestGzipWriterAfterFailedDestination(t *testing.T) { assert.Equal(t, gzipAtLevel(t, payload, gzip.DefaultCompression), buf.Bytes()) } +// After Release the pooled writer must hold no reference to the caller's +// destination: the destination has to be collectable while the writer lives on +// in the pool. Holding gzipWriter across the GC is deliberate — it pins the +// writer so the assertion is about the dst reference, not the writer itself +// being collected. +func TestGzipWriterReleaseDropsDestination(t *testing.T) { + dst := &bytes.Buffer{} + gzipWriter := compression.GzipWriter(dst) + _, err := gzipWriter.Write([]byte("payload")) + require.NoError(t, err) + require.NoError(t, gzipWriter.Close()) + + weakDst := weak.Make(dst) + gzipWriter.Release() + dst = nil //nolint:ineffassign // drop the last strong reference so GC can collect the buffer + runtime.GC() + assert.Nil(t, weakDst.Value(), "released writer still references its destination") + runtime.KeepAlive(gzipWriter) +} + // Releasing twice would hand the same writer to two callers at once, so the // second Release must fail loudly rather than corrupt a concurrent caller. func TestGzipWriterDoubleReleasePanics(t *testing.T) { @@ -177,17 +199,14 @@ func TestGzip64EncodeConcurrent(t *testing.T) { var wg sync.WaitGroup for i := range goroutines { - wg.Add(1) - go func() { - defer wg.Done() - + wg.Go(func() { payload := bytes.Repeat([]byte{byte('a' + i)}, chunk*(i+1)) encoded, err := compression.Gzip64Encode(payload) assert.NoError(t, err) decoded, err := compression.Gzip64Decode(encoded) assert.NoError(t, err) assert.Equal(t, payload, decoded) - }() + }) } wg.Wait() }