Reconnect when RIS Live ends or stalls the stream (closes #53)

Only Stop now ends the reconnect loop. A stream that ends any other way,
with or without an error, marks the feed disconnected and reconnects with
the existing backoff. Before, a response RIS Live ended normally stopped
the loop for good while the health check kept reporting ok.

The streamer also drops the connection when no line arrives for 60
seconds; each line starts that limit over, and it also bounds the wait
for RIS Live to answer. The idle limit and the first backoff delay are
fields so the tests, against a local server, run in well under a second.
The README's Reconnection note describes this behaviour.

Model: opus-5-5

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,13 @@ 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
+		// A stream that lasted more than 30 seconds resets the backoff. That
+		// includes a stream dropped after streamIdleTimeout with no line, so
+		// a silent feed is retried after the first backoff delay every time.
 		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 +483,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 +571,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 +721,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