fix #66

scanner.go

@@ -4,19 +4,23 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"errors"
 	"io"
 
 	typesv1 "github.com/humanlogio/api/go/types/v1"
 	"github.com/humanlogio/humanlog/pkg/sink"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
+const maxBufferSize = 1024 * 1024
+
 // Scan reads JSON-structured lines from src and prettify them onto dst. If
 // the lines aren't JSON-structured, it will simply write them out with no
 // prettification.
 func Scan(ctx context.Context, src io.Reader, sink sink.Sink, opts *HandlerOptions) error {
+
 	in := bufio.NewScanner(src)
-	in.Buffer(make([]byte, 1024*1024), 1024*1024)
+	in.Buffer(make([]byte, 0, maxBufferSize), maxBufferSize)
 	in.Split(bufio.ScanLines)
 
 	var line uint64
@@ -28,7 +32,26 @@ func Scan(ctx context.Context, src io.Reader, sink sink.Sink, opts *HandlerOptio
 	data := new(typesv1.StructuredLogEvent)
 	ev.Structured = data
 
-	for in.Scan() {
+	skipNextScan := false
+	for {
+		if !in.Scan() {
+			err := in.Err()
+			if err == nil || errors.Is(err, io.EOF) {
+				break
+			}
+			if errors.Is(err, bufio.ErrTooLong) {
+				in = bufio.NewScanner(src)
+				in.Buffer(make([]byte, 0, maxBufferSize), maxBufferSize)
+				in.Split(bufio.ScanLines)
+				skipNextScan = true
+				continue
+			}
+			break
+		}
+		if skipNextScan {
+			skipNextScan = false
+			continue
+		}
 
 		line++
 		lineData := in.Bytes()

scanner_test.go

@@ -46,6 +46,58 @@ func TestScannerLongLine(t *testing.T) {
 	}
 }
 
+func TestLargePayload(t *testing.T) {
+
+	ctx := context.Background()
+	payload := `{"msg": "hello world"}`
+	payload += "\n" + `{"msg":` + strings.Repeat("a", maxBufferSize+1) + `}` // more than 1mb long json payload
+	payload += "\n" + `{"msg": "안녕하세요"}`
+	payload += "\n" + `{"msg":` + strings.Repeat("a", maxBufferSize*3+1) + `}` // more than 3mb long json payload
+
+	now := time.Date(2024, 10, 11, 15, 25, 6, 0, time.UTC)
+	want := []*typesv1.LogEvent{
+		{
+			ParsedAt: timestamppb.New(now),
+			Raw:      []byte(`{"msg": "hello world"}`),
+			Structured: &typesv1.StructuredLogEvent{
+				Msg:       "hello world",
+				Timestamp: timestamppb.New(time.Time{}),
+			},
+		},
+		{
+			ParsedAt: timestamppb.New(now),
+			Raw:      []byte(`{"msg": "안녕하세요"}`),
+			Structured: &typesv1.StructuredLogEvent{
+				Msg:       "안녕하세요",
+				Timestamp: timestamppb.New(time.Time{}),
+			},
+		},
+	}
+
+	src := strings.NewReader(payload)
+
+	opts := DefaultOptions()
+	opts.timeNow = func() time.Time {
+		return now
+	}
+
+	sink := bufsink.NewSizedBufferedSink(100, nil)
+	err := Scan(ctx, src, sink, opts)
+	require.NoError(t, err)
+
+	got := sink.Buffered
+	require.Equal(t, pjsonslice(want), pjsonslice(got))
+}
+
+func pjsonslice[E proto.Message](m []E) string {
+	sb := strings.Builder{}
+	for _, e := range m {
+		sb.WriteString(pjson(e))
+		sb.WriteRune('\n')
+	}
+	return sb.String()
+}
+
 func pjson(m proto.Message) string {
 	o, err := protojson.Marshal(m)
 	if err != nil {