From f6373656ea30035c29fea31a351cd17496e34eaf Mon Sep 17 00:00:00 2001 From: sneak Date: Mon, 19 Oct 2026 09:28:12 +0000 Subject: [PATCH] Read feed lines up to 1 MiB (closes #76) The streamer read the feed with a bufio.Scanner at its default limit of 64 KiB per line, so a longer line ended the connection with "token too long" and the updates sent while it reconnected were lost. The scanner now accepts lines up to 1 MiB, well above an extended BGP message (RFC 8654) written as hex with its decoded prefixes. Model: opus-5-5 --- internal/streamer/streamer.go | 4 +++- internal/streamer/streamer_test.go | 30 ++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index 04e4429..3501019 100644 --- a/internal/streamer/streamer.go +++ b/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) diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go index d2da82c..a691efe 100644 --- a/internal/streamer/streamer_test.go +++ b/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 -- 2.54.0