[compress]: Add payload compression algorithms

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

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=

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

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

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

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

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

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

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