Re-vendor the shared files from sneak/prompts at dd4027b (closes #52)

Takes .golangci.yml, .dockerignore, .gitignore, .editorconfig, the
workflow and REPO_POLICIES.md from sneak/prompts dd4027b, keeping this
repository's own entries after the shared content. Lint (golangci-lint
v2.14.0) and test become Dockerfile phases that script/lint and
script/test build alone and uncached; the image build depends on both,
and script/cibuild bootstraps, runs script/check, then builds the image.
The agent guidance file moves to AGENTS.md. The Go code is fixed for
what the new configuration finds; reads that iterate rows now also
report an iteration error.

Model: opus-5-5

internal/streamer/streamer.go

@@ -1,5 +1,6 @@
-// Package streamer implements an HTTP client that connects to the RIPE RIS Live streaming API,
-// parses BGP UPDATE messages from the JSON stream, and dispatches them to registered handlers.
+// Package streamer implements an HTTP client that connects to the RIPE RIS Live
+// streaming API, parses BGP UPDATE messages from the JSON stream, and
+// dispatches them to registered handlers.
 package streamer
 
 import (
@@ -25,31 +26,32 @@ import (
 // countingReader wraps an io.Reader and counts bytes read
 type countingReader struct {
 	reader io.Reader
-	count  int64
+	count  atomic.Int64
 }
 
 // Read implements io.Reader and counts bytes
 func (c *countingReader) Read(p []byte) (int, error) {
 	n, err := c.reader.Read(p)
-	atomic.AddInt64(&c.count, int64(n))
+	c.count.Add(int64(n))
 
 	return n, err
 }
 
 // Count returns the total bytes read
 func (c *countingReader) Count() int64 {
-	return atomic.LoadInt64(&c.count)
+	return c.count.Load()
 }
 
 // Configuration constants for the RIS Live streamer.
 const (
 	risLiveURL = "https://ris-live.ripe.net/v1/stream/?format=json&" +
 		"client=https%3A%2F%2Fgit.eeqj.de%2Fsneak%2Froutewatch"
-	metricsWindowSize     = 60 // seconds for rolling average
-	metricsUpdateRate     = time.Second
-	minBackoffDelay       = 5 * time.Second
-	maxBackoffDelay       = 320 * time.Second
-	streamIdleTimeout     = 60 * time.Second // Reconnect when no line arrives for this long
+	metricsWindowSize = 60 // seconds for rolling average
+	metricsUpdateRate = time.Second
+	minBackoffDelay   = 5 * time.Second
+	maxBackoffDelay   = 320 * time.Second
+	// Reconnect when no line arrives for this long
+	streamIdleTimeout     = 60 * time.Second
 	metricsLogInterval    = 10 * time.Second
 	bytesPerKB            = 1024
 	bytesPerMB            = 1024 * 1024
@@ -60,14 +62,22 @@ 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")
+var (
+	errAlreadyRunning   = errors.New("streamer already running")
+	errUnexpectedStatus = errors.New("unexpected status code")
+	// errStreamEnded is returned by stream when RIS Live ends the response.
+	errStreamEnded = errors.New("RIS Live ended the stream")
+	// errStreamIdle is returned by stream when no line arrived for the idle
+	// timeout.
+	errStreamIdle = errors.New("no line received")
+)
 
 // 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.
 type MessageHandler interface {
-	// WantsMessage returns true if this handler wants to process messages of the given type.
+	// WantsMessage returns true if this handler wants to process messages of
+	// the given type.
 	WantsMessage(messageType string) bool
 
 	// HandleMessage processes a RIS message. This method is called from a dedicated
@@ -109,18 +119,20 @@ type handlerInfo struct {
 // backoff, dispatches messages to registered handlers via per-handler queues, and
 // implements backpressure to prevent queue overflow during high traffic periods.
 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
+	logger *logger.Logger
+	client *http.Client
+	url    string
+	// Drop the connection when no line arrives for this long
+	idleTimeout time.Duration
+	// First wait before reconnecting; doubles up to maxBackoffDelay
+	minBackoff   time.Duration
 	handlers     []*handlerInfo
 	rawHandler   RawMessageHandler
 	mu           sync.RWMutex
 	cancel       context.CancelFunc
 	running      bool
 	metrics      *metrics.Tracker
-	totalDropped uint64          // Total dropped messages across all handlers
+	totalDropped atomic.Uint64   // Total dropped messages across all handlers
 	random       *rand.Rand      // Random number generator for backpressure drops
 	bgpPeers     map[string]bool // Track active BGP peers by peer IP
 	bgpPeersMu   sync.RWMutex    // Protects bgpPeers map
@@ -128,7 +140,8 @@ type Streamer struct {
 
 // New creates a new Streamer instance configured to connect to the RIS Live API.
 // The logger is used for structured logging of connection events and errors.
-// The metrics tracker is used to record message counts, bytes received, and connection status.
+// 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,
@@ -162,7 +175,8 @@ func (s *Streamer) RegisterHandler(handler MessageHandler) {
 		handler: handler,
 		queue:   make(chan *ristypes.RISMessage, handler.QueueCapacity()),
 		metrics: handlerMetrics{
-			minTime: time.Duration(math.MaxInt64), // Initialize to max so first value sets the floor
+			// Initialize to max so first value sets the floor
+			minTime: time.Duration(math.MaxInt64),
 		},
 	}
 
@@ -180,22 +194,24 @@ func (s *Streamer) RegisterHandler(handler MessageHandler) {
 func (s *Streamer) RegisterRawHandler(handler RawMessageHandler) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
+
 	s.rawHandler = handler
 }
 
 // Start begins streaming BGP updates from the RIS Live API in a background goroutine.
 // It starts worker goroutines for each registered handler and manages automatic
 // reconnection with exponential backoff on connection failures.
-// Returns an error if the streamer is already running.
-func (s *Streamer) Start() error {
+// Returns an error if the streamer is already running. It streams until Stop:
+// the end of ctx does not stop it.
+func (s *Streamer) Start(ctx context.Context) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	if s.running {
-		return fmt.Errorf("streamer already running")
+		return errAlreadyRunning
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(context.WithoutCancel(ctx))
 	s.cancel = cancel
 	s.running = true
 
@@ -225,44 +241,21 @@ func (s *Streamer) Stop() {
 
 		return
 	}
+
 	s.cancel()
 	s.cancel = nil
 	// Close all handler queues to signal workers to stop
 	for _, info := range s.handlers {
 		close(info.queue)
 	}
+
 	s.running = false
 	s.mu.Unlock()
 	s.metrics.SetConnected(false)
 }
 
-// runHandlerWorker processes messages for a specific handler
-func (s *Streamer) runHandlerWorker(info *handlerInfo) {
-	for msg := range info.queue {
-		start := time.Now()
-		info.handler.HandleMessage(msg)
-		elapsed := time.Since(start)
-
-		// Update metrics
-		info.metrics.mu.Lock()
-		info.metrics.processedCount++
-		info.metrics.totalTime += elapsed
-
-		// Update min time
-		if elapsed < info.metrics.minTime {
-			info.metrics.minTime = elapsed
-		}
-
-		// Update max time
-		if elapsed > info.metrics.maxTime {
-			info.metrics.maxTime = elapsed
-		}
-		info.metrics.mu.Unlock()
-	}
-}
-
-// IsRunning reports whether the streamer is currently connected and processing messages.
-// This is safe to call concurrently from multiple goroutines.
+// IsRunning reports whether the streamer is currently connected and processing
+// messages. This is safe to call concurrently from multiple goroutines.
 func (s *Streamer) IsRunning() bool {
 	s.mu.RLock()
 	defer s.mu.RUnlock()
@@ -323,12 +316,15 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 		// Calculate average time
 		if info.metrics.processedCount > 0 {
 			processedCount := info.metrics.processedCount
+
 			const maxInt64 = 1<<63 - 1
+
 			if processedCount > maxInt64 {
 				processedCount = maxInt64
 			}
-			//nolint:gosec // processedCount is explicitly bounded above
-			hs.AvgProcessTime = info.metrics.totalTime / time.Duration(processedCount)
+
+			hs.AvgProcessTime = info.metrics.totalTime /
+				time.Duration(processedCount)
 		}
 
 		info.metrics.mu.Unlock()
@@ -339,11 +335,38 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 	return stats
 }
 
-// GetDroppedMessages returns the total number of messages dropped across all handlers
-// due to queue overflow or backpressure. This counter is monotonically increasing
-// and is safe to call concurrently.
+// GetDroppedMessages returns the total number of messages dropped across all
+// handlers due to queue overflow or backpressure. This counter is monotonically
+// increasing and is safe to call concurrently.
 func (s *Streamer) GetDroppedMessages() uint64 {
-	return atomic.LoadUint64(&s.totalDropped)
+	return s.totalDropped.Load()
+}
+
+// runHandlerWorker processes messages for a specific handler
+func (s *Streamer) runHandlerWorker(info *handlerInfo) {
+	for msg := range info.queue {
+		start := time.Now()
+
+		info.handler.HandleMessage(msg)
+
+		elapsed := time.Since(start)
+
+		// Update metrics
+		info.metrics.mu.Lock()
+		info.metrics.processedCount++
+		info.metrics.totalTime += elapsed
+
+		// Update min time
+		if elapsed < info.metrics.minTime {
+			info.metrics.minTime = elapsed
+		}
+
+		// Update max time
+		if elapsed > info.metrics.maxTime {
+			info.metrics.maxTime = elapsed
+		}
+		info.metrics.mu.Unlock()
+	}
 }
 
 // logMetrics logs the current streaming statistics
@@ -352,7 +375,8 @@ func (s *Streamer) logMetrics() {
 	uptime := time.Since(metrics.ConnectedSince)
 
 	const bitsPerMegabit = 1000000
-	totalDropped := atomic.LoadUint64(&s.totalDropped)
+
+	totalDropped := s.totalDropped.Load()
 
 	s.logger.Info(
 		"Stream statistics",
@@ -378,16 +402,19 @@ func (s *Streamer) logMetrics() {
 
 	// Log per-handler statistics
 	s.mu.RLock()
+
 	for i, info := range s.handlers {
 		info.metrics.mu.Lock()
 		if info.metrics.processedCount > 0 {
 			// Safe conversion: processedCount is bounded by maxInt64
 			processedCount := info.metrics.processedCount
+
 			const maxInt64 = 1<<63 - 1
+
 			if processedCount > maxInt64 {
 				processedCount = maxInt64
 			}
-			//nolint:gosec // processedCount is explicitly bounded above
+
 			avgTime := info.metrics.totalTime / time.Duration(processedCount)
 			s.logger.Info(
 				"Handler statistics",
@@ -404,6 +431,7 @@ func (s *Streamer) logMetrics() {
 		}
 		info.metrics.mu.Unlock()
 	}
+
 	s.mu.RUnlock()
 }
 
@@ -412,7 +440,8 @@ func (s *Streamer) updateMetrics(messageBytes int) {
 	s.metrics.RecordMessage(int64(messageBytes))
 }
 
-// streamWithReconnect handles streaming with automatic reconnection and exponential backoff
+// streamWithReconnect handles streaming with automatic reconnection and
+// exponential backoff
 func (s *Streamer) streamWithReconnect(ctx context.Context) {
 	backoffDelay := s.minBackoff
 	consecutiveFailures := 0
@@ -450,6 +479,7 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) {
 		if streamDuration > 30*time.Second {
 			s.logger.Info("Resetting backoff delay due to successful connection",
 				"stream_duration", streamDuration)
+
 			backoffDelay = s.minBackoff
 			consecutiveFailures = 0
 		} else {
@@ -489,7 +519,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 	idleTimer := time.AfterFunc(s.idleTimeout, connCancel)
 	defer idleTimer.Stop()
 
-	req, err := http.NewRequestWithContext(connCtx, "GET", s.url, nil)
+	req, err := http.NewRequestWithContext(connCtx, http.MethodGet, s.url, nil)
 	if err != nil {
 		return fmt.Errorf("failed to create request: %w", err)
 	}
@@ -502,13 +532,14 @@ func (s *Streamer) stream(ctx context.Context) error {
 		return fmt.Errorf("failed to connect to RIS Live: %w", err)
 	}
 	defer func() {
-		if err := resp.Body.Close(); err != nil {
+		err := resp.Body.Close()
+		if err != nil {
 			s.logger.Error("Failed to close response body", "error", err)
 		}
 	}()
 
 	if resp.StatusCode != http.StatusOK {
-		return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return fmt.Errorf("%w: %d", errUnexpectedStatus, resp.StatusCode)
 	}
 
 	// Wrap body with counting reader to track actual wire bytes
@@ -522,7 +553,9 @@ func (s *Streamer) stream(ctx context.Context) error {
 			return fmt.Errorf("failed to create gzip reader: %w", err)
 		}
 		defer func() { _ = gzReader.Close() }()
+
 		reader = gzReader
+
 		s.logger.Info("Connected to RIS Live stream", "compression", "gzip")
 	} else {
 		s.logger.Info("Connected to RIS Live stream", "compression", "none")
@@ -530,9 +563,6 @@ func (s *Streamer) stream(ctx context.Context) error {
 
 	s.metrics.SetConnected(true)
 
-	// Track wire bytes for metrics updates
-	var lastWireBytes int64
-
 	// Start metrics logging goroutine
 	metricsTicker := time.NewTicker(metricsLogInterval)
 	defer metricsTicker.Stop()
@@ -549,25 +579,44 @@ func (s *Streamer) stream(ctx context.Context) error {
 	}()
 
 	// Wire byte update ticker - update metrics with actual wire bytes periodically
+	go s.recordWireBytes(connCtx, wireCounter)
+
+	return s.readStream(ctx, connCtx, reader, idleTimer)
+}
+
+// recordWireBytes adds the bytes wireCounter has counted to the metrics once a
+// second until ctx is cancelled.
+func (s *Streamer) recordWireBytes(
+	ctx context.Context, wireCounter *countingReader,
+) {
 	wireUpdateTicker := time.NewTicker(time.Second)
 	defer wireUpdateTicker.Stop()
 
-	go func() {
-		for {
-			select {
-			case <-wireUpdateTicker.C:
-				currentBytes := wireCounter.Count()
-				delta := currentBytes - lastWireBytes
-				if delta > 0 {
-					s.metrics.RecordWireBytes(delta)
-					lastWireBytes = currentBytes
-				}
-			case <-connCtx.Done():
-				return
-			}
-		}
-	}()
+	var lastWireBytes int64
 
+	for {
+		select {
+		case <-wireUpdateTicker.C:
+			currentBytes := wireCounter.Count()
+
+			delta := currentBytes - lastWireBytes
+			if delta > 0 {
+				s.metrics.RecordWireBytes(delta)
+
+				lastWireBytes = currentBytes
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// readStream hands each line of the stream to processLine until the stream
+// ends, a line fails, or ctx is cancelled. Each line restarts idleTimer, which
+// cancels connCtx when it fires.
+func (s *Streamer) readStream(
+	ctx, connCtx context.Context, reader io.Reader, idleTimer *time.Timer,
+) error {
 	scanner := bufio.NewScanner(reader)
 
 	for scanner.Scan() {
@@ -586,139 +635,10 @@ func (s *Streamer) stream(ctx context.Context) error {
 			continue
 		}
 
-		// Update metrics with decompressed message size
-		s.updateMetrics(len(line))
-
-		// Call raw handler if registered
-		s.mu.RLock()
-		rawHandler := s.rawHandler
-		s.mu.RUnlock()
-
-		if rawHandler != nil {
-			// Call raw handler synchronously to preserve order
-			rawHandler(string(line))
+		err := s.processLine(ctx, line)
+		if err != nil {
+			return err
 		}
-
-		// Parse the message first
-		var wrapper ristypes.RISLiveMessage
-		if err := json.Unmarshal(line, &wrapper); err != nil {
-			// Log the error and return to trigger reconnection
-			s.logger.Error("Failed to parse JSON",
-				"error", err,
-				"line", string(line),
-				"line_length", len(line))
-
-			return fmt.Errorf("JSON parse error: %w", err)
-		}
-
-		// Check if it's a ris_message wrapper
-		if wrapper.Type != "ris_message" {
-			s.logger.Error("Unexpected wrapper type",
-				"type", wrapper.Type,
-				"line", string(line),
-			)
-
-			continue
-		}
-
-		// Get the actual message
-		msg := wrapper.Data
-
-		// Parse the timestamp
-		msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
-
-		// Process based on message type
-		switch msg.Type {
-		case "UPDATE":
-			// Process BGP UPDATE messages
-			// Will be dispatched to handlers
-		case "RIS_PEER_STATE":
-			// RIS peer state messages - silently ignore
-			continue
-		case "KEEPALIVE":
-			// BGP keepalive messages - silently process
-			continue
-		case "OPEN":
-			// BGP open messages - track peer as active
-			s.bgpPeersMu.Lock()
-			s.bgpPeers[msg.Peer] = true
-			peerCount := len(s.bgpPeers)
-			s.bgpPeersMu.Unlock()
-			s.metrics.SetBGPPeerCount(peerCount)
-
-			s.logger.Info("BGP session opened",
-				"peer", msg.Peer,
-				"peer_asn", msg.PeerASN,
-				"total_peers", peerCount,
-			)
-
-			continue
-		case "NOTIFICATION":
-			// BGP notification messages (session closed)
-			s.bgpPeersMu.Lock()
-			delete(s.bgpPeers, msg.Peer)
-			peerCount := len(s.bgpPeers)
-			s.bgpPeersMu.Unlock()
-			s.metrics.SetBGPPeerCount(peerCount)
-
-			s.logger.Warn("BGP notification",
-				"peer", msg.Peer,
-				"peer_asn", msg.PeerASN,
-				"total_peers", peerCount,
-			)
-
-			continue
-		case "STATE":
-			// Peer state changes - silently ignore
-			continue
-		default:
-			s.logger.Warn("Unknown message type, skipping",
-				"type", msg.Type,
-			)
-
-			continue
-		}
-
-		// Dispatch to interested handlers. Stop cancels ctx and closes the
-		// queues under the write lock, so if ctx is cancelled here, under the
-		// read lock, the queues are closed and must not be sent to.
-		s.mu.RLock()
-		if ctx.Err() != nil {
-			s.mu.RUnlock()
-
-			return ctx.Err()
-		}
-		for _, info := range s.handlers {
-			if !info.handler.WantsMessage(msg.Type) {
-				continue
-			}
-
-			// Check if we should drop due to backpressure
-			if s.shouldDropForBackpressure(info) {
-				atomic.AddUint64(&info.metrics.droppedCount, 1)
-				atomic.AddUint64(&s.totalDropped, 1)
-
-				continue
-			}
-
-			// Try to queue the message
-			select {
-			case info.queue <- &msg:
-				// Message queued successfully
-				// Update high water mark if needed
-				queueLen := len(info.queue)
-				info.metrics.mu.Lock()
-				if queueLen > info.metrics.queueHighWaterMark {
-					info.metrics.queueHighWaterMark = queueLen
-				}
-				info.metrics.mu.Unlock()
-			default:
-				// Queue is full, drop the message
-				atomic.AddUint64(&info.metrics.droppedCount, 1)
-				atomic.AddUint64(&s.totalDropped, 1)
-			}
-		}
-		s.mu.RUnlock()
 	}
 
 	if ctx.Err() != nil {
@@ -726,17 +646,173 @@ func (s *Streamer) stream(ctx context.Context) error {
 	}
 
 	if connCtx.Err() != nil {
-		return fmt.Errorf("no line received for %s", s.idleTimeout)
+		return fmt.Errorf("%w for %s", errStreamIdle, s.idleTimeout)
 	}
 
-	if err := scanner.Err(); err != nil {
+	err := scanner.Err()
+	if err != nil {
 		return fmt.Errorf("scanner error: %w", err)
 	}
 
 	return errStreamEnded
 }
 
-// shouldDropForBackpressure determines if a message should be dropped based on queue utilization
+// processLine passes one line of the stream to the raw handler, parses it,
+// and dispatches an UPDATE message to the handlers. An error ends the
+// connection.
+func (s *Streamer) processLine(ctx context.Context, line []byte) error {
+	// Update metrics with decompressed message size
+	s.updateMetrics(len(line))
+
+	// Call raw handler if registered
+	s.mu.RLock()
+	rawHandler := s.rawHandler
+	s.mu.RUnlock()
+
+	if rawHandler != nil {
+		// Call raw handler synchronously to preserve order
+		rawHandler(string(line))
+	}
+
+	// Parse the message first
+	var wrapper ristypes.RISLiveMessage
+
+	err := json.Unmarshal(line, &wrapper)
+	if err != nil {
+		// Log the error and return to trigger reconnection
+		s.logger.Error("Failed to parse JSON",
+			"error", err,
+			"line", string(line),
+			"line_length", len(line))
+
+		return fmt.Errorf("JSON parse error: %w", err)
+	}
+
+	// Check if it's a ris_message wrapper
+	if wrapper.Type != "ris_message" {
+		s.logger.Error("Unexpected wrapper type",
+			"type", wrapper.Type,
+			"line", string(line),
+		)
+
+		return nil
+	}
+
+	// Get the actual message
+	msg := wrapper.Data
+
+	// Parse the timestamp
+	msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
+
+	// Process based on message type
+	switch msg.Type {
+	case "UPDATE":
+		// Process BGP UPDATE messages
+		// Will be dispatched to handlers
+	case "RIS_PEER_STATE":
+		// RIS peer state messages - silently ignore
+		return nil
+	case "KEEPALIVE":
+		// BGP keepalive messages - silently process
+		return nil
+	case "OPEN":
+		// BGP open messages - track peer as active
+		s.bgpPeersMu.Lock()
+		s.bgpPeers[msg.Peer] = true
+		peerCount := len(s.bgpPeers)
+		s.bgpPeersMu.Unlock()
+		s.metrics.SetBGPPeerCount(peerCount)
+
+		s.logger.Info("BGP session opened",
+			"peer", msg.Peer,
+			"peer_asn", msg.PeerASN,
+			"total_peers", peerCount,
+		)
+
+		return nil
+	case "NOTIFICATION":
+		// BGP notification messages (session closed)
+		s.bgpPeersMu.Lock()
+		delete(s.bgpPeers, msg.Peer)
+		peerCount := len(s.bgpPeers)
+		s.bgpPeersMu.Unlock()
+		s.metrics.SetBGPPeerCount(peerCount)
+
+		s.logger.Warn("BGP notification",
+			"peer", msg.Peer,
+			"peer_asn", msg.PeerASN,
+			"total_peers", peerCount,
+		)
+
+		return nil
+	case "STATE":
+		// Peer state changes - silently ignore
+		return nil
+	default:
+		s.logger.Warn("Unknown message type, skipping",
+			"type", msg.Type,
+		)
+
+		return nil
+	}
+
+	return s.dispatchMessage(ctx, &msg)
+}
+
+// dispatchMessage queues msg for each handler that wants it, dropping it for
+// a handler whose queue is full or under backpressure.
+func (s *Streamer) dispatchMessage(
+	ctx context.Context, msg *ristypes.RISMessage,
+) error {
+	// Stop cancels ctx and closes the queues under the write lock, so if ctx
+	// is cancelled here, under the read lock, the queues are closed and must
+	// not be sent to.
+	s.mu.RLock()
+
+	if ctx.Err() != nil {
+		s.mu.RUnlock()
+
+		return ctx.Err()
+	}
+
+	for _, info := range s.handlers {
+		if !info.handler.WantsMessage(msg.Type) {
+			continue
+		}
+
+		// Check if we should drop due to backpressure
+		if s.shouldDropForBackpressure(info) {
+			atomic.AddUint64(&info.metrics.droppedCount, 1)
+			s.totalDropped.Add(1)
+
+			continue
+		}
+
+		// Try to queue the message
+		select {
+		case info.queue <- msg:
+			// Message queued successfully
+			// Update high water mark if needed
+			queueLen := len(info.queue)
+			info.metrics.mu.Lock()
+			if queueLen > info.metrics.queueHighWaterMark {
+				info.metrics.queueHighWaterMark = queueLen
+			}
+			info.metrics.mu.Unlock()
+		default:
+			// Queue is full, drop the message
+			atomic.AddUint64(&info.metrics.droppedCount, 1)
+			s.totalDropped.Add(1)
+		}
+	}
+
+	s.mu.RUnlock()
+
+	return nil
+}
+
+// shouldDropForBackpressure determines if a message should be dropped based on
+// queue utilization
 func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
 	// Calculate queue utilization
 	queueLen := len(info.queue)

internal/streamer/streamer_internal_test.go

@@ -17,6 +17,8 @@ import (
 )
 
 func TestNewStreamer(t *testing.T) {
+	t.Parallel()
+
 	logger := logger.New()
 	metricsTracker := metrics.New()
 	s := New(logger, metricsTracker)
@@ -55,10 +57,13 @@ func TestNewStreamer(t *testing.T) {
 // goroutine count returns to its starting value. Each connection starts two
 // ticker goroutines; before the fix they lived until the streamer's lifetime
 // context was cancelled, so every reconnect leaked two.
+//
+//nolint:paralleltest // counts goroutines, so no other test may run alongside
 func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
 	// The handler returns immediately, so the response body is empty and each
 	// stream call ends at once, standing in for a dropped connection.
-	srv := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {}))
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(_ http.ResponseWriter, _ *http.Request) {}))
 	defer srv.Close()
 
 	s := New(logger.New(), metrics.New())
@@ -66,19 +71,23 @@ 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()); !errors.Is(err, errStreamEnded) {
+	err := s.stream(context.Background())
+	if !errors.Is(err, errStreamEnded) {
 		t.Fatalf("warm-up stream returned %v, want %v", err, errStreamEnded)
 	}
+
 	s.client.CloseIdleConnections()
 
 	baseline := settledGoroutineCount()
 
 	const reconnects = 20
 	for range reconnects {
-		if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
+		err := s.stream(context.Background())
+		if !errors.Is(err, errStreamEnded) {
 			t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
 		}
 	}
+
 	s.client.CloseIdleConnections()
 
 	if !waitForGoroutines(baseline) {
@@ -88,8 +97,8 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
 }
 
 // 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"
+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
@@ -98,8 +107,11 @@ type updateHandler struct {
 	handled chan struct{}
 }
 
-func (updateHandler) WantsMessage(messageType string) bool { return messageType == "UPDATE" }
-func (updateHandler) QueueCapacity() int                   { return 10 }
+func (updateHandler) WantsMessage(messageType string) bool {
+	return messageType == "UPDATE"
+}
+
+func (updateHandler) QueueCapacity() int { return 10 }
 
 func (h updateHandler) HandleMessage(*ristypes.RISMessage) {
 	select {
@@ -108,15 +120,27 @@ func (h updateHandler) HandleMessage(*ristypes.RISMessage) {
 	}
 }
 
+// flush sends what w has buffered to the client. The writers httptest servers
+// hand to handlers can always do that.
+func flush(w http.ResponseWriter) {
+	flusher, ok := w.(http.Flusher)
+	if ok {
+		flusher.Flush()
+	}
+}
+
 // 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) {
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		_, _ = io.WriteString(w, updateLine)
-	}))
+	t.Parallel()
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = io.WriteString(w, updateLine)
+		}))
 	defer srv.Close()
 
 	s := New(logger.New(), metrics.New())
@@ -130,7 +154,8 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	s.cancel = cancel
 
-	if err := s.stream(ctx); !errors.Is(err, context.Canceled) {
+	err := s.stream(ctx)
+	if !errors.Is(err, context.Canceled) {
 		t.Fatalf("stream returned %v, want %v", err, context.Canceled)
 	}
 
@@ -142,6 +167,8 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
 // normally, as when RIS Live restarts. The streamer must treat that as a
 // disconnect and connect again.
 func TestReconnectWhenStreamEnds(t *testing.T) {
+	t.Parallel()
+
 	endResponse := func(http.ResponseWriter, *http.Request) {}
 	checkReconnect(t, endResponse, streamIdleTimeout)
 }
@@ -150,9 +177,12 @@ func TestReconnectWhenStreamEnds(t *testing.T) {
 // open without sending anything. The streamer must drop it once the idle
 // timeout passes and connect again.
 func TestReconnectWhenStreamGoesIdle(t *testing.T) {
+	t.Parallel()
+
 	const idleTimeout = 50 * time.Millisecond
+
 	holdResponse := func(w http.ResponseWriter, r *http.Request) {
-		w.(http.Flusher).Flush()
+		flush(w)
 		<-r.Context().Done()
 	}
 	checkReconnect(t, holdResponse, idleTimeout)
@@ -162,7 +192,10 @@ func TestReconnectWhenStreamGoesIdle(t *testing.T) {
 // request and never answer it, not even with headers. The streamer must give
 // up on it once the idle timeout passes and connect again.
 func TestReconnectWhenServerNeverAnswers(t *testing.T) {
+	t.Parallel()
+
 	const idleTimeout = 50 * time.Millisecond
+
 	neverAnswer := func(_ http.ResponseWriter, r *http.Request) {
 		<-r.Context().Done()
 	}
@@ -174,26 +207,32 @@ func TestReconnectWhenServerNeverAnswers(t *testing.T) {
 // line must start the idle timeout over, so the stream lasts until the server
 // ends it instead of being dropped as idle.
 func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
+	t.Parallel()
+
 	const (
 		idleTimeout  = 100 * time.Millisecond
 		lineInterval = idleTimeout / 10
 		sendFor      = 5 * idleTimeout
 	)
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
-		deadline := time.Now().Add(sendFor)
-		for time.Now().Before(deadline) {
-			_, _ = io.WriteString(w, updateLine)
-			w.(http.Flusher).Flush()
-			time.Sleep(lineInterval)
-		}
-	}))
+
+	srv := httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, _ *http.Request) {
+			deadline := time.Now().Add(sendFor)
+			for time.Now().Before(deadline) {
+				_, _ = io.WriteString(w, updateLine)
+
+				flush(w)
+				time.Sleep(lineInterval)
+			}
+		}))
 	defer srv.Close()
 
 	s := New(logger.New(), metrics.New())
 	s.url = srv.URL
 	s.idleTimeout = idleTimeout
 
-	if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
+	err := s.stream(context.Background())
+	if !errors.Is(err, errStreamEnded) {
 		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
 	}
 }
@@ -203,28 +242,36 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
 // 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) {
+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()
-	}))
+	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)
+
+			flush(w)
+			<-r.Context().Done()
+		}))
 	defer srv.Close()
 
 	s := New(logger.New(), tracker)
@@ -233,7 +280,9 @@ func checkReconnect(t *testing.T, firstResponse http.HandlerFunc, idleTimeout ti
 	s.minBackoff = time.Millisecond
 	handler := updateHandler{handled: make(chan struct{}, 1)}
 	s.RegisterHandler(handler)
-	if err := s.Start(); err != nil {
+
+	err := s.Start(t.Context())
+	if err != nil {
 		t.Fatalf("Start returned error: %v", err)
 	}
 	defer s.Stop()
@@ -245,7 +294,8 @@ func checkReconnect(t *testing.T, firstResponse http.HandlerFunc, idleTimeout ti
 	}
 
 	if <-connectedAtReconnect {
-		t.Error("streamer was still marked connected when it came back for a second response")
+		t.Error("streamer was still marked connected when it came back " +
+			"for a second response")
 	}
 }
 
@@ -253,12 +303,15 @@ func checkReconnect(t *testing.T, firstResponse http.HandlerFunc, idleTimeout ti
 // current count.
 func settledGoroutineCount() int {
 	prev := runtime.NumGoroutine()
+
 	for range 20 {
 		time.Sleep(10 * time.Millisecond)
+
 		cur := runtime.NumGoroutine()
 		if cur == prev {
 			return cur
 		}
+
 		prev = cur
 	}
 
@@ -271,6 +324,7 @@ func waitForGoroutines(target int) bool {
 		if runtime.NumGoroutine() <= target {
 			return true
 		}
+
 		time.Sleep(10 * time.Millisecond)
 	}