From 02d6528afaa4236885a575899240ccfe646a3c15 Mon Sep 17 00:00:00 2001 From: mxssl Date: Wed, 26 Aug 2026 14:03:25 +0000 Subject: [PATCH] feat: add zstd support for http ingress and upstream connectors Both edges only ever spoke gzip. Adds zstd alongside it: the ingress negotiates a coding per client, and the connectors offer zstd to nodes and decode whatever comes back. Introduces internal/compression, shared by both edges, holding the q-value negotiation and the pooled codecs. On the upstream side Go's transport used to negotiate gzip on its own, which is the only coding it knows. Offering zstd means setting Accept-Encoding explicitly, which turns that off, so the connector now owns decoding for both codings. Also drops Content-Encoding from a decompressed request, which was being forwarded to upstreams describing a body that is no longer compressed, and makes the CORS handler add to Vary rather than replace it. --- internal/compression/compression.go | 86 ++++++++ internal/compression/compression_test.go | 40 ++++ internal/compression/reader.go | 166 +++++++++++++++ internal/compression/reader_test.go | 151 ++++++++++++++ internal/compression/writer.go | 94 +++++++++ internal/compression/writer_test.go | 100 +++++++++ internal/server/http_server/compress.go | 189 ++++++++---------- internal/server/http_server/compress_test.go | 96 +++++++++ internal/server/http_server/cors_vary_test.go | 44 ++++ internal/server/http_server/decompress.go | 52 +++++ .../server/http_server/decompress_test.go | 156 +++++++++++++++ internal/server/http_server/http_server.go | 11 +- .../upstreams/connectors/http_connector.go | 75 ++++++- .../http_connector_compression_test.go | 159 +++++++++++++++ .../connectors/http_connector_test.go | 15 +- 15 files changed, 1311 insertions(+), 123 deletions(-) create mode 100644 internal/compression/compression.go create mode 100644 internal/compression/compression_test.go create mode 100644 internal/compression/reader.go create mode 100644 internal/compression/reader_test.go create mode 100644 internal/compression/writer.go create mode 100644 internal/compression/writer_test.go create mode 100644 internal/server/http_server/compress_test.go create mode 100644 internal/server/http_server/cors_vary_test.go create mode 100644 internal/server/http_server/decompress.go create mode 100644 internal/server/http_server/decompress_test.go create mode 100644 internal/upstreams/connectors/http_connector_compression_test.go diff --git a/internal/compression/compression.go b/internal/compression/compression.go new file mode 100644 index 00000000..458f8c87 --- /dev/null +++ b/internal/compression/compression.go @@ -0,0 +1,86 @@ +// Package compression implements the HTTP content codings nodecore speaks on +// both of its edges: the client-facing ingress and the upstream connectors. +// +// Both edges need the same three things - decide which coding to use, encode +// a body, decode a body - so the codec pools live here once rather than in +// each edge. Levels are fixed at the fastest setting of each codec: a proxy +// pays the compression cost on the hot path of every request, where CPU time +// costs more than the extra few percent of ratio. +package compression + +import ( + "strconv" + "strings" +) + +// Scheme is a content coding nodecore can encode and decode. +type Scheme string + +const ( + // Identity means "no compression"; it is the zero value so an + // unparseable or absent Accept-Encoding degrades to plain bodies. + Identity Scheme = "" + Gzip Scheme = "gzip" + Zstd Scheme = "zstd" +) + +// Offer is the Accept-Encoding nodecore sends upstream. zstd leads on +// preference, but a node that knows neither simply answers identity - content +// negotiation degrades on its own, which is why this needs no config knob. +const Offer = "zstd, gzip" + +// Negotiate picks the coding to encode a response with, given the client's +// Accept-Encoding (RFC 9110 §12.5.3). The highest q wins; zstd breaks a tie +// because it decodes faster and compresses denser than gzip at these levels. +// Anything unrecognised, refused with q=0, or absent yields Identity. +func Negotiate(acceptEncoding string) Scheme { + if acceptEncoding == "" { + return Identity + } + + best, bestQ := Identity, 0.0 + for _, part := range strings.Split(acceptEncoding, ",") { + coding, quality := parseCoding(part) + if quality == 0 { + continue + } + switch coding { + case "zstd", "*": + coding = string(Zstd) + case "gzip": + // keep + default: + continue + } + // A strictly higher q always wins; an equal q only promotes zstd, so + // the tie-break can never demote a coding the client ranked higher. + if quality > bestQ || (quality == bestQ && Scheme(coding) == Zstd) { + best, bestQ = Scheme(coding), quality + } + } + return best +} + +// parseCoding splits one Accept-Encoding element into its coding name and its +// q value. A missing or malformed q means q=1: a client that garbled the +// parameter still asked for the coding, and treating that as a refusal would +// silently drop compression instead of failing loudly. +func parseCoding(part string) (string, float64) { + name, params, hasParams := strings.Cut(part, ";") + name = strings.ToLower(strings.TrimSpace(name)) + if !hasParams { + return name, 1 + } + for _, param := range strings.Split(params, ";") { + key, value, ok := strings.Cut(param, "=") + if !ok || strings.ToLower(strings.TrimSpace(key)) != "q" { + continue + } + quality, err := strconv.ParseFloat(strings.TrimSpace(value), 64) + if err != nil { + return name, 1 + } + return name, quality + } + return name, 1 +} diff --git a/internal/compression/compression_test.go b/internal/compression/compression_test.go new file mode 100644 index 00000000..e350ab37 --- /dev/null +++ b/internal/compression/compression_test.go @@ -0,0 +1,40 @@ +package compression_test + +import ( + "testing" + + "github.com/drpcorg/nodecore/internal/compression" + "github.com/stretchr/testify/assert" +) + +// Negotiate picks the response coding from a client's Accept-Encoding. +// Highest q wins; zstd breaks a tie because it is both faster to decode +// and denser than gzip at comparable levels. +func TestNegotiate(t *testing.T) { + tests := []struct { + name string + acceptEncoding string + expected compression.Scheme + }{ + {"no header means no compression", "", compression.Identity}, + {"gzip only", "gzip", compression.Gzip}, + {"zstd only", "zstd", compression.Zstd}, + {"both offered, zstd wins the tie", "gzip, zstd", compression.Zstd}, + {"both offered, order does not matter", "zstd, gzip", compression.Zstd}, + {"zstd explicitly refused", "zstd;q=0, gzip", compression.Gzip}, + {"higher q wins over the tie-break", "gzip;q=0.5, zstd;q=0.1", compression.Gzip}, + {"unsupported codings are ignored", "br, deflate", compression.Identity}, + {"wildcard offers everything", "*", compression.Zstd}, + {"identity is not a compression", "identity", compression.Identity}, + {"coding names are case-insensitive", "GZIP", compression.Gzip}, + {"everything refused", "zstd;q=0, gzip;q=0", compression.Identity}, + {"malformed q is treated as acceptable", "gzip;q=abc", compression.Gzip}, + {"whitespace around parameters", " zstd ; q=0.9 , gzip ", compression.Gzip}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + assert.Equal(te, tt.expected, compression.Negotiate(tt.acceptEncoding)) + }) + } +} diff --git a/internal/compression/reader.go b/internal/compression/reader.go new file mode 100644 index 00000000..b8388f9b --- /dev/null +++ b/internal/compression/reader.go @@ -0,0 +1,166 @@ +package compression + +import ( + "bufio" + "encoding/binary" + "errors" + "fmt" + "io" + "strings" + "sync" + + "github.com/klauspost/compress/gzip" + "github.com/klauspost/compress/zstd" +) + +// ErrUnsupportedEncoding reports a Content-Encoding nodecore cannot decode. +// It is never the client's fault on the upstream edge - nodecore offers only +// the codings in Offer, so anything else is a misbehaving node. +var ErrUnsupportedEncoding = errors.New("unsupported content encoding") + +// decoderMaxWindow caps the zstd window a decoder will allocate for. The +// library defaults to 64GiB, which lets a hostile or broken peer name a +// window far larger than any real HTTP body needs and make nodecore allocate +// it. Real encoders top out at 8MiB at these levels, so 64MiB accepts +// everything legitimate with room to spare. +const decoderMaxWindow = 64 << 20 + +// Frame magic numbers from RFC 8878 §3.1: one for a regular frame, and a +// range for skippable frames, which a stream is allowed to lead with. +const ( + zstdMagicSize = 4 + zstdPeekSize = 512 + zstdFrameMagic = 0xFD2FB528 + zstdSkippableMagicMin = 0x184D2A50 + zstdSkippableMagicMax = 0x184D2A5F +) + +var gzipReaderPool = sync.Pool{ + New: func() any { return new(gzip.Reader) }, +} + +// Decoders are pooled rather than created per response: a zstd decoder +// allocates its window up front, which is far too expensive to repeat on +// every proxied request. Concurrency 1 keeps a pooled decoder to a single +// synchronous worker instead of one goroutine per GOMAXPROCS per decoder. +var zstdDecoderPool = sync.Pool{ + New: func() any { + decoder, err := zstd.NewReader( + nil, + zstd.WithDecoderConcurrency(1), + zstd.WithDecoderMaxWindow(decoderMaxWindow), + ) + if err != nil { + return err + } + return decoder + }, +} + +// WrapReader returns a reader that decodes r according to contentEncoding. +// An empty or identity encoding passes r through untouched. +// +// The returned Close releases the pooled codec and MUST be called; it does +// not close r, whose lifetime stays with the caller. +func WrapReader(contentEncoding string, r io.Reader) (io.ReadCloser, error) { + switch Scheme(strings.ToLower(strings.TrimSpace(contentEncoding))) { + case Identity, "identity": + return io.NopCloser(r), nil + case Gzip: + return wrapGzipReader(r) + case Zstd: + return wrapZstdReader(r) + default: + return nil, fmt.Errorf("%w: %q", ErrUnsupportedEncoding, contentEncoding) + } +} + +func wrapGzipReader(r io.Reader) (io.ReadCloser, error) { + pooled := gzipReaderPool.Get() + reader, ok := pooled.(*gzip.Reader) + if !ok { + return nil, fmt.Errorf("cannot take a gzip reader from the pool: %w", pooled.(error)) + } + // Reset parses the gzip header eagerly, so a body that is not gzip at all + // fails here rather than halfway through the caller's first Read. + if err := reader.Reset(r); err != nil { + gzipReaderPool.Put(reader) + return nil, fmt.Errorf("invalid gzip body: %w", err) + } + return &pooledReader{ + Reader: reader, + release: func() { + // Close drops the reference to r without touching r itself, so a + // pooled reader never pins a finished response body. + _ = reader.Close() + gzipReaderPool.Put(reader) + }, + }, nil +} + +func wrapZstdReader(r io.Reader) (io.ReadCloser, error) { + // gzip validates its header the moment the reader is reset, so a body + // that is not gzip is rejected before anyone reads it. zstd starts + // decoding lazily, which would push the same mistake out to the caller's + // first Read - as a read failure, long after the context that could + // explain it. Checking the frame magic here restores the symmetry. + buffered := bufio.NewReaderSize(r, zstdPeekSize) + if err := checkZstdMagic(buffered); err != nil { + return nil, err + } + + pooled := zstdDecoderPool.Get() + decoder, ok := pooled.(*zstd.Decoder) + if !ok { + return nil, fmt.Errorf("cannot take a zstd decoder from the pool: %w", pooled.(error)) + } + if err := decoder.Reset(buffered); err != nil { + return nil, fmt.Errorf("invalid zstd body: %w", err) + } + return &pooledReader{ + Reader: decoder, + release: func() { + // Reset(nil) drains any undelivered output and drops r. Close is + // deliberately not called: it retires the decoder permanently, + // which would defeat the pool. + _ = decoder.Reset(nil) + zstdDecoderPool.Put(decoder) + }, + }, nil +} + +// checkZstdMagic reports whether the stream opens with a zstd frame header, +// without consuming it. An empty body is not a malformed frame: a zero-byte +// payload decodes to zero bytes. +func checkZstdMagic(r *bufio.Reader) error { + header, err := r.Peek(zstdMagicSize) + if errors.Is(err, io.EOF) && len(header) == 0 { + return nil + } + if err != nil { + return fmt.Errorf("invalid zstd body: cannot read the frame header: %w", err) + } + magic := binary.LittleEndian.Uint32(header) + if magic == zstdFrameMagic { + return nil + } + if magic >= zstdSkippableMagicMin && magic <= zstdSkippableMagicMax { + return nil + } + return fmt.Errorf("invalid zstd body: frame magic %#08x is not zstd", magic) +} + +// pooledReader hands a decoder back to its pool on Close. Close is idempotent +// because a streaming response can be torn down from both the read side and +// an explicit teardown, and returning one decoder to the pool twice would let +// two requests decode through the same one. +type pooledReader struct { + io.Reader + release func() + once sync.Once +} + +func (p *pooledReader) Close() error { + p.once.Do(p.release) + return nil +} diff --git a/internal/compression/reader_test.go b/internal/compression/reader_test.go new file mode 100644 index 00000000..1d6dfe18 --- /dev/null +++ b/internal/compression/reader_test.go @@ -0,0 +1,151 @@ +package compression_test + +import ( + "bytes" + "io" + "testing" + + "github.com/drpcorg/nodecore/internal/compression" + "github.com/klauspost/compress/gzip" + "github.com/klauspost/compress/zstd" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func gzipBytes(t *testing.T, plain []byte) []byte { + t.Helper() + var buf bytes.Buffer + w := gzip.NewWriter(&buf) + _, err := w.Write(plain) + require.NoError(t, err) + require.NoError(t, w.Close()) + return buf.Bytes() +} + +func zstdBytes(t *testing.T, plain []byte) []byte { + t.Helper() + var buf bytes.Buffer + w, err := zstd.NewWriter(&buf) + require.NoError(t, err) + _, err = w.Write(plain) + require.NoError(t, err) + require.NoError(t, w.Close()) + return buf.Bytes() +} + +func TestWrapReaderDecodesSupportedCodings(t *testing.T) { + plain := []byte(`{"jsonrpc":"2.0","id":1,"result":"0x10"}`) + tests := []struct { + name string + contentEncoding string + body []byte + }{ + {"gzip", "gzip", gzipBytes(t, plain)}, + {"zstd", "zstd", zstdBytes(t, plain)}, + {"case-insensitive", "ZSTD", zstdBytes(t, plain)}, + {"no encoding is passed through", "", plain}, + {"identity is passed through", "identity", plain}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + reader, err := compression.WrapReader(tt.contentEncoding, bytes.NewReader(tt.body)) + require.NoError(te, err) + defer func() { require.NoError(te, reader.Close()) }() + + got, err := io.ReadAll(reader) + + require.NoError(te, err) + assert.Equal(te, plain, got) + }) + } +} + +// An upstream answering a coding nodecore never offered must be reported, not +// passed on: the body would reach the client as bytes it cannot read, and the +// connector strips Content-Encoding so it would not even know why. +func TestWrapReaderRejectsUnsupportedCodings(t *testing.T) { + for _, contentEncoding := range []string{"br", "deflate", "gzip, gzip"} { + t.Run(contentEncoding, func(te *testing.T) { + _, err := compression.WrapReader(contentEncoding, bytes.NewReader(nil)) + + assert.ErrorIs(te, err, compression.ErrUnsupportedEncoding) + }) + } +} + +// Codecs are pooled, so a reader returned by Close must come back clean: +// a decoder still holding the previous stream's state produces garbage on +// its next use. +func TestWrapReaderIsReusableAfterClose(t *testing.T) { + for _, scheme := range []string{"gzip", "zstd"} { + t.Run(scheme, func(te *testing.T) { + for _, plain := range [][]byte{[]byte("first body"), []byte("a completely different second body")} { + var body []byte + if scheme == "gzip" { + body = gzipBytes(te, plain) + } else { + body = zstdBytes(te, plain) + } + + reader, err := compression.WrapReader(scheme, bytes.NewReader(body)) + require.NoError(te, err) + got, err := io.ReadAll(reader) + require.NoError(te, err) + require.NoError(te, reader.Close()) + + assert.Equal(te, plain, got) + } + }) + } +} + +// A truncated frame is a broken upstream, not a panic: the error must reach +// the caller so the request fails cleanly. +func TestWrapReaderReportsCorruptBody(t *testing.T) { + truncated := zstdBytes(t, bytes.Repeat([]byte("x"), 1024))[:20] + + reader, err := compression.WrapReader("zstd", bytes.NewReader(truncated)) + require.NoError(t, err) + defer func() { _ = reader.Close() }() + + _, err = io.ReadAll(reader) + + assert.Error(t, err) +} + +// A frame that does not start with zstd's magic number is not zstd at all, +// and saying so at wrap time turns a peer's mislabelled body into a clean +// rejection instead of an error surfacing mid-read from somewhere deeper. +func TestWrapReaderRejectsBodyThatIsNotZstd(t *testing.T) { + tests := []struct { + name string + body []byte + }{ + {"plain json", []byte(`{"jsonrpc":"2.0","id":1}`)}, + {"gzip bytes under a zstd label", gzipBytes(t, []byte("hello"))}, + {"magic truncated", zstdBytes(t, []byte("hello"))[:2]}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + _, err := compression.WrapReader("zstd", bytes.NewReader(tt.body)) + + assert.Error(te, err) + assert.NotErrorIs(te, err, compression.ErrUnsupportedEncoding, + "the coding is supported; it is this body that is wrong") + }) + } +} + +// An empty body is a legitimate zero-byte payload, not a malformed frame. +func TestWrapReaderAcceptsEmptyZstdBody(t *testing.T) { + reader, err := compression.WrapReader("zstd", bytes.NewReader(nil)) + require.NoError(t, err) + defer func() { _ = reader.Close() }() + + got, err := io.ReadAll(reader) + + require.NoError(t, err) + assert.Empty(t, got) +} diff --git a/internal/compression/writer.go b/internal/compression/writer.go new file mode 100644 index 00000000..129326a6 --- /dev/null +++ b/internal/compression/writer.go @@ -0,0 +1,94 @@ +package compression + +import ( + "fmt" + "io" + "sync" + + "github.com/klauspost/compress/gzip" + "github.com/klauspost/compress/zstd" +) + +// encoderWindow caps the back-reference distance of a pooled zstd encoder, +// and with it the memory each one holds while idle in the pool. JSON-RPC and +// REST bodies repeat within a few kilobytes - method names, hex prefixes, key +// names - so the default multi-megabyte window would buy almost no ratio for +// memory multiplied by every encoder in flight. +const encoderWindow = 256 << 10 + +var gzipWriterPool = sync.Pool{ + New: func() any { + // BestSpeed is what the ingress has always used: on a proxy the + // compression sits on the request's critical path, so CPU time is + // worth more than the last few percent of ratio. + writer, err := gzip.NewWriterLevel(io.Discard, gzip.BestSpeed) + if err != nil { + return err + } + return writer + }, +} + +var zstdEncoderPool = sync.Pool{ + New: func() any { + encoder, err := zstd.NewWriter( + io.Discard, + zstd.WithEncoderLevel(zstd.SpeedFastest), + // Concurrency 1 keeps an encoder to one synchronous worker. The + // default spawns GOMAXPROCS goroutines per encoder, which on a + // proxy holding thousands of concurrent responses is a goroutine + // count nobody asked for. + zstd.WithEncoderConcurrency(1), + zstd.WithWindowSize(encoderWindow), + ) + if err != nil { + return err + } + return encoder + }, +} + +// Writer is a compressing writer for one response body. Both pooled codecs +// satisfy it natively. +type Writer interface { + io.WriteCloser + // Flush pushes everything written so far to the underlying writer, so a + // streamed chunk reaches the client without waiting for Close. + Flush() error + // Reset redirects the writer, discarding any state from a previous body. + Reset(w io.Writer) +} + +// AcquireWriter takes a pooled encoder for scheme, encoding into w. The +// caller must Close it to terminate the stream and then ReleaseWriter it. +// Identity is not an encoder and is rejected. +func AcquireWriter(scheme Scheme, w io.Writer) (Writer, error) { + var pooled any + switch scheme { + case Gzip: + pooled = gzipWriterPool.Get() + case Zstd: + pooled = zstdEncoderPool.Get() + default: + return nil, fmt.Errorf("%w: no encoder for %q", ErrUnsupportedEncoding, scheme) + } + writer, ok := pooled.(Writer) + if !ok { + return nil, fmt.Errorf("cannot take a %s writer from the pool: %w", scheme, pooled.(error)) + } + writer.Reset(w) + return writer, nil +} + +// ReleaseWriter returns a writer to its pool. It resets the writer onto +// io.Discard first so a pooled encoder never pins the response it just +// finished writing to. +func ReleaseWriter(w Writer) { + w.Reset(io.Discard) + switch writer := w.(type) { + case *gzip.Writer: + gzipWriterPool.Put(writer) + case *zstd.Encoder: + zstdEncoderPool.Put(writer) + } +} diff --git a/internal/compression/writer_test.go b/internal/compression/writer_test.go new file mode 100644 index 00000000..702f6922 --- /dev/null +++ b/internal/compression/writer_test.go @@ -0,0 +1,100 @@ +package compression_test + +import ( + "bytes" + "io" + "testing" + + "github.com/drpcorg/nodecore/internal/compression" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAcquireWriterRoundTrips(t *testing.T) { + plain := []byte(`{"jsonrpc":"2.0","id":1,"result":{"number":"0x1337"}}`) + + for _, scheme := range []compression.Scheme{compression.Gzip, compression.Zstd} { + t.Run(string(scheme), func(te *testing.T) { + var buf bytes.Buffer + writer, err := compression.AcquireWriter(scheme, &buf) + require.NoError(te, err) + + _, err = writer.Write(plain) + require.NoError(te, err) + require.NoError(te, writer.Close()) + compression.ReleaseWriter(writer) + + reader, err := compression.WrapReader(string(scheme), &buf) + require.NoError(te, err) + defer func() { require.NoError(te, reader.Close()) }() + got, err := io.ReadAll(reader) + + require.NoError(te, err) + assert.Equal(te, plain, got) + }) + } +} + +// Identity has no encoder. Asking for one is a caller bug - the middleware +// must decide not to compress before it reaches for a writer - so it fails +// loudly rather than silently handing back a passthrough. +func TestAcquireWriterRejectsIdentity(t *testing.T) { + _, err := compression.AcquireWriter(compression.Identity, io.Discard) + + assert.ErrorIs(t, err, compression.ErrUnsupportedEncoding) +} + +// Encoders are pooled, so one released mid-stream must not leak its state +// into the next response that picks it up. +func TestAcquireWriterIsReusableAfterRelease(t *testing.T) { + for _, scheme := range []compression.Scheme{compression.Gzip, compression.Zstd} { + t.Run(string(scheme), func(te *testing.T) { + for _, plain := range [][]byte{[]byte("first response"), []byte("an entirely different second response")} { + var buf bytes.Buffer + writer, err := compression.AcquireWriter(scheme, &buf) + require.NoError(te, err) + _, err = writer.Write(plain) + require.NoError(te, err) + require.NoError(te, writer.Close()) + compression.ReleaseWriter(writer) + + reader, err := compression.WrapReader(string(scheme), &buf) + require.NoError(te, err) + got, err := io.ReadAll(reader) + require.NoError(te, err) + require.NoError(te, reader.Close()) + + assert.Equal(te, plain, got) + } + }) + } +} + +// Streaming responses are flushed chunk by chunk: whatever has been written +// must be decodable by the client before the stream is closed, otherwise a +// subscription-style response would stall until it ended. +func TestWriterFlushDeliversDecodableBytes(t *testing.T) { + plain := []byte(`{"chunk":"first"}`) + + for _, scheme := range []compression.Scheme{compression.Gzip, compression.Zstd} { + t.Run(string(scheme), func(te *testing.T) { + var buf bytes.Buffer + writer, err := compression.AcquireWriter(scheme, &buf) + require.NoError(te, err) + defer compression.ReleaseWriter(writer) + + _, err = writer.Write(plain) + require.NoError(te, err) + require.NoError(te, writer.Flush()) + + reader, err := compression.WrapReader(string(scheme), bytes.NewReader(buf.Bytes())) + require.NoError(te, err) + defer func() { _ = reader.Close() }() + got := make([]byte, len(plain)) + _, err = io.ReadFull(reader, got) + + require.NoError(te, err) + assert.Equal(te, plain, got) + }) + } +} diff --git a/internal/server/http_server/compress.go b/internal/server/http_server/compress.go index 806c842e..3a172b1a 100644 --- a/internal/server/http_server/compress.go +++ b/internal/server/http_server/compress.go @@ -1,154 +1,137 @@ package http_server -// the package is copied from echo's compress middleware -// with replacement of gzip library +// the package is adapted from echo's compress middleware // https://github.com/labstack/echo/blob/master/middleware/compress.go +// with the hard-wired gzip codec replaced by internal/compression, which +// negotiates zstd as well, and without the MinLength buffering echo grew for +// its threshold option. import ( "bufio" "io" "net" "net/http" - "strings" - "sync" - "github.com/klauspost/compress/gzip" + "github.com/drpcorg/nodecore/internal/compression" "github.com/labstack/echo/v4" - emiddleware "github.com/labstack/echo/v4/middleware" "github.com/rs/zerolog/log" ) -type ( - // GzipConfig defines the config for Gzip middleware. - GzipConfig struct { - // Skipper defines a function to skip middleware. - Skipper emiddleware.Skipper - - // Gzip compression level. - // Optional. Default value -1. - Level int `yaml:"level"` - } - - gzipResponseWriter struct { - io.Writer - http.ResponseWriter - wroteBody bool - } -) - -const ( - gzipScheme = "gzip" -) - -// DefaultGzipConfig is the default Gzip middleware config. -var DefaultGzipConfig = GzipConfig{ - Skipper: emiddleware.DefaultSkipper, - Level: -1, -} - -// Gzip returns a middleware which compresses HTTP response using gzip compression -// scheme. -func Gzip() echo.MiddlewareFunc { - return GzipWithConfig(DefaultGzipConfig) +// compressResponseWriter encodes the response body with the coding the client +// negotiated. The status line is held back until the first byte of body: +// headers freeze once the status goes out, and until then we do not know +// whether this response has a body to label with a Content-Encoding. +type compressResponseWriter struct { + http.ResponseWriter + writer compression.Writer + scheme compression.Scheme + code int + wroteHeader bool + committed bool } -// GzipWithConfig return Gzip middleware with config. -// See: `Gzip()`. -func GzipWithConfig(config GzipConfig) echo.MiddlewareFunc { - // Defaults - if config.Skipper == nil { - config.Skipper = DefaultGzipConfig.Skipper - } - if config.Level == 0 { - config.Level = DefaultGzipConfig.Level - } - - pool := gzipCompressPool(config) - +// Compress returns a middleware that compresses the response body with the +// coding the client asked for - zstd or gzip, whichever Negotiate picks. +// A client that asks for neither is served plain bytes. +func Compress() echo.MiddlewareFunc { return func(next echo.HandlerFunc) echo.HandlerFunc { return func(c echo.Context) error { - if config.Skipper(c) { + res := c.Response() + // Announced even when nothing is compressed: a cache that skips + // this key hands a zstd body to a gzip-only client. + res.Header().Add(echo.HeaderVary, echo.HeaderAcceptEncoding) + + scheme := compression.Negotiate(c.Request().Header.Get(echo.HeaderAcceptEncoding)) + if scheme == compression.Identity { return next(c) } - res := c.Response() - res.Header().Add(echo.HeaderVary, echo.HeaderAcceptEncoding) - if strings.Contains(c.Request().Header.Get(echo.HeaderAcceptEncoding), gzipScheme) { - res.Header().Set(echo.HeaderContentEncoding, gzipScheme) // Issue #806 - i := pool.Get() - w, ok := i.(*gzip.Writer) - if !ok { - return echo.NewHTTPError(http.StatusInternalServerError, i.(error).Error()) - } - rw := res.Writer - w.Reset(rw) - grw := &gzipResponseWriter{Writer: w, ResponseWriter: rw} - defer func() { - if !grw.wroteBody { - if res.Header().Get(echo.HeaderContentEncoding) == gzipScheme { - res.Header().Del(echo.HeaderContentEncoding) - } - // We have to reset response to it's pristine state when - // nothing is written to body or error is returned. - // See issue #424, #407. - res.Writer = rw - w.Reset(io.Discard) - } - err := w.Close() - if err != nil { - log.Error().Err(err).Msg("couldn't close writer") - } - pool.Put(w) - }() - res.Writer = grw + rw := res.Writer + writer, err := compression.AcquireWriter(scheme, rw) + if err != nil { + // An unusable codec pool is an operator problem, not a reason + // to fail the request: serving the body uncompressed is + // something every client understands. + log.Error().Err(err).Str("scheme", string(scheme)).Msg("couldn't acquire a compressing writer") + return next(c) } + + crw := &compressResponseWriter{ResponseWriter: rw, writer: writer, scheme: scheme} + defer func() { + if !crw.committed { + // Nothing was ever written, so no Content-Encoding went + // out and the codec must not append an empty frame to the + // body. The status still has to reach the client. + if crw.wroteHeader { + rw.WriteHeader(crw.code) + } + res.Writer = rw + writer.Reset(io.Discard) + } + if closeErr := writer.Close(); closeErr != nil { + log.Error().Err(closeErr).Msg("couldn't close a compressing writer") + } + compression.ReleaseWriter(writer) + }() + res.Writer = crw + return next(c) } } } -func (w *gzipResponseWriter) WriteHeader(code int) { +func (w *compressResponseWriter) WriteHeader(code int) { w.Header().Del(echo.HeaderContentLength) // Issue #444 - w.ResponseWriter.WriteHeader(code) + w.wroteHeader = true + w.code = code } -func (w *gzipResponseWriter) Write(b []byte) (int, error) { +// commit labels the response with its coding and releases the held status. +// It runs exactly once, on whichever comes first of the first body byte and +// an explicit flush. +func (w *compressResponseWriter) commit() { + if w.committed { + return + } + w.committed = true + w.Header().Set(echo.HeaderContentEncoding, string(w.scheme)) // Issue #806 + if w.wroteHeader { + w.ResponseWriter.WriteHeader(w.code) + } +} + +func (w *compressResponseWriter) Write(b []byte) (int, error) { if w.Header().Get(echo.HeaderContentType) == "" { w.Header().Set(echo.HeaderContentType, http.DetectContentType(b)) } - w.wroteBody = true - return w.Writer.Write(b) + w.commit() + return w.writer.Write(b) } -func (w *gzipResponseWriter) Flush() { - err := w.Writer.(*gzip.Writer).Flush() - if err != nil { - log.Error().Err(err).Msg("couldn't flush") +// Flush pushes a streamed chunk all the way to the socket: through the codec +// first, since bytes still buffered in an encoder have not been produced yet. +func (w *compressResponseWriter) Flush() { + w.commit() + if err := w.writer.Flush(); err != nil { + log.Error().Err(err).Msg("couldn't flush a compressing writer") } if flusher, ok := w.ResponseWriter.(http.Flusher); ok { flusher.Flush() } } -func (w *gzipResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { +func (w *compressResponseWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) { return w.ResponseWriter.(http.Hijacker).Hijack() } -func (w *gzipResponseWriter) Push(target string, opts *http.PushOptions) error { +func (w *compressResponseWriter) Push(target string, opts *http.PushOptions) error { if p, ok := w.ResponseWriter.(http.Pusher); ok { return p.Push(target, opts) } return http.ErrNotSupported } -func gzipCompressPool(config GzipConfig) sync.Pool { - return sync.Pool{ - New: func() interface{} { - w, err := gzip.NewWriterLevel(io.Discard, config.Level) - if err != nil { - return err - } - return w - }, - } +// Unwrap exposes the underlying writer to http.ResponseController. +func (w *compressResponseWriter) Unwrap() http.ResponseWriter { + return w.ResponseWriter } diff --git a/internal/server/http_server/compress_test.go b/internal/server/http_server/compress_test.go new file mode 100644 index 00000000..5c42d66f --- /dev/null +++ b/internal/server/http_server/compress_test.go @@ -0,0 +1,96 @@ +package http_server_test + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/drpcorg/nodecore/internal/compression" + "github.com/drpcorg/nodecore/internal/server/http_server" + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var compressBody = []byte(`{"jsonrpc":"2.0","id":1,"result":"0x1010101010101010101010"}`) + +func serveCompressed(t *testing.T, acceptEncoding string) *httptest.ResponseRecorder { + t.Helper() + e := echo.New() + e.Use(http_server.Compress()) + e.GET("/", func(c echo.Context) error { + return c.Blob(http.StatusOK, echo.MIMEApplicationJSON, compressBody) + }) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + if acceptEncoding != "" { + req.Header.Set(echo.HeaderAcceptEncoding, acceptEncoding) + } + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + return rec +} + +func TestCompressServesTheNegotiatedCoding(t *testing.T) { + tests := []struct { + name string + acceptEncoding string + contentEncoding string + }{ + {"zstd client", "zstd", "zstd"}, + {"gzip client", "gzip", "gzip"}, + {"client offering both prefers zstd", "gzip, zstd", "zstd"}, + {"client refusing zstd still gets gzip", "zstd;q=0, gzip", "gzip"}, + {"unknown coding is not compressed", "br", ""}, + {"no header is not compressed", "", ""}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + rec := serveCompressed(te, tt.acceptEncoding) + + require.Equal(te, http.StatusOK, rec.Code) + assert.Equal(te, tt.contentEncoding, rec.Header().Get(echo.HeaderContentEncoding)) + + reader, err := compression.WrapReader(tt.contentEncoding, bytes.NewReader(rec.Body.Bytes())) + require.NoError(te, err) + defer func() { require.NoError(te, reader.Close()) }() + got, err := io.ReadAll(reader) + + require.NoError(te, err) + assert.Equal(te, compressBody, got, "the client must be able to decode what it asked for") + }) + } +} + +// Caches key on Accept-Encoding or they hand a zstd body to a gzip-only +// client, so the header is announced whether or not this response was +// compressed. +func TestCompressAlwaysVariesOnAcceptEncoding(t *testing.T) { + for _, acceptEncoding := range []string{"", "gzip", "zstd"} { + rec := serveCompressed(t, acceptEncoding) + + assert.Contains(t, rec.Header().Values(echo.HeaderVary), echo.HeaderAcceptEncoding) + } +} + +// A handler that writes no body must not leave a Content-Encoding behind, +// or the client tries to decode zero bytes as a compressed frame. +func TestCompressLeavesEmptyResponsesUnencoded(t *testing.T) { + e := echo.New() + e.Use(http_server.Compress()) + e.GET("/", func(c echo.Context) error { + return c.NoContent(http.StatusNoContent) + }) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.Header.Set(echo.HeaderAcceptEncoding, "zstd") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusNoContent, rec.Code) + assert.Empty(t, rec.Header().Get(echo.HeaderContentEncoding)) + assert.Empty(t, rec.Body.Bytes()) +} diff --git a/internal/server/http_server/cors_vary_test.go b/internal/server/http_server/cors_vary_test.go new file mode 100644 index 00000000..9381c163 --- /dev/null +++ b/internal/server/http_server/cors_vary_test.go @@ -0,0 +1,44 @@ +package http_server + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/assert" +) + +// corsContext builds a context whose response already carries the +// Vary: Accept-Encoding the compression middleware adds on the way in. +func corsContext(origin string) echo.Context { + req := httptest.NewRequest(http.MethodPost, "/", nil) + req.Header.Set("Origin", origin) + c := echo.New().NewContext(req, httptest.NewRecorder()) + c.Response().Header().Add(echo.HeaderVary, echo.HeaderAcceptEncoding) + return c +} + +// Two codings are now negotiable, so a shared cache that stops keying on +// Accept-Encoding will eventually hand a zstd body to a gzip-only client. +// Announcing Origin must therefore add to Vary, not replace what is there. +func TestSetCorsHeadersKeepsTheAcceptEncodingVary(t *testing.T) { + tests := []struct { + name string + corsOrigins []string + }{ + {"configured origins", []string{"http://localhost:123"}}, + {"wildcard origins", nil}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + c := corsContext("http://localhost:123") + + setCorsHeaders(c, tt.corsOrigins) + + assert.Contains(te, c.Response().Header().Values(echo.HeaderVary), echo.HeaderAcceptEncoding, + "the coding this response was compressed with must stay part of the cache key") + }) + } +} diff --git a/internal/server/http_server/decompress.go b/internal/server/http_server/decompress.go new file mode 100644 index 00000000..8fec5b10 --- /dev/null +++ b/internal/server/http_server/decompress.go @@ -0,0 +1,52 @@ +package http_server + +// the package is adapted from echo's decompress middleware +// https://github.com/labstack/echo/blob/master/middleware/decompress.go +// which only ever understood gzip. + +import ( + "errors" + "io" + "net/http" + "strings" + + "github.com/drpcorg/nodecore/internal/compression" + "github.com/labstack/echo/v4" + "github.com/rs/zerolog" +) + +// Decompress returns a middleware that decodes a compressed request body, so +// handlers always read plain bytes whatever the client sent. Codings nodecore +// does not speak are passed through untouched rather than guessed at. +func Decompress() echo.MiddlewareFunc { + return func(next echo.HandlerFunc) echo.HandlerFunc { + return func(c echo.Context) error { + req := c.Request() + encoding := strings.TrimSpace(req.Header.Get(echo.HeaderContentEncoding)) + if encoding == "" || strings.EqualFold(encoding, "identity") { + return next(c) + } + + reader, err := compression.WrapReader(encoding, req.Body) + if errors.Is(err, compression.ErrUnsupportedEncoding) { + return next(c) + } + if err != nil { + zerolog.Ctx(req.Context()).Debug().Err(err).Msg("client sent an undecodable request body") + return echo.NewHTTPError(http.StatusBadRequest, "invalid compressed request body") + } + // Releasing the codec here rather than through req.Body keeps it + // out of reach of the server's own Close of the original body, + // which owns the connection and must stay the one to close it. + defer func() { _ = reader.Close() }() + req.Body = io.NopCloser(reader) + + // The header described the bytes that arrived, not the ones the + // handler now reads. Left in place it would be forwarded to an + // upstream and tell a node to decompress a plain body. + req.Header.Del(echo.HeaderContentEncoding) + + return next(c) + } + } +} diff --git a/internal/server/http_server/decompress_test.go b/internal/server/http_server/decompress_test.go new file mode 100644 index 00000000..e770cc8b --- /dev/null +++ b/internal/server/http_server/decompress_test.go @@ -0,0 +1,156 @@ +package http_server_test + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/drpcorg/nodecore/internal/compression" + "github.com/drpcorg/nodecore/internal/server/http_server" + "github.com/labstack/echo/v4" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// postCompressed sends body under contentEncoding and reports what the +// handler behind the middleware actually received. +func postCompressed(t *testing.T, contentEncoding string, body []byte) (*httptest.ResponseRecorder, []byte) { + t.Helper() + var seen []byte + e := echo.New() + e.Use(http_server.Decompress()) + e.POST("/", func(c echo.Context) error { + read, err := io.ReadAll(c.Request().Body) + if err != nil { + return err + } + seen = read + return c.NoContent(http.StatusOK) + }) + + req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(body)) + if contentEncoding != "" { + req.Header.Set(echo.HeaderContentEncoding, contentEncoding) + } + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + return rec, seen +} + +func compress(t *testing.T, scheme compression.Scheme, plain []byte) []byte { + t.Helper() + var buf bytes.Buffer + writer, err := compression.AcquireWriter(scheme, &buf) + require.NoError(t, err) + defer compression.ReleaseWriter(writer) + _, err = writer.Write(plain) + require.NoError(t, err) + require.NoError(t, writer.Close()) + return buf.Bytes() +} + +func TestDecompressDecodesRequestBodies(t *testing.T) { + plain := []byte(`{"jsonrpc":"2.0","method":"eth_blockNumber","id":1}`) + tests := []struct { + name string + contentEncoding string + body []byte + }{ + {"zstd", "zstd", compress(t, compression.Zstd, plain)}, + {"gzip", "gzip", compress(t, compression.Gzip, plain)}, + {"case-insensitive", "ZSTD", compress(t, compression.Zstd, plain)}, + {"no encoding", "", plain}, + {"identity", "identity", plain}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + rec, seen := postCompressed(te, tt.contentEncoding, tt.body) + + require.Equal(te, http.StatusOK, rec.Code) + assert.Equal(te, plain, seen) + }) + } +} + +// Once the body has been decoded the header no longer describes it. It has to +// go, or it rides along to the upstream and tells a node to decompress bytes +// that are already plain. +func TestDecompressDropsTheContentEncodingHeader(t *testing.T) { + plain := []byte(`{"jsonrpc":"2.0","method":"eth_blockNumber","id":1}`) + var seen string + e := echo.New() + e.Use(http_server.Decompress()) + e.POST("/", func(c echo.Context) error { + seen = c.Request().Header.Get(echo.HeaderContentEncoding) + return c.NoContent(http.StatusOK) + }) + + req := httptest.NewRequest(http.MethodPost, "/", bytes.NewReader(compress(t, compression.Zstd, plain))) + req.Header.Set(echo.HeaderContentEncoding, "zstd") + e.ServeHTTP(httptest.NewRecorder(), req) + + assert.Empty(t, seen) +} + +// A coding nodecore does not decode is left alone rather than guessed at: the +// handler sees exactly the bytes the client sent, as it always has. +func TestDecompressPassesUnknownCodingsThrough(t *testing.T) { + body := []byte("\x1b\x2f\x00 brotli-ish bytes") + + rec, seen := postCompressed(t, "br", body) + + require.Equal(t, http.StatusOK, rec.Code) + assert.Equal(t, body, seen) +} + +// A client that declares a coding its body is not in gets a clean 400. This +// is a realistic client bug - a library that sets the header but forgets the +// encoder - and it must not read as a server fault. +func TestDecompressRejectsBodiesThatAreNotTheDeclaredCoding(t *testing.T) { + plain := []byte(`{"jsonrpc":"2.0","method":"eth_blockNumber","id":1}`) + tests := []struct { + name string + contentEncoding string + body []byte + }{ + {"gzip declared, plain body", "gzip", plain}, + {"zstd declared, plain body", "zstd", plain}, + {"gzip declared, header truncated", "gzip", compress(t, compression.Gzip, plain)[:5]}, + {"zstd declared, magic truncated", "zstd", compress(t, compression.Zstd, plain)[:2]}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + rec, _ := postCompressed(te, tt.contentEncoding, tt.body) + + assert.Equal(te, http.StatusBadRequest, rec.Code) + }) + } +} + +// A body that starts as valid but stops early cannot be caught before the +// handler reads it. What matters is that the request fails instead of being +// served as though the client had sent a short body. +func TestDecompressFailsOnTruncatedStream(t *testing.T) { + plain := bytes.Repeat([]byte("x"), 4096) + tests := []struct { + name string + contentEncoding string + body []byte + }{ + {"zstd", "zstd", compress(t, compression.Zstd, plain)[:16]}, + {"gzip", "gzip", compress(t, compression.Gzip, plain)[:14]}, + } + + for _, tt := range tests { + t.Run(tt.name, func(te *testing.T) { + rec, seen := postCompressed(te, tt.contentEncoding, tt.body) + + assert.NotEqual(te, http.StatusOK, rec.Code) + assert.NotEqual(te, plain, seen) + }) + } +} diff --git a/internal/server/http_server/http_server.go b/internal/server/http_server/http_server.go index 7f668942..648a3101 100644 --- a/internal/server/http_server/http_server.go +++ b/internal/server/http_server/http_server.go @@ -18,9 +18,7 @@ import ( "github.com/drpcorg/nodecore/internal/protocol" "github.com/drpcorg/nodecore/internal/quorum" "github.com/drpcorg/nodecore/pkg/utils" - "github.com/klauspost/compress/gzip" "github.com/labstack/echo/v4" - "github.com/labstack/echo/v4/middleware" "github.com/prometheus/client_golang/prometheus" "github.com/rs/zerolog/log" "github.com/samber/lo" @@ -73,8 +71,8 @@ func NewHttpServer(ctx context.Context, appCtx *server_ctx.ApplicationServerCont configureServer(ctx, httpServer.Server) configureServer(ctx, httpServer.TLSServer) httpServer.JSONSerializer = &FastJSONSerializer{} - httpServer.Use(middleware.Decompress()) - httpServer.Use(GzipWithConfig(GzipConfig{Level: gzip.BestSpeed})) + httpServer.Use(Decompress()) + httpServer.Use(Compress()) httpGroup := httpServer.Group("/queries/:chain") @@ -282,7 +280,10 @@ func setCorsHeaders(reqCtx echo.Context, corsOrigins []string) { for _, item := range corsOrigins { if utils.MatchWildcards(item, origin) { reqCtx.Response().Header().Set("Access-Control-Allow-Origin", origin) - reqCtx.Response().Header().Set("Vary", "Origin") + // Added, not set: the compression middleware has already put + // Accept-Encoding here, and dropping it lets a shared cache + // hand a zstd body to a client that only reads gzip. + reqCtx.Response().Header().Add("Vary", "Origin") return } } diff --git a/internal/upstreams/connectors/http_connector.go b/internal/upstreams/connectors/http_connector.go index 1e110796..8c1f410d 100644 --- a/internal/upstreams/connectors/http_connector.go +++ b/internal/upstreams/connectors/http_connector.go @@ -16,6 +16,7 @@ import ( "github.com/bytedance/sonic" mapset "github.com/deckarep/golang-set/v2" + "github.com/drpcorg/nodecore/internal/compression" "github.com/drpcorg/nodecore/internal/config" "github.com/drpcorg/nodecore/internal/protocol" "github.com/drpcorg/nodecore/internal/quorum" @@ -58,10 +59,14 @@ var defaultResponseHeaderDeny = []string{ // defaultRequestHeaderDeny is the request-side mirror of the response deny // list: RFC 7230 §6.1 hop-by-hop headers plus Host and Content-Length (both // owned by the transport for the *outgoing* request) and Accept-Encoding. -// Accept-Encoding must stay with the transport: forwarding the client's value -// disables Go's transparent decompression, so a gzip upstream body would ride -// through unmarked (Content-Encoding is stripped from responses) and get -// compressed a second time by the server gzip middleware (issue #268). +// +// Accept-Encoding belongs to this hop alone. The two hops compress +// independently - the connector decodes whatever the node sends and the +// server re-encodes for the client - so a client's preference says nothing +// about what this connector should ask a node for. Forwarding it is also how +// issue #268 happened: the upstream's compressed body lost its +// Content-Encoding to the response deny list and was then compressed a second +// time on the way out. var defaultRequestHeaderDeny = mapset.NewThreadUnsafeSet( "Connection", "Keep-Alive", @@ -283,6 +288,44 @@ func (h *HttpConnector) applyConfigHeaders(req *http.Request) { for k, v := range h.additionalHeaders { req.Header.Set(k, v) } + // Go's transport would negotiate gzip on its own, but only gzip, and only + // while no Accept-Encoding is set. Asking for zstd here therefore also + // takes over decoding the answer - see decodeResponseBody. An operator who + // pinned the header in the connector config keeps it: a node that + // mishandles a coding is exactly what that setting is for. + if req.Header.Get(acceptEncodingHeader) == "" { + req.Header.Set(acceptEncodingHeader, compression.Offer) + } +} + +const acceptEncodingHeader = "Accept-Encoding" + +// decodeResponseBody wraps the response body in the decoder its +// Content-Encoding calls for. The returned reader owns both the codec and the +// body: closing it releases the pooled decoder and then the connection. +func decodeResponseBody(resp *http.Response) (io.ReadCloser, error) { + decoded, err := compression.WrapReader(resp.Header.Get("Content-Encoding"), resp.Body) + if err != nil { + return nil, err + } + return &decodedBody{Reader: decoded, decoder: decoded, raw: resp.Body}, nil +} + +// decodedBody ties the lifetime of a pooled decoder to the response body it +// decodes, so neither the buffered nor the streaming path has to remember +// there are two things to close. +type decodedBody struct { + io.Reader + decoder io.Closer + raw io.Closer +} + +func (d *decodedBody) Close() error { + err := d.decoder.Close() + if rawErr := d.raw.Close(); err == nil { + err = rawErr + } + return err } // applyClientHeaders forwards per-request client headers onto the upstream @@ -447,20 +490,34 @@ func (h *HttpConnector) dispatch( ) } + // Decoding happens before anything reads the body, so the buffered and + // streaming paths downstream only ever see plain bytes - which is what + // they must be, since Content-Encoding is stripped from the headers this + // response carries onward. + body, err := decodeResponseBody(resp) + if err != nil { + utils.CloseBodyReader(ctx, resp.Body) + zerolog.Ctx(ctx).Warn().Err(err).Str("upstream", h.upstreamId).Msg("cannot decode the upstream response body") + return protocol.NewPartialFailure( + request, + protocol.ServerErrorWithCause(fmt.Errorf("cannot decode the response from upstream %s", h.upstreamId)), + ) + } + if request.IsStream() && isSuccessStatus(resp.StatusCode) && !quorumRequested { - bufReader := bufio.NewReaderSize(resp.Body, protocol.MaxChunkSize) + bufReader := bufio.NewReaderSize(body, protocol.MaxChunkSize) if decision := allowStream(bufReader); decision.stream { zerolog.Ctx(ctx).Debug().Msgf("streaming response of method %s", request.Method()) - streamResp := protocol.NewHttpUpstreamResponseStream(request.Id(), protocol.NewCloseReader(ctx, bufReader, resp.Body), request.RequestType()). + streamResp := protocol.NewHttpUpstreamResponseStream(request.Id(), protocol.NewCloseReader(ctx, bufReader, body), request.RequestType()). WithStreamHint(decision.hint) return streamResp.WithResponseHeaders(h.filterResponseHeaders(resp.Header)) } - defer utils.CloseBodyReader(ctx, resp.Body) + defer utils.CloseBodyReader(ctx, body) return h.receiveWholeResponse(ctx, request, resp.StatusCode, resp.Header, bufReader) } - defer utils.CloseBodyReader(ctx, resp.Body) - return h.receiveWholeResponse(ctx, request, resp.StatusCode, resp.Header, resp.Body) + defer utils.CloseBodyReader(ctx, body) + return h.receiveWholeResponse(ctx, request, resp.StatusCode, resp.Header, body) } func (h *HttpConnector) receiveWholeResponse( diff --git a/internal/upstreams/connectors/http_connector_compression_test.go b/internal/upstreams/connectors/http_connector_compression_test.go new file mode 100644 index 00000000..713e79f5 --- /dev/null +++ b/internal/upstreams/connectors/http_connector_compression_test.go @@ -0,0 +1,159 @@ +package connectors_test + +import ( + "bytes" + "context" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/drpcorg/nodecore/internal/config" + "github.com/drpcorg/nodecore/internal/protocol" + "github.com/drpcorg/nodecore/internal/upstreams/connectors" + "github.com/drpcorg/nodecore/pkg/methods" + "github.com/klauspost/compress/gzip" + "github.com/klauspost/compress/zstd" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var upstreamBody = []byte(`{"jsonrpc":"2.0","id":1,"result":{"number":"0x1337"}}`) + +func encodeUpstream(t *testing.T, scheme string, plain []byte) []byte { + t.Helper() + var buf bytes.Buffer + switch scheme { + case "gzip": + w := gzip.NewWriter(&buf) + _, err := w.Write(plain) + require.NoError(t, err) + require.NoError(t, w.Close()) + case "zstd": + w, err := zstd.NewWriter(&buf) + require.NoError(t, err) + _, err = w.Write(plain) + require.NoError(t, err) + require.NoError(t, w.Close()) + default: + return plain + } + return buf.Bytes() +} + +// upstreamServing answers every request with plain encoded as scheme, and +// records the Accept-Encoding the connector offered. +func upstreamServing(t *testing.T, scheme string, plain []byte) (*httptest.Server, *string) { + t.Helper() + offered := new(string) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + *offered = r.Header.Get("Accept-Encoding") + if scheme != "" { + w.Header().Set("Content-Encoding", scheme) + } + _, _ = w.Write(encodeUpstream(t, scheme, plain)) + })) + t.Cleanup(srv.Close) + return srv, offered +} + +func restConnectorFor(t *testing.T, cfg *config.ApiConnectorConfig) *connectors.HttpConnector { + t.Helper() + connector, err := connectors.NewHttpConnector(cfg, specs.RestConnector, "", "test-upstream") + require.NoError(t, err) + return connector +} + +// Go's transport only ever negotiates gzip on its own, so zstd has to be +// offered explicitly - which also hands nodecore the job of decoding both. +func TestUpstreamRequestOffersZstdAndGzip(t *testing.T) { + srv, offered := upstreamServing(t, "", upstreamBody) + connector := restConnectorFor(t, &config.ApiConnectorConfig{Url: srv.URL}) + + r := connector.SendRequest(context.Background(), protocol.NewUpstreamRestRequest("1", "GET#/status", nil, nil, "")) + + require.False(t, r.HasError()) + assert.Equal(t, "zstd, gzip", *offered) +} + +// Whatever coding the node answers with, the framework above the connector +// must see plain JSON: the connector strips Content-Encoding, so compressed +// bytes leaving here would reach the client unlabelled and unreadable. +func TestUpstreamResponseIsDecoded(t *testing.T) { + for _, scheme := range []string{"zstd", "gzip", ""} { + name := scheme + if name == "" { + name = "identity" + } + t.Run(name, func(te *testing.T) { + srv, _ := upstreamServing(te, scheme, upstreamBody) + connector := restConnectorFor(te, &config.ApiConnectorConfig{Url: srv.URL}) + + r := connector.SendRequest(context.Background(), protocol.NewUpstreamRestRequest("1", "GET#/status", nil, nil, "")) + + require.False(te, r.HasError()) + assert.Equal(te, upstreamBody, r.ResponseResult()) + carrier, ok := r.(protocol.HasResponseHeaders) + require.True(te, ok) + assert.Empty(te, carrier.ResponseHeaders().Get("Content-Encoding"), + "the body is plain now, so nothing may claim otherwise") + }) + } +} + +// The streaming path never buffers the body, so it needs the decoder wired +// into the stream itself rather than around a finished response. +func TestUpstreamStreamedResponseIsDecoded(t *testing.T) { + for _, scheme := range []string{"zstd", "gzip"} { + t.Run(scheme, func(te *testing.T) { + plain := bytes.Repeat([]byte(`{"chunk":"0123456789"}`), 512) + srv, _ := upstreamServing(te, scheme, plain) + connector := restConnectorFor(te, &config.ApiConnectorConfig{Url: srv.URL}) + + r := connector.SendRequest( + context.Background(), + protocol.NewStreamUpstreamRestRequest("1", "GET#/status", nil, nil, ""), + ) + + require.False(te, r.HasError()) + require.True(te, r.HasStream()) + got, err := io.ReadAll(r.EncodeResponse([]byte("1"))) + require.NoError(te, err) + assert.Equal(te, plain, got) + }) + } +} + +// An operator who pins Accept-Encoding on the connector has a reason - a node +// that mishandles one of the codings, most likely - and the connector must +// not talk over them. +func TestConfiguredAcceptEncodingIsNotOverridden(t *testing.T) { + srv, offered := upstreamServing(t, "gzip", upstreamBody) + connector := restConnectorFor(t, &config.ApiConnectorConfig{ + Url: srv.URL, + Headers: map[string]string{"Accept-Encoding": "gzip"}, + }) + + r := connector.SendRequest(context.Background(), protocol.NewUpstreamRestRequest("1", "GET#/status", nil, nil, "")) + + require.False(t, r.HasError()) + assert.Equal(t, "gzip", *offered) + assert.Equal(t, upstreamBody, r.ResponseResult(), + "a pinned coding must still be decoded") +} + +// A node answering with a coding nodecore never offered has broken the +// negotiation. Failing is the only honest outcome: the bytes are undecodable +// here and would be unreadable at the client. +func TestUnsupportedUpstreamCodingFails(t *testing.T) { + srv, _ := upstreamServing(t, "", upstreamBody) + srv.Config.Handler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Encoding", "br") + _, _ = w.Write(upstreamBody) + }) + connector := restConnectorFor(t, &config.ApiConnectorConfig{Url: srv.URL}) + + r := connector.SendRequest(context.Background(), protocol.NewUpstreamRestRequest("1", "GET#/status", nil, nil, "")) + + assert.True(t, r.HasError(), "an undecodable body must not be passed off as a result") +} diff --git a/internal/upstreams/connectors/http_connector_test.go b/internal/upstreams/connectors/http_connector_test.go index 08f2b533..200df7c3 100644 --- a/internal/upstreams/connectors/http_connector_test.go +++ b/internal/upstreams/connectors/http_connector_test.go @@ -11,6 +11,7 @@ import ( "strings" "testing" + "github.com/drpcorg/nodecore/internal/compression" "github.com/drpcorg/nodecore/internal/config" "github.com/drpcorg/nodecore/internal/protocol" "github.com/drpcorg/nodecore/internal/upstreams/connectors" @@ -532,11 +533,11 @@ func TestRestRequest_ConfigHeadersWinAcrossCasing(t *testing.T) { } // Hop-by-hop headers (RFC 7230 §6.1) and Accept-Encoding must not be -// forwarded from the client to the upstream. Forwarding Accept-Encoding is -// how double-gzip happens (issue #268): an explicit Accept-Encoding on the -// outgoing request disables Go's transparent decompression, the compressed -// body then loses its Content-Encoding in the response deny list, and the -// server-side gzip middleware compresses it a second time. +// forwarded from the client to the upstream. The two hops compress +// independently, so the connector answers with its own offer instead - +// letting the client's value through is how double-gzip happened (issue +// #268): the upstream's compressed body lost its Content-Encoding in the +// response deny list and was compressed a second time on the way out. func TestRestRequest_HopByHopClientHeadersNotForwarded(t *testing.T) { httpmock.Activate(t) defer httpmock.Deactivate() @@ -574,12 +575,14 @@ func TestRestRequest_HopByHopClientHeadersNotForwarded(t *testing.T) { require.False(t, r.HasError()) for _, denied := range []string{ - "Accept-Encoding", "Connection", "Keep-Alive", "Proxy-Authorization", + "Connection", "Keep-Alive", "Proxy-Authorization", "Te", "Trailer", "Transfer-Encoding", "Upgrade", "Host", "Content-Length", } { assert.Empty(t, gotHeaders.Values(denied), "client %s must not be forwarded to the upstream", denied) } + assert.Equal(t, []string{compression.Offer}, gotHeaders.Values("Accept-Encoding"), + "the client's Accept-Encoding must be replaced by the connector's own offer, not appended to") assert.Equal(t, []string{"hello"}, gotHeaders.Values("X-Custom"), "non-denied client headers must still pass through") }