feat: add zstd support for http ingress and upstream connectors

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
+}

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))
+		})
+	}
+}

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
+}