From 7b10d79f0c1a787866a35c929a75a81f29f66500 Mon Sep 17 00:00:00 2001 From: "David Muto (pseudomuto)" Date: Thu, 2 Jul 2026 08:41:51 -0400 Subject: [PATCH] [compress]: Add payload compression algorithms The payload compression RFC (#50) settles on LZ4, ZSTD, and Snappy as the algorithms the proxy will support. This adds the package implementing them so later work (metadata tagging, config, wiring into the proxy) has a concrete compressor to build on. Each algorithm is a small type (LZ4, Zstd, Snappy) exposing the same shape: an Encoding method returning the content-encoding identifier that will drive decompression, plus Compress/Decompress over whole in-memory buffers. All treat their input as read-only and return freshly allocated output, which keeps them safe for concurrent use. --- go.mod | 3 + go.sum | 6 ++ pkg/compress/compressor_test.go | 147 ++++++++++++++++++++++++++++++++ pkg/compress/doc.go | 8 ++ pkg/compress/lz4.go | 50 +++++++++++ pkg/compress/lz4_test.go | 21 +++++ pkg/compress/snappy.go | 38 +++++++++ pkg/compress/snappy_test.go | 21 +++++ pkg/compress/zstd.go | 57 +++++++++++++ pkg/compress/zstd_test.go | 49 +++++++++++ 10 files changed, 400 insertions(+) create mode 100644 pkg/compress/compressor_test.go create mode 100644 pkg/compress/doc.go create mode 100644 pkg/compress/lz4.go create mode 100644 pkg/compress/lz4_test.go create mode 100644 pkg/compress/snappy.go create mode 100644 pkg/compress/snappy_test.go create mode 100644 pkg/compress/zstd.go create mode 100644 pkg/compress/zstd_test.go diff --git a/go.mod b/go.mod index 2a5ea51..9304487 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,9 @@ go 1.26.4 require ( github.com/goccy/go-yaml v1.19.2 + github.com/golang/snappy v1.0.0 + github.com/klauspost/compress v1.18.0 + github.com/pierrec/lz4/v4 v4.1.27 github.com/prometheus/client_golang v1.23.2 github.com/rs/zerolog v1.35.1 github.com/stretchr/testify v1.11.1 diff --git a/go.sum b/go.sum index 4f24f64..7bf483c 100644 --- a/go.sum +++ b/go.sum @@ -12,12 +12,16 @@ github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM= github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/golang/snappy v1.0.0 h1:Oy607GVXHs7RtbggtPBnr2RmDArIsAefDwvrdWvRhGs= +github.com/golang/snappy v1.0.0/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1 h1:X5VWvz21y3gzm9Nw/kaUeku/1+uBhcekkmy4IkffJww= github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.1/go.mod h1:Zanoh4+gvIgluNqcfMVTJueD4wSS5hT7zTt4Mrutd90= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= @@ -30,6 +34,8 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/nexus-rpc/nexus-proto-annotations v0.1.0 h1:2fELd+9sqUtNu6Fg//pw8YFsxOvp8vZ8hfP0nHhNI80= github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHvbxJ+27yPWrhpo8w/Yg1IOuY0Y= +github.com/pierrec/lz4/v4 v4.1.27 h1:+PhzhWDrjRj89TH2sw43nE3+4+W8lSxIuQadEHZyjUk= +github.com/pierrec/lz4/v4 v4.1.27/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= diff --git a/pkg/compress/compressor_test.go b/pkg/compress/compressor_test.go new file mode 100644 index 0000000..6232ca1 --- /dev/null +++ b/pkg/compress/compressor_test.go @@ -0,0 +1,147 @@ +package compress_test + +import ( + "bytes" + "sync" + "testing" + + "github.com/stretchr/testify/require" +) + +type compressor interface { + Compress([]byte) ([]byte, error) + Decompress([]byte) ([]byte, error) +} + +func runCompressorSuite(t *testing.T, c compressor) { + t.Helper() + + t.Run("round trip", func(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + data []byte + }{ + {"empty", []byte{}}, + {"ascii", []byte("hello, world")}, + {"binary", []byte{0x00, 0xFF, 0x42, 0x13}}, + {"large", bytes.Repeat([]byte("a"), 64*1024)}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + compressed, err := c.Compress(tc.data) + require.NoError(t, err) + + got, err := c.Decompress(compressed) + require.NoError(t, err) + require.True(t, bytes.Equal(tc.data, got)) + }) + } + }) + + t.Run("compress", func(t *testing.T) { + t.Parallel() + + t.Run("output differs from input", func(t *testing.T) { + t.Parallel() + + data := bytes.Repeat([]byte("a"), 1024) + compressed, err := c.Compress(data) + require.NoError(t, err) + require.NotEqual(t, data, compressed) + }) + + t.Run("compressible data shrinks", func(t *testing.T) { + t.Parallel() + + data := bytes.Repeat([]byte("a"), 1024) + compressed, err := c.Compress(data) + require.NoError(t, err) + require.Less(t, len(compressed), len(data)) + }) + + t.Run("does not modify input", func(t *testing.T) { + t.Parallel() + + data := []byte("hello, world") + original := bytes.Clone(data) + _, err := c.Compress(data) + require.NoError(t, err) + require.Equal(t, original, data) + }) + }) + + t.Run("decompress", func(t *testing.T) { + t.Parallel() + + t.Run("invalid data", func(t *testing.T) { + t.Parallel() + + _, err := c.Decompress([]byte("not compressed")) + require.Error(t, err) + }) + + t.Run("truncated frame", func(t *testing.T) { + t.Parallel() + + compressed, err := c.Compress([]byte("hello, world")) + require.NoError(t, err) + + _, err = c.Decompress(compressed[:len(compressed)/2]) + require.Error(t, err) + }) + + t.Run("does not modify input", func(t *testing.T) { + t.Parallel() + + compressed, err := c.Compress([]byte("hello, world")) + require.NoError(t, err) + + original := bytes.Clone(compressed) + _, err = c.Decompress(compressed) + require.NoError(t, err) + require.Equal(t, original, compressed) + }) + }) + + t.Run("concurrent", func(t *testing.T) { + t.Parallel() + + const n = 20 + + var wg sync.WaitGroup + + results := make([][]byte, n) + errs := make([]error, n) + + for i := range n { + wg.Go(func() { + data := bytes.Repeat([]byte{byte(i)}, 512) + compressed, compErr := c.Compress(data) + if compErr != nil { + errs[i] = compErr + return + } + + decompressed, decompErr := c.Decompress(compressed) + if decompErr != nil { + errs[i] = decompErr + return + } + + results[i] = decompressed + }) + } + + wg.Wait() + + for i := range n { + require.NoError(t, errs[i]) + require.Equal(t, bytes.Repeat([]byte{byte(i)}, 512), results[i]) + } + }) +} diff --git a/pkg/compress/doc.go b/pkg/compress/doc.go new file mode 100644 index 0000000..b77da1d --- /dev/null +++ b/pkg/compress/doc.go @@ -0,0 +1,8 @@ +// Package compress provides byte-slice compressors used to shrink payloads +// as they pass through the proxy. +// +// Each compressor ([Snappy], [Zstd], [LZ4]) exposes the same shape: an Encoding +// method returning the associated HTTP/gRPC content-encoding identifier, and +// Compress/Decompress methods that operate on whole in-memory buffers. Both +// treat their input as read-only and always return a newly allocated result. +package compress diff --git a/pkg/compress/lz4.go b/pkg/compress/lz4.go new file mode 100644 index 0000000..1a05edd --- /dev/null +++ b/pkg/compress/lz4.go @@ -0,0 +1,50 @@ +package compress + +import ( + "bytes" + "fmt" + "io" + + "github.com/pierrec/lz4/v4" +) + +// LZ4 compresses data using the LZ4 algorithm, which favors very fast +// compression and decompression over ratio. A zero value is ready to use and +// safe for concurrent use. +type LZ4 struct{} + +// NewLZ4 returns a ready-to-use [LZ4] compressor. +func NewLZ4() *LZ4 { + return new(LZ4) +} + +// Encoding returns the content-encoding identifier for LZ4-framed data. +func (c *LZ4) Encoding() string { + return "application/x-lz4" +} + +// Compress returns the LZ4-encoded form of data. It does not modify data. +func (c *LZ4) Compress(data []byte) ([]byte, error) { + var buf bytes.Buffer + w := lz4.NewWriter(&buf) + if _, err := io.Copy(w, bytes.NewReader(data)); err != nil { + return nil, fmt.Errorf("failed to compress with LZ4: %w", err) + } + + if err := w.Close(); err != nil { + return nil, fmt.Errorf("failed to compress with LZ4: %w", err) + } + + return buf.Bytes(), nil +} + +// Decompress returns the decoded form of LZ4-encoded data, or an error if data +// is not a valid LZ4 frame. It does not modify data. +func (c *LZ4) Decompress(data []byte) ([]byte, error) { + out, err := io.ReadAll(lz4.NewReader(bytes.NewReader(data))) + if err != nil { + return nil, fmt.Errorf("failed to decompress with LZ4: %w", err) + } + + return out, nil +} diff --git a/pkg/compress/lz4_test.go b/pkg/compress/lz4_test.go new file mode 100644 index 0000000..16bf0ef --- /dev/null +++ b/pkg/compress/lz4_test.go @@ -0,0 +1,21 @@ +package compress_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/temporalio/temporal-proxy/pkg/compress" +) + +func TestLZ4(t *testing.T) { + t.Parallel() + + runCompressorSuite(t, compress.NewLZ4()) +} + +func TestLZ4Encoding(t *testing.T) { + t.Parallel() + + require.Equal(t, "application/x-lz4", compress.NewLZ4().Encoding()) +} diff --git a/pkg/compress/snappy.go b/pkg/compress/snappy.go new file mode 100644 index 0000000..3c8d6e3 --- /dev/null +++ b/pkg/compress/snappy.go @@ -0,0 +1,38 @@ +package compress + +import ( + "fmt" + + "github.com/golang/snappy" +) + +// Snappy compresses data using Google's Snappy algorithm, which favors speed +// over ratio. A zero value is ready to use and safe for concurrent use. +type Snappy struct{} + +// NewSnappy returns a ready-to-use [Snappy] compressor. +func NewSnappy() *Snappy { + return new(Snappy) +} + +// Encoding returns the content-encoding identifier for Snappy-framed data. +func (c *Snappy) Encoding() string { + return "application/x-snappy-framed" +} + +// Compress returns the Snappy-encoded form of data. It never fails and does +// not modify data. +func (c *Snappy) Compress(data []byte) ([]byte, error) { + return snappy.Encode(nil, data), nil +} + +// Decompress returns the decoded form of Snappy-encoded data, or an error if +// data is not a valid Snappy frame. It does not modify data. +func (c *Snappy) Decompress(data []byte) ([]byte, error) { + out, err := snappy.Decode(nil, data) + if err != nil { + return nil, fmt.Errorf("failed to decompress with snappy: %w", err) + } + + return out, nil +} diff --git a/pkg/compress/snappy_test.go b/pkg/compress/snappy_test.go new file mode 100644 index 0000000..f12ccbd --- /dev/null +++ b/pkg/compress/snappy_test.go @@ -0,0 +1,21 @@ +package compress_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/temporalio/temporal-proxy/pkg/compress" +) + +func TestSnappy(t *testing.T) { + t.Parallel() + + runCompressorSuite(t, compress.NewSnappy()) +} + +func TestSnappyEncoding(t *testing.T) { + t.Parallel() + + require.Equal(t, "application/x-snappy-framed", compress.NewSnappy().Encoding()) +} diff --git a/pkg/compress/zstd.go b/pkg/compress/zstd.go new file mode 100644 index 0000000..cf13e9c --- /dev/null +++ b/pkg/compress/zstd.go @@ -0,0 +1,57 @@ +package compress + +import ( + "fmt" + + "github.com/klauspost/compress/zstd" +) + +// Zstd compresses data using the Zstandard algorithm, which offers higher +// ratios than [Snappy] at the cost of more CPU. It holds a reusable encoder +// and decoder and is safe for concurrent use; construct it with [NewZstd]. +type Zstd struct { + enc *zstd.Encoder + dec *zstd.Decoder +} + +// NewZstd returns a [Zstd] compressor configured at the given zstd level (see +// [zstd.EncoderLevelFromZstd]). It returns an error if the encoder or decoder +// cannot be created. +func NewZstd(level int) (*Zstd, error) { + enc, err := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.EncoderLevelFromZstd(level))) + if err != nil { + return nil, fmt.Errorf("failed to create Zstd encoder level: %d, %w", level, err) + } + + dec, err := zstd.NewReader(nil) + if err != nil { + return nil, fmt.Errorf("failed to create Zstd decoder: %w", err) + } + + return &Zstd{ + enc: enc, + dec: dec, + }, nil +} + +// Encoding returns the content-encoding identifier for Zstandard data. +func (c *Zstd) Encoding() string { + return "application/zstd" +} + +// Compress returns the Zstandard-encoded form of data. It never fails and does +// not modify data. +func (c *Zstd) Compress(data []byte) ([]byte, error) { + return c.enc.EncodeAll(data, make([]byte, 0, len(data))), nil +} + +// Decompress returns the decoded form of Zstandard-encoded data, or an error +// if data is not a valid Zstandard frame. It does not modify data. +func (c *Zstd) Decompress(data []byte) ([]byte, error) { + ed, err := c.dec.DecodeAll(data, nil) + if err != nil { + return nil, fmt.Errorf("failed to decompress with zstd: %w", err) + } + + return ed, nil +} diff --git a/pkg/compress/zstd_test.go b/pkg/compress/zstd_test.go new file mode 100644 index 0000000..481378c --- /dev/null +++ b/pkg/compress/zstd_test.go @@ -0,0 +1,49 @@ +package compress_test + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/temporalio/temporal-proxy/pkg/compress" +) + +func TestNewZstdCompressor(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + level int + }{ + {"fastest", 1}, + {"default", 3}, + {"best compression", 19}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + c, err := compress.NewZstd(tc.level) + require.NoError(t, err) + require.NotNil(t, c) + }) + } +} + +func TestZstdCompressor(t *testing.T) { + t.Parallel() + + c, err := compress.NewZstd(3) + require.NoError(t, err) + + runCompressorSuite(t, c) +} + +func TestZstdEncoding(t *testing.T) { + t.Parallel() + + c, err := compress.NewZstd(3) + require.NoError(t, err) + require.Equal(t, "application/zstd", c.Encoding()) +}