internal/streamer/streamer.go

@@ -53,7 +53,8 @@ const (
 	metricsLogInterval    = 10 * time.Second
 	bytesPerKB            = 1024
 	bytesPerMB            = 1024 * 1024
-	maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers
+	maxLineSize           = bytesPerMB // Well above an extended BGP message (RFC 8654) in hex plus its prefixes
+	maxConcurrentHandlers = 800        // Maximum number of concurrent message handlers
 
 	// Backpressure constants
 	backpressureThreshold = 0.5 // Start dropping at 50% queue utilization
@@ -569,6 +570,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 	}()
 
 	scanner := bufio.NewScanner(reader)
+	scanner.Buffer(nil, maxLineSize)
 
 	for scanner.Scan() {
 		idleTimer.Reset(s.idleTimeout)

internal/streamer/streamer_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"runtime"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -198,6 +199,35 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
 	}
 }
 
+// TestStreamReadsLongLine has the server send one UPDATE line of about
+// 200 KiB, padded with raw hex the way RIS Live writes an extended BGP
+// message, and then end the response. The line is longer than a
+// bufio.Scanner reads by default. The streamer must hand the message to the
+// handler's queue and read on until the server ends the stream, instead of
+// dropping the connection at the long line.
+func TestStreamReadsLongLine(t *testing.T) {
+	const rawHexLength = 200 * bytesPerKB
+	longLine := `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` +
+		`"peer_asn":"64496","timestamp":1700000000,"raw":"` +
+		strings.Repeat("0", rawHexLength) + `"}}` + "\n"
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, longLine)
+	}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+	s.RegisterHandler(updateHandler{})
+
+	if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
+		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+	}
+
+	if queued := len(s.handlers[0].queue); queued != 1 {
+		t.Errorf("handler queue holds %d messages, want 1", queued)
+	}
+}
+
 // checkReconnect runs a streamer against a fake RIS Live server. The first
 // response is served by firstResponse; every later one carries one UPDATE
 // line and then stays open. The streamer must come back for a second