diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index b8b75a7..f5d4e18 100644 --- a/internal/streamer/streamer.go +++ b/internal/streamer/streamer.go @@ -7,6 +7,7 @@ import ( "compress/gzip" "context" "encoding/json" + "errors" "fmt" "io" "math" @@ -48,6 +49,7 @@ const ( metricsUpdateRate = time.Second minBackoffDelay = 5 * time.Second maxBackoffDelay = 320 * time.Second + streamIdleTimeout = 60 * time.Second // Reconnect when no line arrives for this long metricsLogInterval = 10 * time.Second bytesPerKB = 1024 bytesPerMB = 1024 * 1024 @@ -58,6 +60,9 @@ const ( backpressureSlope = 2.0 // Slope for linear drop probability increase ) +// errStreamEnded is returned by stream when RIS Live ends the response. +var errStreamEnded = errors.New("RIS Live ended the stream") + // MessageHandler defines the interface for processing RIS messages. // Implementations must specify which message types they want to receive, // how to process messages, and their desired queue capacity. @@ -107,6 +112,8 @@ type Streamer struct { logger *logger.Logger client *http.Client url string + idleTimeout time.Duration // Drop the connection when no line arrives for this long + minBackoff time.Duration // First wait before reconnecting; doubles up to maxBackoffDelay handlers []*handlerInfo rawHandler RawMessageHandler mu sync.RWMutex @@ -124,8 +131,10 @@ type Streamer struct { // The metrics tracker is used to record message counts, bytes received, and connection status. func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer { return &Streamer{ - logger: logger, - url: risLiveURL, + logger: logger, + url: risLiveURL, + idleTimeout: streamIdleTimeout, + minBackoff: minBackoffDelay, client: &http.Client{ Timeout: 0, // No timeout for streaming Transport: &http.Transport{ @@ -405,7 +414,7 @@ func (s *Streamer) updateMetrics(messageBytes int) { // streamWithReconnect handles streaming with automatic reconnection and exponential backoff func (s *Streamer) streamWithReconnect(ctx context.Context) { - backoffDelay := minBackoffDelay + backoffDelay := s.minBackoff consecutiveFailures := 0 for { @@ -422,8 +431,9 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) { err := s.stream(ctx) streamDuration := time.Since(startTime) - if err == nil { - // Clean exit (context cancelled) + // Only Stop ends the streamer. A stream that ends any other way, + // with or without an error, is a disconnect and is reconnected. + if ctx.Err() != nil { return } @@ -434,17 +444,12 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) { "stream_duration", streamDuration) s.metrics.SetConnected(false) - // Check if context is cancelled - if ctx.Err() != nil { - return - } - // If we streamed for more than 30 seconds, reset the backoff // This indicates we had a successful connection that received data if streamDuration > 30*time.Second { s.logger.Info("Resetting backoff delay due to successful connection", "stream_duration", streamDuration) - backoffDelay = minBackoffDelay + backoffDelay = s.minBackoff consecutiveFailures = 0 } else { // Increment consecutive failures @@ -477,7 +482,13 @@ func (s *Streamer) stream(ctx context.Context) error { connCtx, connCancel := context.WithCancel(ctx) defer connCancel() - req, err := http.NewRequestWithContext(ctx, "GET", s.url, nil) + // The idle timer drops the connection, by cancelling connCtx, when no + // line arrives for s.idleTimeout. Each line read starts it over. It + // also covers the wait for the server to answer. + idleTimer := time.AfterFunc(s.idleTimeout, connCancel) + defer idleTimer.Stop() + + req, err := http.NewRequestWithContext(connCtx, "GET", s.url, nil) if err != nil { return fmt.Errorf("failed to create request: %w", err) } @@ -559,6 +570,8 @@ func (s *Streamer) stream(ctx context.Context) error { scanner := bufio.NewScanner(reader) for scanner.Scan() { + idleTimer.Reset(s.idleTimeout) + select { case <-ctx.Done(): s.logger.Info("Stream stopped by context") @@ -707,11 +720,19 @@ func (s *Streamer) stream(ctx context.Context) error { s.mu.RUnlock() } + if ctx.Err() != nil { + return ctx.Err() + } + + if connCtx.Err() != nil { + return fmt.Errorf("no line received for %s", s.idleTimeout) + } + if err := scanner.Err(); err != nil { return fmt.Errorf("scanner error: %w", err) } - return nil + return errStreamEnded } // shouldDropForBackpressure determines if a message should be dropped based on queue utilization diff --git a/internal/streamer/streamer_test.go b/internal/streamer/streamer_test.go index 5d9582b..a56ec27 100644 --- a/internal/streamer/streamer_test.go +++ b/internal/streamer/streamer_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "runtime" + "sync/atomic" "testing" "time" @@ -57,8 +58,8 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) { // One warm-up connection so any persistent HTTP transport goroutine exists // before we take the baseline. - if err := s.stream(context.Background()); err != nil { - t.Fatalf("warm-up stream returned error: %v", err) + if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) { + t.Fatalf("warm-up stream returned %v, want %v", err, errStreamEnded) } s.client.CloseIdleConnections() @@ -66,8 +67,8 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) { const reconnects = 20 for range reconnects { - if err := s.stream(context.Background()); err != nil { - t.Fatalf("stream returned error: %v", err) + if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) { + t.Fatalf("stream returned %v, want %v", err, errStreamEnded) } } s.client.CloseIdleConnections() @@ -78,23 +79,35 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) { } } -// updateHandler wants UPDATE messages and does nothing with them. -type updateHandler struct{} +// updateLine is one UPDATE message as RIS Live sends it. +const updateLine = `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` + + `"peer_asn":"64496","timestamp":1700000000}}` + "\n" + +// updateHandler wants UPDATE messages. For each one it handles it sends on +// handled if there is room, and otherwise does nothing; a nil handled is +// never sent on. +type updateHandler struct { + handled chan struct{} +} func (updateHandler) WantsMessage(messageType string) bool { return messageType == "UPDATE" } -func (updateHandler) HandleMessage(*ristypes.RISMessage) {} func (updateHandler) QueueCapacity() int { return 10 } +func (h updateHandler) HandleMessage(*ristypes.RISMessage) { + select { + case h.handled <- struct{}{}: + default: + } +} + // TestStopBeforeMessageReachesQueues stops the streamer after the read loop // has checked for cancellation but before it hands the message to the handler // queues. That is the gap a stop from another goroutine can land in, and it // used to end in "send on closed channel". The raw handler runs in that gap on // the read loop itself, so calling Stop from it hits the gap every time. func TestStopBeforeMessageReachesQueues(t *testing.T) { - const line = `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` + - `"peer_asn":"64496","timestamp":1700000000}}` + "\n" srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - _, _ = io.WriteString(w, line) + _, _ = io.WriteString(w, updateLine) })) defer srv.Close() @@ -117,6 +130,77 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) { s.Stop() } +// TestReconnectWhenStreamEnds has the server end the first response +// normally, as when RIS Live restarts. The streamer must treat that as a +// disconnect and connect again. +func TestReconnectWhenStreamEnds(t *testing.T) { + endResponse := func(http.ResponseWriter, *http.Request) {} + checkReconnect(t, endResponse, streamIdleTimeout) +} + +// TestReconnectWhenStreamGoesIdle has the server hold the first response +// open without sending anything. The streamer must drop it once the idle +// timeout passes and connect again. +func TestReconnectWhenStreamGoesIdle(t *testing.T) { + const idleTimeout = 50 * time.Millisecond + holdResponse := func(w http.ResponseWriter, r *http.Request) { + w.(http.Flusher).Flush() + <-r.Context().Done() + } + checkReconnect(t, holdResponse, idleTimeout) +} + +// 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 +// response, must not be marked connected when it does, and must handle the +// UPDATE in it. +func checkReconnect(t *testing.T, firstResponse http.HandlerFunc, idleTimeout time.Duration) { + t.Helper() + + const maxWait = 5 * time.Second + + tracker := metrics.New() + connectedAtReconnect := make(chan bool, 1) + var requests atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if requests.Add(1) == 1 { + firstResponse(w, r) + + return + } + select { + case connectedAtReconnect <- tracker.IsConnected(): + default: + } + _, _ = io.WriteString(w, updateLine) + w.(http.Flusher).Flush() + <-r.Context().Done() + })) + defer srv.Close() + + s := New(logger.New(), tracker) + s.url = srv.URL + s.idleTimeout = idleTimeout + s.minBackoff = time.Millisecond + handler := updateHandler{handled: make(chan struct{}, 1)} + s.RegisterHandler(handler) + if err := s.Start(); err != nil { + t.Fatalf("Start returned error: %v", err) + } + defer s.Stop() + + select { + case <-handler.handled: + case <-time.After(maxWait): + t.Fatalf("no message from a second response handled within %s", maxWait) + } + + if <-connectedAtReconnect { + t.Error("streamer was still marked connected when it came back for a second response") + } +} + // settledGoroutineCount lets transient goroutines finish, then reports the // current count. func settledGoroutineCount() int {