Repository: cloudwego/hertz

Issue:
SSE foreach:  bufio.Scanner: token too long
sse.NewReader方法中的buffer最大大小只有64KB，没有看到设置的入口，如何扩大这个限制？


Discussion:
@xiaost: 谢谢你的反馈，我们优化一下。

Solution patch:
diff --git a/go.mod b/go.mod
index 2d4d38f..0e8d1ef 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,7 @@ go 1.19
 
 require (
 	github.com/bytedance/gopkg v0.1.1
-	github.com/bytedance/sonic v1.13.2
+	github.com/bytedance/sonic v1.14.0
 	github.com/cloudwego/netpoll v0.7.0
 	github.com/fsnotify/fsnotify v1.5.4
 	github.com/nyaruka/phonenumbers v1.0.55
@@ -16,7 +16,7 @@ require (
 )
 
 require (
-	github.com/bytedance/sonic/loader v0.2.4 // indirect
+	github.com/bytedance/sonic/loader v0.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.5 // indirect
 	github.com/cloudwego/gopkg v0.1.4 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
diff --git a/go.sum b/go.sum
index b0faaf3..8c9c2c7 100644
--- a/go.sum
+++ b/go.sum
@@ -1,10 +1,10 @@
 github.com/bytedance/gopkg v0.1.1 h1:3azzgSkiaw79u24a+w9arfH8OfnQQ4MHUt9lJFREEaE=
 github.com/bytedance/gopkg v0.1.1/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM=
-github.com/bytedance/sonic v1.13.2 h1:8/H1FempDZqC4VqjptGo14QQlJx8VdZJegxs6wwfqpQ=
-github.com/bytedance/sonic v1.13.2/go.mod h1:o68xyaF9u2gvVBuGHPlUVCy+ZfmNNO5ETf1+KgkJhz4=
+github.com/bytedance/sonic v1.14.0 h1:/OfKt8HFw0kh2rj8N0F6C/qPGRESq0BbaNZgcNXXzQQ=
+github.com/bytedance/sonic v1.14.0/go.mod h1:WoEbx8WTcFJfzCe0hbmyTGrfjt8PzNEBdxlNUO24NhA=
 github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU=
-github.com/bytedance/sonic/loader v0.2.4 h1:ZWCw4stuXUsn1/+zQDqeE7JKP+QO47tz7QCNan80NzY=
-github.com/bytedance/sonic/loader v0.2.4/go.mod h1:N8A3vUdtUebEY2/VQC0MyhYeKUFosQU6FxH2JmUe6VI=
+github.com/bytedance/sonic/loader v0.3.0 h1:dskwH8edlzNMctoruo8FPTJDF3vLtDT0sXZwvZJyqeA=
+github.com/bytedance/sonic/loader v0.3.0/go.mod h1:N8A3vUdtUebEY2/VQC0MyhYeKUFosQU6FxH2JmUe6VI=
 github.com/cloudwego/base64x v0.1.5 h1:XPciSp1xaq2VCSt6lF0phncD4koWyULpl5bUxbfCyP4=
 github.com/cloudwego/base64x v0.1.5/go.mod h1:0zlkT4Wn5C6NdauXdJRhSKRlJvmclQ1hhJgA0rcu/8w=
 github.com/cloudwego/gopkg v0.1.4 h1:EoQiCG4sTonTPHxOGE0VlQs+sQR+Hsi2uN0qqwu8O50=
diff --git a/pkg/protocol/sse/reader.go b/pkg/protocol/sse/reader.go
index d6daebe..4c4561f 100644
--- a/pkg/protocol/sse/reader.go
+++ b/pkg/protocol/sse/reader.go
@@ -63,6 +63,18 @@ func NewReader(resp *protocol.Response) (*Reader, error) {
 	return r, nil
 }
 
+// SetMaxBufferSize sets the maximum buffer size for the scanner.
+//
+// The scanner will allocate its own buffer as needed, up to max bytes.
+// The default max size without calling this method is bufio.MaxScanTokenSize (64KB).
+//
+// It panics if it is called after reading event has started.
+func (r *Reader) SetMaxBufferSize(max int) {
+	// NOTE: Consider using bytebufferpool if GC becomes an issue.
+	// Currently using nil to let scanner manage its own buffer internally.
+	r.s.Buffer(nil, max)
+}
+
 type forceCloseIf interface {
 	ForceClose() error // implemented by *clientRespStream
 }
@@ -70,13 +82,14 @@ type forceCloseIf interface {
 // ForEach iterates over all SSE events in the response body,
 // invoking the provided handler function for each event.
 //
-// `f` MUST NOT keep the Event reference after the func returns.
-// Use (*Event).Clone to create a copy instead.
-//
-// Iteration stops if the handler returns an error or if reading fails.
-// `ctx` is used to cancel the iteration if ctx.Done() != nil
+// The handler MUST NOT keep the Event reference after returning.
+// Use (*Event).Clone to create a copy if needed.
 //
-// It returns nil if all events are processed successfully.
+// Iteration stops when:
+//   - The handler returns an error
+//   - Reading fails (e.g., bufio.ErrTooLong for events exceeding buffer size)
+//   - Context is cancelled (if ctx.Done() != nil)
+//   - All events are processed (returns nil)
 func (r *Reader) ForEach(ctx context.Context, f func(e *Event) error) error {
 	if ctx.Done() != nil {
 		ch := make(chan struct{})
@@ -130,7 +143,9 @@ func (r *Reader) onEventRead(e *Event) {
 // Read reads a single SSE event from the response body.
 //
 // It populates the provided Event struct with the parsed data.
-// Returns nil if an event was successfully read, or an error otherwise.
+// Returns nil on success, io.EOF when no more events, or an error
+// (e.g., bufio.ErrTooLong if an event line exceeds the buffer size).
+// Use SetMaxBufferSize to handle larger events.
 func (r *Reader) Read(e *Event) error {
 	e.Reset()
 	for i := 0; r.s.Scan(); i++ {



Test patch:
diff --git a/pkg/protocol/sse/reader_test.go b/pkg/protocol/sse/reader_test.go
index 63af535..6a5c645 100644
--- a/pkg/protocol/sse/reader_test.go
+++ b/pkg/protocol/sse/reader_test.go
@@ -17,6 +17,7 @@
 package sse
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	"errors"
@@ -343,3 +344,59 @@ func TestReader_ForEach(t *testing.T) {
 	})
 	assert.Assert(t, err == ctx.Err())
 }
+
+func TestReader_SetMaxBufferSize(t *testing.T) {
+	// Test that default buffer size fails for events > 64KB
+	t.Run("default buffer size fails for large events", func(t *testing.T) {
+		// Create a response with a large event (65KB) - just over default 64KB
+		largeData := strings.Repeat("x", 65*1024)
+		input := "event: large\ndata: " + largeData + "\n\n"
+
+		resp := &protocol.Response{}
+		resp.Header.SetContentType(string(bytestr.MIMETextEventStream))
+		resp.SetBody([]byte(input))
+
+		r, err := NewReader(resp)
+		assert.Assert(t, err == nil)
+
+		// Don't call SetMaxBufferSize, use default (64KB)
+		// Reading should fail because the line is too long
+		e := NewEvent()
+		err = r.Read(e)
+		assert.Assert(t, errors.Is(err, bufio.ErrTooLong))
+		e.Release()
+	})
+
+	// Test with custom buffer size for large events
+	t.Run("custom buffer size", func(t *testing.T) {
+		// Create a response with a large event (65KB) - just over default 64KB
+		largeData := strings.Repeat("x", 65*1024)
+		input := "event: large\ndata: " + largeData + "\n\n"
+
+		resp := &protocol.Response{}
+		resp.Header.SetContentType(string(bytestr.MIMETextEventStream))
+		resp.SetBody([]byte(input))
+
+		r, err := NewReader(resp)
+		assert.Assert(t, err == nil)
+
+		// Set max buffer size to 70KB to handle the large event
+		r.SetMaxBufferSize(70 * 1024)
+
+		// Should be able to read the large event
+		e := NewEvent()
+		err = r.Read(e)
+		assert.Assert(t, err == nil)
+		assert.DeepEqual(t, "large", e.Type)
+		assert.DeepEqual(t, largeData, string(e.Data))
+		e.Release()
+
+		// Test panic when SetMaxBufferSize is called after reading
+		defer func() {
+			if r := recover(); r == nil {
+				t.Error("SetMaxBufferSize should panic after reading has started")
+			}
+		}()
+		r.SetMaxBufferSize(80 * 1024)
+	})
+}