fix(spop): support configurable frame sizes

spop/README.md

@@ -1,9 +1,28 @@
 # A HAProxy SPOE implementation in Go
 
+## Frame size
 
-# References
+Agents accept SPOP frames up to 65,535 bytes by default. Set `Agent.MaxFrameSize`
+when HAProxy is configured to advertise larger frames:
+
+```go
+agent := spop.Agent{
+	Addr:         ":9000",
+	Handler:      handler,
+	MaxFrameSize: 256*1024 - 4,
+}
+log.Fatal(agent.ListenAndServe())
+```
+
+The configured value excludes the four-byte frame-length prefix. It is an
+upper bound: each connection negotiates the lower of this value and HAProxy's
+advertised `max-frame-size`. Values below 256 are invalid. Each active pooled
+frame buffer is allocated at the configured size, so use the smallest ceiling
+that accommodates the HAProxy configuration.
+
+## References
 https://www.haproxy.org/download/2.0/doc/SPOE.txt
 
-# Alternative implementations
+## Alternative implementations
 https://github.com/criteo/haproxy-spoe-go
 https://github.com/negasus/haproxy-spoe-go

spop/agent.go

@@ -14,6 +14,11 @@ type Agent struct {
 	Handler     Handler
 	BaseContext context.Context
 	Addr        string
+
+	// MaxFrameSize is the maximum SPOP frame size, excluding the four-byte
+	// length prefix. A zero value uses DefaultMaxFrameSize; values below 256
+	// are invalid. Each pooled frame buffer is sized to this limit.
+	MaxFrameSize uint32
 }
 
 func ListenAndServe(addr string, handler Handler) error {
@@ -37,6 +42,12 @@ func (a *Agent) Serve(l net.Listener) error {
 		a.BaseContext = context.Background()
 	}
 
+	maxFrameSize, err := a.configuredMaxFrameSize()
+	if err != nil {
+		return err
+	}
+	framePool := newFramePool(maxFrameSize)
+
 	go func() {
 		<-a.BaseContext.Done()
 		l.Close()
@@ -60,7 +71,7 @@ func (a *Agent) Serve(l net.Listener) error {
 			}
 		}
 
-		p := newProtocolClient(a.BaseContext, nc, as, a.Handler)
+		p := newProtocolClient(a.BaseContext, nc, as, framePool, a.Handler)
 		go func() {
 			defer nc.Close()
 			defer p.Close()
@@ -73,6 +84,22 @@ func (a *Agent) Serve(l net.Listener) error {
 	}
 }
 
+func (a *Agent) configuredMaxFrameSize() (uint32, error) {
+	maxFrameSize := a.MaxFrameSize
+	if maxFrameSize == 0 {
+		maxFrameSize = DefaultMaxFrameSize
+	}
+
+	if maxFrameSize < minFrameSize {
+		return 0, fmt.Errorf("max frame size must be at least %d: %d", minFrameSize, maxFrameSize)
+	}
+	if uint64(maxFrameSize) > uint64(^uint(0)>>1) {
+		return 0, fmt.Errorf("max frame size exceeds platform limit: %d", maxFrameSize)
+	}
+
+	return maxFrameSize, nil
+}
+
 func wrapPanic(fn func() error) (err error) {
 	didPanic := true
 	defer func() {

spop/agent_test.go

@@ -0,0 +1,36 @@
+package spop
+
+import "testing"
+
+func TestAgentConfiguredMaxFrameSize(t *testing.T) {
+	tests := []struct {
+		name    string
+		value   uint32
+		want    uint32
+		wantErr bool
+	}{
+		{name: "default", want: DefaultMaxFrameSize},
+		{name: "protocol minimum", value: minFrameSize, want: minFrameSize},
+		{name: "large frame", value: 262140, want: 262140},
+		{name: "below protocol minimum", value: minFrameSize - 1, wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			a := Agent{MaxFrameSize: tt.value}
+			got, err := a.configuredMaxFrameSize()
+			if tt.wantErr {
+				if err == nil {
+					t.Fatal("expected an error")
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("configured max frame size: %v", err)
+			}
+			if got != tt.want {
+				t.Fatalf("expected %d, got %d", tt.want, got)
+			}
+		})
+	}
+}

spop/error.go

@@ -1,9 +1,35 @@
 package spop
 
 import (
+	"errors"
 	"fmt"
 )
 
+type protocolError struct {
+	err  error
+	code errorCode
+}
+
+func newProtocolError(code errorCode, format string, args ...any) error {
+	return &protocolError{code: code, err: fmt.Errorf(format, args...)}
+}
+
+func (e *protocolError) Error() string {
+	return e.err.Error()
+}
+
+func (e *protocolError) Unwrap() error {
+	return e.err
+}
+
+func protocolErrorCode(err error) errorCode {
+	var protocolErr *protocolError
+	if errors.As(err, &protocolErr) {
+		return protocolErr.code
+	}
+	return ErrorUnknown
+}
+
 type errorCode int
 
 func (e errorCode) String() string {

spop/frame.go

@@ -13,25 +13,39 @@ import (
 
 const uint32Len = 4
 
-var framePool = sync.Pool{
-	New: func() any {
+type framePool struct {
+	pool         sync.Pool
+	maxFrameSize uint32
+}
+
+func newFramePool(maxFrameSize uint32) *framePool {
+	p := &framePool{maxFrameSize: maxFrameSize}
+	p.pool.New = func() any {
 		return &frame{
+			buf:    buffer.NewSliceBuffer(int(maxFrameSize)),
 			length: make([]byte, uint32Len),
-			buf:    buffer.NewSliceBuffer(maxFrameSize),
+			pool:   p,
 		}
-	},
+	}
+	return p
+}
+
+var defaultFramePool = newFramePool(DefaultMaxFrameSize)
+
+func (p *framePool) acquire() *frame {
+	return p.pool.Get().(*frame)
 }
 
 func acquireFrame() *frame {
-	return framePool.Get().(*frame)
+	return defaultFramePool.acquire()
 }
 
 func releaseFrame(f *frame) {
 	f.buf.Reset()
 	f.frameType = 0
 	f.meta = frameMetadata{}
 
-	framePool.Put(f)
+	f.pool.pool.Put(f)
 }
 
 type frameMetadata struct {
@@ -41,7 +55,8 @@ type frameMetadata struct {
 }
 
 type frame struct {
-	buf *buffer.SliceBuffer
+	buf  *buffer.SliceBuffer
+	pool *framePool
 
 	length []byte
 	meta   frameMetadata
@@ -50,13 +65,25 @@ type frame struct {
 }
 
 func (f *frame) ReadFrom(r io.Reader) (int64, error) {
+	return f.readFrom(r, f.pool.maxFrameSize)
+}
+
+func (f *frame) readFrom(r io.Reader, maxFrameSize uint32) (int64, error) {
 	if _, err := io.ReadFull(r, f.length); err != nil {
 		return 0, fmt.Errorf("reading frame length: %w", err)
 	}
 	frameLen := binary.BigEndian.Uint32(f.length)
 
+	if maxFrameSize > f.pool.maxFrameSize {
+		maxFrameSize = f.pool.maxFrameSize
+	}
 	if frameLen > maxFrameSize {
-		return int64(len(f.length)), fmt.Errorf("frame length %d exceeds maximum %d", frameLen, maxFrameSize)
+		return int64(len(f.length)), newProtocolError(
+			ErrorTooBig,
+			"frame length %d exceeds maximum %d",
+			frameLen,
+			maxFrameSize,
+		)
 	}
 
 	f.buf.Reset()
@@ -72,7 +99,24 @@ func (f *frame) ReadFrom(r io.Reader) (int64, error) {
 }
 
 func (f *frame) WriteTo(w io.Writer) (int64, error) {
-	binary.BigEndian.PutUint32(f.length, uint32(f.buf.Len()))
+	return f.writeTo(w, f.pool.maxFrameSize)
+}
+
+func (f *frame) writeTo(w io.Writer, maxFrameSize uint32) (int64, error) {
+	if maxFrameSize > f.pool.maxFrameSize {
+		maxFrameSize = f.pool.maxFrameSize
+	}
+	frameLen := uint32(f.buf.Len())
+	if frameLen > maxFrameSize {
+		return 0, newProtocolError(
+			ErrorTooBig,
+			"frame length %d exceeds maximum %d",
+			frameLen,
+			maxFrameSize,
+		)
+	}
+
+	binary.BigEndian.PutUint32(f.length, frameLen)
 
 	if n, err := w.Write(f.length); err != nil {
 		return int64(n), err

spop/frame_test.go

@@ -37,9 +37,9 @@ func TestFrame_ReadFrom_ExceedsMaxFrameSize(t *testing.T) {
 	})
 
 	t.Run("accepts frame length at maxFrameSize boundary", func(t *testing.T) {
-		buf := writeFrameLength(maxFrameSize)
+		buf := writeFrameLength(DefaultMaxFrameSize)
 
-		frameData := make([]byte, maxFrameSize)
+		frameData := make([]byte, DefaultMaxFrameSize)
 		frameData[0] = byte(frameTypeIDHaproxyHello)
 		binary.BigEndian.PutUint32(frameData[1:5], 0)
 		frameData[5] = 0 // streamID varint
@@ -59,12 +59,42 @@ func TestFrame_ReadFrom_ExceedsMaxFrameSize(t *testing.T) {
 	})
 
 	t.Run("rejects frame length one byte over maxFrameSize", func(t *testing.T) {
-		buf := writeFrameLength(maxFrameSize + 1)
+		buf := writeFrameLength(DefaultMaxFrameSize + 1)
 
 		f := acquireFrame()
 		defer releaseFrame(f)
 
 		_, err := f.ReadFrom(buf)
 		assertError(t, err)
 	})
-}
+
+	t.Run("accepts configured frame length above default", func(t *testing.T) {
+		const configuredMaxFrameSize uint32 = 262140
+		buf := writeFrameLength(configuredMaxFrameSize)
+
+		frameData := make([]byte, configuredMaxFrameSize)
+		frameData[0] = byte(frameTypeIDHaproxyHello)
+		binary.BigEndian.PutUint32(frameData[1:5], 0)
+		frameData[5] = 0
+		frameData[6] = 0
+		buf.Write(frameData)
+
+		f := newFramePool(configuredMaxFrameSize).acquire()
+		defer releaseFrame(f)
+
+		if _, err := f.ReadFrom(buf); err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+
+	t.Run("rejects frame above configured maximum", func(t *testing.T) {
+		const configuredMaxFrameSize uint32 = 262140
+		buf := writeFrameLength(configuredMaxFrameSize + 1)
+
+		f := newFramePool(configuredMaxFrameSize).acquire()
+		defer releaseFrame(f)
+
+		_, err := f.ReadFrom(buf)
+		assertError(t, err)
+	})
+}

spop/frames.go

@@ -43,7 +43,11 @@ type AgentDisconnectFrame struct {
 }
 
 func (a *AgentDisconnectFrame) WriteTo(w io.Writer) (int64, error) {
-	f := acquireFrame()
+	return a.writeTo(w, defaultFramePool, DefaultMaxFrameSize)
+}
+
+func (a *AgentDisconnectFrame) writeTo(w io.Writer, pool *framePool, maxFrameSize uint32) (int64, error) {
+	f := pool.acquire()
 	defer releaseFrame(f)
 
 	f.frameType = frameTypeIDAgentDisconnect
@@ -66,7 +70,7 @@ func (a *AgentDisconnectFrame) WriteTo(w io.Writer) (int64, error) {
 
 	f.buf.AdvanceW(kvw.Off())
 
-	return f.WriteTo(w)
+	return f.writeTo(w, maxFrameSize)
 }
 
 const (
@@ -89,7 +93,11 @@ type AgentHelloFrame struct {
 }
 
 func (a *AgentHelloFrame) WriteTo(w io.Writer) (int64, error) {
-	f := acquireFrame()
+	return a.writeTo(w, defaultFramePool, DefaultMaxFrameSize)
+}
+
+func (a *AgentHelloFrame) writeTo(w io.Writer, pool *framePool, maxFrameSize uint32) (int64, error) {
+	f := pool.acquire()
 	defer releaseFrame(f)
 
 	f.frameType = frameTypeIDAgentHello
@@ -116,7 +124,7 @@ func (a *AgentHelloFrame) WriteTo(w io.Writer) (int64, error) {
 	}
 	f.buf.AdvanceW(kvw.Off())
 
-	return f.WriteTo(w)
+	return f.writeTo(w, maxFrameSize)
 }
 
 type AckFrame struct {
@@ -126,7 +134,11 @@ type AckFrame struct {
 }
 
 func (a *AckFrame) WriteTo(w io.Writer) (int64, error) {
-	f := acquireFrame()
+	return a.writeTo(w, defaultFramePool, DefaultMaxFrameSize)
+}
+
+func (a *AckFrame) writeTo(w io.Writer, pool *framePool, maxFrameSize uint32) (int64, error) {
+	f := pool.acquire()
 	defer releaseFrame(f)
 
 	f.frameType = frameTypeIDAck
@@ -148,5 +160,5 @@ func (a *AckFrame) WriteTo(w io.Writer) (int64, error) {
 
 	f.buf.AdvanceW(aw.Off())
 
-	return f.WriteTo(w)
+	return f.writeTo(w, maxFrameSize)
 }