prevent excessive buffer allocation in stream marshal/unmarshal

docs/streaming.md

@@ -168,7 +168,7 @@ Creates a new stream encoder:
 ```go
 import "github.com/pk910/dynamic-ssz/sszutils"
 
-encoder := sszutils.NewStreamEncoder(writer, 0) // 0 = default 2KB buffer
+encoder := sszutils.NewStreamEncoder(writer, 0, 0) // 0 = default 2KB write buffer, 0 = default 200KB max delegation buffer
 ```
 
 The `StreamEncoder`:
@@ -182,7 +182,7 @@ The `StreamEncoder`:
 Creates a new stream decoder:
 
 ```go
-decoder := sszutils.NewStreamDecoder(reader, totalSize, 0) // 0 = default 2KB buffer
+decoder := sszutils.NewStreamDecoder(reader, totalSize, 0, 0) // 0 = default 2KB read buffer, 0 = default 200KB max delegation buffer
 ```
 
 The `StreamDecoder`:

dynssz.go

@@ -397,7 +397,7 @@ func (d *DynSsz) MarshalSSZTo(source any, buf []byte, opts ...CallOption) ([]byt
 //	err = ds.MarshalSSZWriter(block, conn)
 func (d *DynSsz) MarshalSSZWriter(source any, w io.Writer, opts ...CallOption) error {
 	cfg := applyCallOptions(opts)
-	encoder := sszutils.NewStreamEncoder(w, d.options.StreamWriterBufferSize)
+	encoder := sszutils.NewStreamEncoder(w, d.options.StreamWriterBufferSize, d.options.StreamWriterMaxBufferSize)
 
 	// Skip view descriptor logic for types implementing DynamicEncoder
 	if cfg == nil || cfg.viewDescriptor == nil {
@@ -662,7 +662,7 @@ func (d *DynSsz) UnmarshalSSZ(target any, ssz []byte, opts ...CallOption) error
 //	err = ds.UnmarshalSSZReader(&block, conn, -1)
 func (d *DynSsz) UnmarshalSSZReader(target any, r io.Reader, size int, opts ...CallOption) error {
 	cfg := applyCallOptions(opts)
-	decoder := sszutils.NewStreamDecoder(r, size, d.options.StreamReaderBufferSize)
+	decoder := sszutils.NewStreamDecoder(r, size, d.options.StreamReaderBufferSize, d.options.StreamReaderMaxBufferSize)
 	decoder.PushLimit(size)
 
 	// Skip view descriptor logic for types implementing DynamicDecoder

options.go

@@ -10,13 +10,15 @@ type DynSszOption func(*DynSszOptions)
 
 // DynSszOptions holds the configuration options for a DynSsz instance.
 type DynSszOptions struct {
-	NoFastSsz              bool
-	NoFastHash             bool
-	ExtendedTypes          bool
-	Verbose                bool
-	LogCb                  func(format string, args ...any)
-	StreamWriterBufferSize int
-	StreamReaderBufferSize int
+	NoFastSsz                 bool
+	NoFastHash                bool
+	ExtendedTypes             bool
+	Verbose                   bool
+	LogCb                     func(format string, args ...any)
+	StreamWriterBufferSize    int
+	StreamWriterMaxBufferSize int
+	StreamReaderBufferSize    int
+	StreamReaderMaxBufferSize int
 }
 
 // WithNoFastSsz disables fastssz fallback for types that implement fastssz
@@ -68,14 +70,36 @@ func WithStreamWriterBufferSize(size int) DynSszOption {
 	}
 }
 
-// WithStreamReaderBufferSize sets the maximum internal buffer size for the
-// streaming SSZ decoder used by UnmarshalSSZReader. Defaults to 2KB if not set.
+// WithStreamWriterMaxBufferSize sets the maximum buffer size for delegating to
+// buffer-based marshal methods during streaming SSZ encoding. When a type's
+// serialized size exceeds this limit, the encoder falls through to
+// reflection-based field-by-field marshalling instead of buffering the entire
+// object. Defaults to 200KB if not set.
+func WithStreamWriterMaxBufferSize(size int) DynSszOption {
+	return func(opts *DynSszOptions) {
+		opts.StreamWriterMaxBufferSize = size
+	}
+}
+
+// WithStreamReaderBufferSize sets the internal buffer size for the streaming
+// SSZ decoder used by UnmarshalSSZReader. Defaults to 2KB if not set.
 func WithStreamReaderBufferSize(size int) DynSszOption {
 	return func(opts *DynSszOptions) {
 		opts.StreamReaderBufferSize = size
 	}
 }
 
+// WithStreamReaderMaxBufferSize sets the maximum buffer size for delegating to
+// buffer-based unmarshal methods during streaming SSZ decoding. When a type's
+// serialized size exceeds this limit, the decoder falls through to
+// reflection-based field-by-field unmarshalling instead of buffering the entire
+// object. Defaults to 200KB if not set.
+func WithStreamReaderMaxBufferSize(size int) DynSszOption {
+	return func(opts *DynSszOptions) {
+		opts.StreamReaderMaxBufferSize = size
+	}
+}
+
 // CallOption is a functional option for per-call configuration of MarshalSSZ,
 // UnmarshalSSZ, and HashTreeRoot operations. These options allow runtime
 // customization of SSZ encoding behavior without modifying the DynSsz instance.

reflection/marshal.go

@@ -74,12 +74,14 @@
 
 		if useFastSsz {
 			if marshaller, ok := getPtr(sourceValue).Interface().(sszutils.FastsszMarshaler); ok {
-				newBuf, err := marshaller.MarshalSSZTo(encoder.GetBuffer())
-				if err != nil {
-					return err
+				if int(sourceType.Size) <= encoder.MaxEncodeBufferSize() || sourceType.SszType == ssztypes.SszCustomType {
+					newBuf, err := marshaller.MarshalSSZTo(encoder.GetBuffer())
+					if err != nil {
+						return err
+					}
+					encoder.SetBuffer(newBuf)
+					return nil
 				}
-				encoder.SetBuffer(newBuf)
-				return nil
 			}
 		}
 
@@ -94,12 +96,14 @@
 
 		if useDynamicMarshal {
 			if marshaller, ok := getPtr(sourceValue).Interface().(sszutils.DynamicMarshaler); ok {
-				newBuf, err := marshaller.MarshalSSZDyn(ctx.ds, encoder.GetBuffer())
-				if err != nil {
-					return err
+				if encoder.Seekable() || (sourceType.Size > 0 && int(sourceType.Size) <= encoder.MaxEncodeBufferSize()) {
+					newBuf, err := marshaller.MarshalSSZDyn(ctx.ds, encoder.GetBuffer())
+					if err != nil {
+						return err
+					}
+					encoder.SetBuffer(newBuf)
+					return nil
 				}
-				encoder.SetBuffer(newBuf)
-				return nil
 			}
 		}
 	}
@@ -214,12 +218,14 @@
 	if useViewMarshaler {
 		if marshaller, ok := getPtr(sourceValue).Interface().(sszutils.DynamicViewMarshaler); ok {
 			if marshalFn := marshaller.MarshalSSZDynView(*sourceType.CodegenInfo); marshalFn != nil {
-				newBuf, err := marshalFn(ctx.ds, encoder.GetBuffer())
-				if err != nil {
-					return true, err
+				if encoder.Seekable() || (sourceType.Size > 0 && int(sourceType.Size) <= encoder.MaxEncodeBufferSize()) {
+					newBuf, err := marshalFn(ctx.ds, encoder.GetBuffer())
+					if err != nil {
+						return true, err
+					}
+					encoder.SetBuffer(newBuf)
+					return true, nil
 				}
-				encoder.SetBuffer(newBuf)
-				return true, nil
 			}
 		}
 	}

reflection/marshal_test.go

@@ -1504,7 +1504,7 @@ func TestMarshalDynamicListNonSeekableSizeError(t *testing.T) {
 	listDesc.ElemDesc = &elemDescCopy
 
 	ctx := reflection.NewReflectionCtx(nil, nil, false, true)
-	encoder := sszutils.NewStreamEncoder(bytes.NewBuffer(nil), 0)
+	encoder := sszutils.NewStreamEncoder(bytes.NewBuffer(nil), 0, 0)
 	data := []DynElem{{Value: 1}, {Value: 2}}
 	err = ctx.MarshalSSZ(listDesc, reflect.ValueOf(data), encoder)
 	if err == nil {

reflection/overflow_internal_test.go

@@ -306,7 +306,7 @@ func TestMarshalLargeVectorStreaming(t *testing.T) {
 	val := reflect.ValueOf(src)
 
 	// Use StreamEncoder writing to io.Discard so we don't buffer output.
-	enc := sszutils.NewStreamEncoder(io.Discard, 4096)
+	enc := sszutils.NewStreamEncoder(io.Discard, 4096, 0)
 
 	err := ctx.marshalType(td, val, enc, 0)
 	if err != nil {

reflection/unmarshal.go

@@ -80,11 +80,13 @@ func (ctx *ReflectionCtx) unmarshalType(targetType *ssztypes.TypeDescriptor, tar
 			if unmarshaler, ok := targetValue.Addr().Interface().(sszutils.DynamicViewUnmarshaler); ok {
 				if unmarshalFn := unmarshaler.UnmarshalSSZDynView(*targetType.CodegenInfo); unmarshalFn != nil {
 					bufLen := decoder.GetLength()
-					buf, err := decoder.DecodeBytesBuf(bufLen)
-					if err != nil {
-						return err
+					if bufLen <= decoder.MaxDecodeBufferSize() {
+						buf, err := decoder.DecodeBytesBuf(bufLen)
+						if err != nil {
+							return err
+						}
+						return unmarshalFn(ctx.ds, buf)
 					}
-					return unmarshalFn(ctx.ds, buf)
 				}
 			}
 		}
@@ -109,11 +111,13 @@ func (ctx *ReflectionCtx) unmarshalType(targetType *ssztypes.TypeDescriptor, tar
 					}
 					sszLen = int(typeSize)
 				}
-				sszBuf, err := decoder.DecodeBytesBuf(sszLen)
-				if err != nil {
-					return err
+				if sszLen <= decoder.MaxDecodeBufferSize() || targetType.SszType == ssztypes.SszCustomType {
+					sszBuf, err := decoder.DecodeBytesBuf(sszLen)
+					if err != nil {
+						return err
+					}
+					return unmarshaller.UnmarshalSSZ(sszBuf)
 				}
-				return unmarshaller.UnmarshalSSZ(sszBuf)
 			}
 		}
 
@@ -136,11 +140,13 @@ func (ctx *ReflectionCtx) unmarshalType(targetType *ssztypes.TypeDescriptor, tar
 					}
 					sszLen = int(typeSize)
 				}
-				sszBuf, err := decoder.DecodeBytesBuf(sszLen)
-				if err != nil {
-					return err
+				if sszLen <= decoder.MaxDecodeBufferSize() {
+					sszBuf, err := decoder.DecodeBytesBuf(sszLen)
+					if err != nil {
+						return err
+					}
+					return unmarshaller.UnmarshalSSZDyn(ctx.ds, sszBuf)
 				}
-				return unmarshaller.UnmarshalSSZDyn(ctx.ds, sszBuf)
 			}
 		}
 	}

sszutils/decoder.go

@@ -22,4 +22,5 @@ type Decoder interface {
 	DecodeOffset() (uint32, error)
 	DecodeOffsetAt(pos int) uint32
 	SkipBytes(n int)
+	MaxDecodeBufferSize() int // max size for efficient DecodeBytesBuf without excessive allocation
 }

sszutils/decoder_buffer.go

@@ -6,6 +6,7 @@ package sszutils
 
 import (
 	"encoding/binary"
+	"math"
 )
 
 // BufferDecoder is a seekable Decoder implementation backed by an in-memory
@@ -39,6 +40,12 @@ func (e *BufferDecoder) Seekable() bool {
 	return true
 }
 
+// MaxDecodeBufferSize returns math.MaxInt because the entire buffer is already
+// in memory and DecodeBytesBuf only returns a slice—no allocation needed.
+func (e *BufferDecoder) MaxDecodeBufferSize() int {
+	return math.MaxInt
+}
+
 // GetPosition returns the current read position in the buffer.
 func (e *BufferDecoder) GetPosition() int {
 	return e.position

sszutils/decoder_stream.go

@@ -9,10 +9,15 @@ import (
 	"io"
 )
 
-// DefaultStreamDecoderBufSize is the default maximum buffer size for
+// DefaultStreamDecoderBufSize is the default internal read buffer size for
 // StreamDecoder (2KB).
 const DefaultStreamDecoderBufSize = 2 * 1024
 
+// DefaultStreamDecoderMaxBufSize is the default maximum buffer size for
+// delegation to buffer-based unmarshal methods (200KB). Reads exceeding this
+// threshold fall through to reflection-based field-by-field unmarshalling.
+const DefaultStreamDecoderMaxBufSize = 200 * 1024
+
 // StreamDecoder is a non-seekable Decoder implementation that reads SSZ data
 // from an io.Reader. It uses an internal buffer for efficient sequential reads
 // but does not support DecodeOffsetAt or SkipBytes.
@@ -24,23 +29,28 @@ type StreamDecoder struct {
 	position  int
 
 	// Internal buffer for reading from stream
-	buffer    []byte
-	bufferPos int // Current read position within buffer
-	bufferLen int // Amount of valid data in buffer
+	buffer     []byte
+	bufferPos  int // Current read position within buffer
+	bufferLen  int // Amount of valid data in buffer
+	maxBufSize int // Configured maximum buffer size
 }
 
 var _ Decoder = (*StreamDecoder)(nil)
 
 // NewStreamDecoder creates a new StreamDecoder that reads SSZ data from the
 // provided io.Reader. totalLen specifies the total expected byte length of the
-// SSZ payload. maxBufSize controls the maximum internal read buffer size; if
-// <= 0, DefaultStreamDecoderBufSize is used.
-func NewStreamDecoder(reader io.Reader, totalLen, maxBufSize int) *StreamDecoder {
+// SSZ payload. bufSize controls the internal read buffer size (defaults to 2KB
+// if <= 0). maxBufSize controls the maximum buffer size for delegation to
+// buffer-based unmarshal methods (defaults to 200KB if <= 0).
+func NewStreamDecoder(reader io.Reader, totalLen, bufSize, maxBufSize int) *StreamDecoder {
+	if bufSize <= 0 {
+		bufSize = DefaultStreamDecoderBufSize
+	}
 	if maxBufSize <= 0 {
-		maxBufSize = DefaultStreamDecoderBufSize
+		maxBufSize = DefaultStreamDecoderMaxBufSize
 	}
 	// Use smaller buffer for small streams
-	bufferSize := maxBufSize
+	bufferSize := bufSize
 	if totalLen < bufferSize {
 		bufferSize = totalLen
 	}
@@ -49,21 +59,29 @@ func NewStreamDecoder(reader io.Reader, totalLen, maxBufSize int) *StreamDecoder
 	}
 
 	return &StreamDecoder{
-		reader:    reader,
-		limits:    make([]int, 0, 16),
-		lastLimit: totalLen,
-		streamLen: totalLen,
-		position:  0,
-		buffer:    make([]byte, bufferSize),
-		bufferPos: 0,
-		bufferLen: 0,
+		reader:     reader,
+		limits:     make([]int, 0, 16),
+		lastLimit:  totalLen,
+		streamLen:  totalLen,
+		position:   0,
+		buffer:     make([]byte, bufferSize),
+		bufferPos:  0,
+		bufferLen:  0,
+		maxBufSize: maxBufSize,
 	}
 }
 
 func (e *StreamDecoder) Seekable() bool {
 	return false
 }
 
+// MaxDecodeBufferSize returns the configured maximum buffer size. Callers
+// should avoid DecodeBytesBuf calls that exceed this size to prevent the
+// stream decoder from allocating large temporary buffers.
+func (e *StreamDecoder) MaxDecodeBufferSize() int {
+	return e.maxBufSize
+}
+
 func (e *StreamDecoder) GetPosition() int {
 	return e.position
 }

sszutils/encoder.go

@@ -20,4 +20,5 @@ type Encoder interface {
 	EncodeOffset(v uint32)
 	EncodeOffsetAt(pos int, v uint32)
 	EncodeZeroPadding(n int)
+	MaxEncodeBufferSize() int // max size for efficient GetBuffer/SetBuffer delegation
 }

sszutils/encoder_buffer.go

@@ -6,6 +6,7 @@ package sszutils
 
 import (
 	"encoding/binary"
+	"math"
 )
 
 // BufferEncoder is a seekable Encoder implementation backed by an in-memory
@@ -33,6 +34,12 @@ func (e *BufferEncoder) Seekable() bool {
 	return true
 }
 
+// MaxEncodeBufferSize returns math.MaxInt because the entire output is already
+// buffered in memory—no additional allocation overhead from delegation.
+func (e *BufferEncoder) MaxEncodeBufferSize() int {
+	return math.MaxInt
+}
+
 // GetPosition returns the current write position in the buffer.
 func (e *BufferEncoder) GetPosition() int {
 	return e.pos